WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-11 12:17:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

[haik]: refactor: canonicalize settings imports and public API naming - move all load_settings imports from backend.apps.settings.settings to backend.apps.settings.store (agent_manager, agents, browser_agent, mcp_preflight, anthropic_proxy, auth/router, dashboards, outputs, service, subscription/router, web) so consumers point at the source module instead of the re-export; rename _atomic_write_settings to atomic_write_settings and _has_own_model to has_own_model since both are called across module boundaries; rename _summarize_block to summarize_block and UPLOAD_DIR to P_UPLOAD_DIR in settings.py; clean up unused var names (_base to _, _ext to _, _httpx to httpx); update test_auth_router.py imports and _save_settings references to match

Browse Source
This commit is contained in:
haikdc
2026-06-13 22:22:31 -07:00
parent 87f09e9695
commit 67fce8ee89
16 changed files with 72 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/agent_manager.py
+1 -1
View File
@@ -13,7 +13,7 @@ from backend.apps.agents.core.models import (
AgentConfig, AgentSession, Message, MessageBranch, ApprovalRequest, ToolGroupMeta,
)
from backend.apps.agents.core.ws_manager import ws_manager
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
from backend.apps.tools_lib.tools_lib import (
_load_all as load_all_tools,
derive_mcp_config,
backend/apps/agents/agents.py
+5 -5
View File
@@ -402,7 +402,7 @@ async def subscriptions_poll(body: dict):
)
if result.get("success"):
from backend.apps.service.client import sync as _sync
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
_sync(load_settings().model_dump())
return result
except Exception as e:
@@ -426,7 +426,7 @@ async def subscriptions_exchange(body: dict):
result = await exchange_oauth(provider, code, redirect_uri, code_verifier, state)
if result.get("success"):
from backend.apps.service.client import sync as _sync
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
_sync(load_settings().model_dump())
return result
except Exception as e:
@@ -458,7 +458,7 @@ async def probe_model(body: dict):
_find_builtin_model,
_NINEROUTER_MODEL_PREFIXES,
)
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
from backend.apps.nine_router.process import is_running
settings = load_settings()
api_type = get_api_type(short_name)
@@ -525,7 +525,7 @@ async def list_models():
"""Picker model list, grouped by provider, intersected with available creds."""
from backend.apps.agents.providers.registry import BUILTIN_MODELS
from backend.apps.nine_router.process import is_running, get_providers
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
settings = load_settings()
nine_router_up = is_running()
@@ -818,7 +818,7 @@ async def subscriptions_disconnect(body: dict):
removed = await _delete_provider_connections(to_remove)
if removed:
from backend.apps.service.client import sync as _sync
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
_sync(load_settings().model_dump())
return {"ok": True}
return {"ok": False, "error": "Connection not found"}
backend/apps/agents/browser/browser_agent.py
+1 -1
View File
@@ -448,7 +448,7 @@ async def run_browser_agent(
# host (for replay) and front-load turn 1 (this path used to start cold).
preloaded_perception, current_url, preloaded_reads = await _perceive("")
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
from backend.apps.settings.credentials import get_anthropic_client_for_model
from backend.apps.agents.providers.registry import (
_find_builtin_model,
backend/apps/agents/core/mcp_preflight.py
+1 -1
View File
@@ -10,7 +10,7 @@ from typing import Any
from backend.apps.agents.providers.registry import resolve_aux_model
from backend.apps.settings.credentials import get_anthropic_client_for_model
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
from backend.apps.tools_lib.tools_lib import _load_all as load_all_tools
logger = logging.getLogger(__name__)
backend/apps/agents/proxy/anthropic_proxy.py
+2 -2
View File
@@ -323,7 +323,7 @@ def p_pick_upstream(model: str) -> tuple[str, dict[str, str]]:
user has their own key, avoids the 8h OAuth expiry pain)
3. Fallback → 9router (cc/ OAuth subscription, may 401 if expired)
Everything non-Claude goes to 9router for translation."""
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
s = load_settings()
if p_is_claude_model(model):
@@ -384,7 +384,7 @@ async def proxy(rest: str, request: Request):
forward_to_openai,
forward_to_openrouter,
)
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
s = load_settings()
if p_is_openai_max_completion_tokens_model(model):
oak = (getattr(s, "openai_api_key", "") or "").strip()
backend/apps/auth/router.py
+2 -1
View File
@@ -27,7 +27,8 @@ from pydantic import BaseModel
from backend.config.Apps import SubApp
from backend.apps.settings.credentials import OPENSWARM_DEFAULT_PROXY_URL
from backend.apps.settings.settings import load_settings, save_settings_async
from backend.apps.settings.store import load_settings
from backend.apps.settings.settings import save_settings_async
logger = logging.getLogger(__name__)
backend/apps/dashboards/dashboards.py
+1 -1
View File
@@ -315,7 +315,7 @@ async def generate_name(dashboard_id: str):
fallback = " ".join(prompts[0].split()[:4])[:36] or "Untitled Dashboard"
try:
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
from backend.apps.settings.credentials import get_anthropic_client_for_model
from backend.apps.agents.providers.registry import resolve_aux_model
global_settings = load_settings()
backend/apps/outputs/html_inject.py
+1 -1
View File
@@ -32,7 +32,7 @@ def get_anthropic_client(api_model: str | None = None):
get_anthropic_client,
get_anthropic_client_for_model,
)
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
settings = load_settings()
if api_model:
return get_anthropic_client_for_model(settings, api_model)
backend/apps/outputs/outputs.py
+1 -1
View File
@@ -19,7 +19,7 @@ from backend.apps.outputs.view_builder_templates import (
load_app_builder_skill,
seed_webapp_template_workspace,
)
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
from backend.config.paths import OUTPUTS_DIR as DATA_DIR, OUTPUTS_WORKSPACE_DIR as WORKSPACE_DIR
from backend.apps.outputs.html_inject import (
get_anthropic_client,
backend/apps/service/service.py
+1 -1
View File
@@ -118,7 +118,7 @@ async def service_lifespan():
global P_PULSE_TASK, P_DRAIN_TASK
try:
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
from backend.apps.settings.store import save_settings
settings = load_settings()
backend/apps/settings/settings.py
+39 -37
View File
@@ -13,8 +13,8 @@ from backend.config.Apps import SubApp
from backend.apps.settings.models import AppSettings, DEFAULT_SYSTEM_PROMPT
from backend.apps.settings.store import (
DATA_DIR,
load_settings,
_atomic_write_settings,
load_settings as load_settings_from_store,
atomic_write_settings,
)
logger = logging.getLogger(__name__)
@@ -27,7 +27,7 @@ async def settings_lifespan():
from backend.apps.nine_router.process import ensure_running, is_running
from backend.apps.nine_router.sync import sync_gemini_api_key, sync_openai_api_key, sync_openrouter_api_key
from backend.apps.nine_router.sync_custom import sync_openswarm_pro_as_claude, sync_custom_providers
s = load_settings()
s = load_settings_from_store()
async def boot_router_then_sync():
"""Boot 9Router then push key-based connections (sequential: sync helpers no-op pre-boot)."""
@@ -67,7 +67,7 @@ async def settings_lifespan():
async def p_upload_dir_gc_loop():
"""Daily GC of UPLOAD_DIR. Without this, every PDF/image the user
"""Daily GC of P_UPLOAD_DIR. Without this, every PDF/image the user
drops sits in the OS temp dir forever, growing unbounded across
sessions. We keep files for 7 days to make resume-after-restart
work, then delete. macOS temp under /var/folders/... is auto-purged
@@ -78,9 +78,9 @@ async def p_upload_dir_gc_loop():
try:
now = time.time()
cutoff = now - 7 * 86400
if os.path.isdir(UPLOAD_DIR):
for entry in os.listdir(UPLOAD_DIR):
p = os.path.join(UPLOAD_DIR, entry)
if os.path.isdir(P_UPLOAD_DIR):
for entry in os.listdir(P_UPLOAD_DIR):
p = os.path.join(P_UPLOAD_DIR, entry)
try:
if os.path.isfile(p) and os.path.getmtime(p) < cutoff:
os.remove(p)
@@ -94,21 +94,22 @@ async def p_upload_dir_gc_loop():
settings = SubApp("settings", settings_lifespan)
# Public - Used by auth.router.py, free_trial.py, subscription.router.py
async def save_settings_async(settings_obj: AppSettings) -> None:
"""Async atomic save via thread pool; shares the lock with the sync variant."""
payload = settings_obj.model_dump()
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, _atomic_write_settings, payload)
await loop.run_in_executor(None, atomic_write_settings, payload)
@settings.router.get("")
async def get_settings():
return load_settings().model_dump()
return load_settings_from_store().model_dump()
# Written only by their dedicated flows (Stripe activate, sign-in, signout, OAuth connects);
# a full-object PUT from a stale renderer snapshot must never revert or forge them.
SERVER_OWNED_FIELDS = (
P_SERVER_OWNED_FIELDS = (
"connection_mode",
"openswarm_bearer_token",
"openswarm_proxy_url",
@@ -131,8 +132,8 @@ SERVER_OWNED_FIELDS = (
async def update_settings(body: AppSettings):
from backend.apps.service.client import sync
old = load_settings()
for k in SERVER_OWNED_FIELDS:
old = load_settings_from_store()
for k in P_SERVER_OWNED_FIELDS:
setattr(body, k, getattr(old, k, None))
# If the user connects their own model while the free trial is armed, hand
@@ -140,8 +141,8 @@ async def update_settings(body: AppSettings):
# owned, so the loop above just restored it to "free-trial") would keep them
# pinned to the forced Haiku lane even though they pasted a real key.
if getattr(old, "connection_mode", "own_key") == "free-trial":
from backend.apps.subscription.free_trial import _has_own_model
if _has_own_model(body):
from backend.apps.subscription.free_trial import has_own_model
if has_own_model(body):
body.connection_mode = "own_key"
body.free_trial_token = None
body.free_trial_remaining = None
@@ -256,13 +257,13 @@ class AppThemeOverridePayload(BaseModel):
@settings.router.get("/app-theme-override")
async def get_app_theme_override():
"""Cross-app theme preference for App Builder workspaces; backend-held because each app uses its own localStorage origin."""
return {"mode": load_settings().app_template_theme_override}
return {"mode": load_settings_from_store().app_template_theme_override}
@settings.router.put("/app-theme-override")
async def put_app_theme_override(body: AppThemeOverridePayload):
"""MERGE the override; the general PUT /api/settings replaces the whole object and would blank secrets, logging the user out."""
current = load_settings()
current = load_settings_from_store()
current.app_template_theme_override = body.mode
await save_settings_async(current)
return {"ok": True, "mode": current.app_template_theme_override}
@@ -275,7 +276,7 @@ async def get_default_system_prompt():
@settings.router.post("/reset-system-prompt")
async def reset_system_prompt():
current = load_settings()
current = load_settings_from_store()
current.default_system_prompt = DEFAULT_SYSTEM_PROMPT
await save_settings_async(current)
return {"ok": True, "settings": current.model_dump()}
@@ -288,10 +289,11 @@ class BrowseResponse(BaseModel):
files: list[str]
UPLOAD_DIR = os.path.join(tempfile.gettempdir(), "self-swarm-uploads")
os.makedirs(UPLOAD_DIR, exist_ok=True)
P_UPLOAD_DIR = os.path.join(tempfile.gettempdir(), "self-swarm-uploads")
os.makedirs(P_UPLOAD_DIR, exist_ok=True)
# Public - Used by attatchments.py
def sniff_file_kind(contents: bytes) -> tuple[str, str | None]:
"""Classify an uploaded file as text/pdf/image/binary so the agent
layer can route it (inline as text, send as native document/image
@@ -333,7 +335,7 @@ def sniff_file_kind(contents: bytes) -> tuple[str, str | None]:
return ("binary", None)
def _estimate_pdf_tokens(contents: bytes) -> int:
def p_estimate_pdf_tokens(contents: bytes) -> int:
"""Conservative PDF token estimate without a parser dep.
We use two signals and take the MAX so the chip + dry-run never
@@ -350,15 +352,15 @@ def _estimate_pdf_tokens(contents: bytes) -> int:
Taking max() means a small page count on a huge PDF (image-heavy)
still reads as expensive, and a huge page count on a small PDF still
reads as expensive. The chip never lies that an attachment is cheap."""
import re as _re
import re
by_pages = 0
try:
# Prefer the root catalog's /Pages entry. PDFs can have nested
# /Count fields (outlines, sub-pages), so anchor on /Type /Pages.
m = _re.search(rb"/Type\s*/Pages\b[^>]{0,200}?/Count\s+(\d+)", contents, _re.DOTALL)
m = re.search(rb"/Type\s*/Pages\b[^>]{0,200}?/Count\s+(\d+)", contents, re.DOTALL)
if not m:
# Fallback: catalog declares /Pages then references /Count via /Kids.
m = _re.search(rb"/Pages[^>]{0,200}?/Count\s+(\d+)", contents, _re.DOTALL)
m = re.search(rb"/Pages[^>]{0,200}?/Count\s+(\d+)", contents, re.DOTALL)
if m:
pages = int(m.group(1))
if 0 < pages < 10_000:
@@ -401,7 +403,7 @@ async def upload_files(files: list[UploadFile] = File(...)):
# would observe `dest` free and both would write, with the second
# winning. O_EXCL fails the create if anyone else got there first.
base, ext = os.path.splitext(safe_name)
dest = os.path.join(UPLOAD_DIR, safe_name)
dest = os.path.join(P_UPLOAD_DIR, safe_name)
counter = 0
fd = None
while fd is None:
@@ -411,7 +413,7 @@ async def upload_files(files: list[UploadFile] = File(...)):
counter += 1
if counter > 10_000:
raise HTTPException(status_code=500, detail="upload dedup exhausted")
dest = os.path.join(UPLOAD_DIR, f"{base}_{counter}{ext}")
dest = os.path.join(P_UPLOAD_DIR, f"{base}_{counter}{ext}")
try:
with os.fdopen(fd, "wb") as fh:
fh.write(contents)
@@ -432,7 +434,7 @@ async def upload_files(files: list[UploadFile] = File(...)):
except Exception:
tokens_est = min(len(contents), 512_000) // 4
elif kind == "pdf":
tokens_est = _estimate_pdf_tokens(contents)
tokens_est = p_estimate_pdf_tokens(contents)
elif kind == "image":
tokens_est = 1_500
else:
@@ -450,20 +452,20 @@ async def upload_files(files: list[UploadFile] = File(...)):
return JSONResponse({"files": results})
class _SummarizeRequest(BaseModel):
class P_SummarizeRequest(BaseModel):
path: str
target_tokens: int = 4_000
primary_model: Optional[str] = None
@settings.router.post("/summarize-file")
async def summarize_file(req: _SummarizeRequest):
async def summarize_file(req: P_SummarizeRequest):
"""Compress an attached file down to a fact-dense summary the agent can
still reason over without paying the full token cost.
Called from the chat-input attach handler when one file alone would
exceed 50% of the selected model's context window. The summary is
written to a sibling file with `.summary.txt` suffix in UPLOAD_DIR so
written to a sibling file with `.summary.txt` suffix in P_UPLOAD_DIR so
the existing attachment plumbing (paths flow through context_paths)
works unchanged. Aux model picked via provider-agnostic
resolve_aux_model, so users on OpenAI/Gemini/OpenRouter get summarized
@@ -472,7 +474,7 @@ async def summarize_file(req: _SummarizeRequest):
src = req.path
if not os.path.isfile(src):
raise HTTPException(status_code=404, detail="file not found")
if not os.path.commonpath([os.path.realpath(src), os.path.realpath(UPLOAD_DIR)]) == os.path.realpath(UPLOAD_DIR):
if not os.path.commonpath([os.path.realpath(src), os.path.realpath(P_UPLOAD_DIR)]) == os.path.realpath(P_UPLOAD_DIR):
raise HTTPException(status_code=400, detail="path outside upload dir")
try:
@@ -487,8 +489,8 @@ async def summarize_file(req: _SummarizeRequest):
try:
from backend.apps.agents.providers.registry import resolve_aux_model, get_api_type
from backend.apps.settings.credentials import get_anthropic_client_for_model
s = load_settings()
aux_model, _base = await resolve_aux_model(
s = load_settings_from_store()
aux_model, _ = await resolve_aux_model(
s,
preferred_tier="haiku",
primary_api=get_api_type(req.primary_model) if req.primary_model else None,
@@ -512,7 +514,7 @@ async def summarize_file(req: _SummarizeRequest):
CHUNK_CHARS = 200_000
is_chunked = len(raw) > CHUNK_CHARS
async def _summarize_block(text: str, target_tokens: int, label: str) -> str:
async def summarize_block(text: str, target_tokens: int, label: str) -> str:
user = (
f"Target length: ~{target_tokens} tokens.\n\n"
f"<document path=\"{label}\">\n{text}\n</document>\n\n"
@@ -534,7 +536,7 @@ async def summarize_file(req: _SummarizeRequest):
return out
if not is_chunked:
summary = await _summarize_block(raw, req.target_tokens, os.path.basename(src))
summary = await summarize_block(raw, req.target_tokens, os.path.basename(src))
else:
chunks = [raw[i:i + CHUNK_CHARS] for i in range(0, len(raw), CHUNK_CHARS)]
per_chunk_budget = max(800, req.target_tokens // len(chunks) + 600)
@@ -544,15 +546,15 @@ async def summarize_file(req: _SummarizeRequest):
# provider's per-key rate limit, and a single user summarizing
# one file will never hit that.
partials = await asyncio.gather(*[
_summarize_block(ch, per_chunk_budget, f"{os.path.basename(src)} (part {i + 1} of {len(chunks)})")
summarize_block(ch, per_chunk_budget, f"{os.path.basename(src)} (part {i + 1} of {len(chunks)})")
for i, ch in enumerate(chunks)
])
merge_input = "\n\n".join(f"## Part {i + 1}\n{p}" for i, p in enumerate(partials))
summary = await _summarize_block(merge_input, req.target_tokens, f"merged summary of {os.path.basename(src)}")
summary = await summarize_block(merge_input, req.target_tokens, f"merged summary of {os.path.basename(src)}")
except Exception as e:
raise HTTPException(status_code=502, detail=f"summarize failed: {e}")
base, _ext = os.path.splitext(src)
base, _ = os.path.splitext(src)
dest = f"{base}.summary.txt"
counter = 1
while os.path.exists(dest):
backend/apps/settings/store.py
+2 -2
View File
@@ -114,7 +114,7 @@ def load_settings() -> AppSettings:
_settings_write_lock = threading.Lock()
def _atomic_write_settings(payload: dict) -> None:
def atomic_write_settings(payload: dict) -> None:
"""Atomic SETTINGS_FILE write; call via save_settings*, not directly."""
global _cached_settings, _cached_sig
with _settings_write_lock:
@@ -147,4 +147,4 @@ def _atomic_write_settings(payload: dict) -> None:
def save_settings(settings_obj: AppSettings) -> None:
"""Sync atomic persist; thread-safe. Async callers should prefer save_settings_async (Defender can stretch writes to 50-200ms)."""
_atomic_write_settings(settings_obj.model_dump())
atomic_write_settings(settings_obj.model_dump())
backend/apps/subscription/free_trial.py
+2 -2
View File
@@ -78,7 +78,7 @@ def _fingerprint(settings_obj) -> str | None:
return hashlib.sha256((_FP_SALT + raw).encode("utf-8")).hexdigest()
def _has_own_model(s) -> bool:
def has_own_model(s) -> bool:
"""True if the user already has any real model path in settings; never shadow it."""
if any(getattr(s, k, None) for k in (
"anthropic_api_key", "openai_api_key", "google_api_key", "openrouter_api_key",
@@ -142,7 +142,7 @@ async def arm_free_trial(settings_obj) -> dict:
mode = getattr(settings_obj, "connection_mode", "own_key")
if mode not in ("own_key", "free-trial"):
return {"armed": False, "reason": "other_mode"}
if _has_own_model(settings_obj) or await _has_connected_subscription():
if has_own_model(settings_obj) or await _has_connected_subscription():
# A real model exists now (key, custom provider, or a 9Router sub). If we
# were on the free lane, hand the wheel back instead of re-arming.
if mode == "free-trial":
backend/apps/subscription/router.py
+2 -1
View File
@@ -12,7 +12,8 @@ from pydantic import BaseModel
from backend.config.Apps import SubApp
from backend.apps.settings.credentials import OPENSWARM_DEFAULT_PROXY_URL
from backend.apps.settings.settings import load_settings, save_settings_async
from backend.apps.settings.store import load_settings
from backend.apps.settings.settings import save_settings_async
logger = logging.getLogger(__name__)
backend/apps/web/web.py
+2 -2
View File
@@ -164,7 +164,7 @@ def _format_grounded_as_fetch(grounded: dict, url: str) -> str:
def _resolve_gemini_api_key() -> str | None:
"""Pull the AI Studio API key from settings, or None."""
try:
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
s = load_settings()
return getattr(s, "google_api_key", None) or None
except Exception:
@@ -173,7 +173,7 @@ def _resolve_gemini_api_key() -> str | None:
def _resolve_openai_api_key() -> str | None:
try:
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
s = load_settings()
return getattr(s, "openai_api_key", None) or None
except Exception:
backend/tests/test_auth_router.py
+9 -9
View File
@@ -31,11 +31,11 @@ def client():
@pytest.fixture
def reset_settings():
"""Snapshot + restore settings around each test so writes don't leak."""
from backend.apps.settings.settings import load_settings, _save_settings
from backend.apps.settings.store import load_settings, save_settings
original = load_settings().model_copy(deep=True)
yield
_save_settings(original)
save_settings(original)
# ---------------------------------------------------------------------------
@@ -72,7 +72,7 @@ def test_signin_activate_persists_user_id(client):
assert body["plan"] == "free"
# Persisted to settings.
from backend.apps.settings.settings import load_settings
from backend.apps.settings.store import load_settings
s = load_settings()
assert s.user_id == "u-1234"
assert s.user_email == "smoke@example.com"
@@ -145,14 +145,14 @@ def test_signin_activate_short_token_rejected_locally(client):
@pytest.mark.usefixtures("reset_settings")
def test_signout_clears_local_identity(client):
from backend.apps.settings.settings import load_settings, _save_settings
from backend.apps.settings.store import load_settings, save_settings
s = load_settings()
s.user_id = "u-bye"
s.user_email = "bye@example.com"
s.signin_method = "google"
s.openswarm_bearer_token = "bearer-to-revoke-xxxxxxxx"
s.connection_mode = "openswarm-pro"
_save_settings(s)
save_settings(s)
fake_response = AsyncMock()
fake_response.status_code = 200
@@ -174,16 +174,16 @@ def test_signout_clears_local_identity(client):
@pytest.mark.usefixtures("reset_settings")
def test_signout_succeeds_even_when_cloud_unreachable(client):
"""A flaky network shouldn't strand the user signed-in locally."""
from backend.apps.settings.settings import load_settings, _save_settings
from backend.apps.settings.store import load_settings, save_settings
s = load_settings()
s.user_id = "u-flaky"
s.openswarm_bearer_token = "bearer-flaky-network-xxxx"
_save_settings(s)
save_settings(s)
with patch("httpx.AsyncClient") as MockClient:
instance = MockClient.return_value.__aenter__.return_value
import httpx as _httpx
instance.post = AsyncMock(side_effect=_httpx.HTTPError("network down"))
import httpx
instance.post = AsyncMock(side_effect=httpx.HTTPError("network down"))
r = client.post("/api/auth/signout")
assert r.status_code == 200