from pathlib import Path from unittest.mock import AsyncMock, patch import httpx import pytest from fastapi.testclient import TestClient from free_claude_code.application.connected_accounts import ( ConnectedAccountLoginMode, ConnectedAccountState, ConnectedAccountStatus, ) from free_claude_code.application.model_metadata import ( ProviderModelInfo, ProviderModelRefreshResult, ) from free_claude_code.config.admin.values import MASKED_SECRET from free_claude_code.config.server_urls import local_admin_url from free_claude_code.config.settings import OpenAICompatibleInstance, Settings from tests.api.support import create_test_app, provider_manager_for_app def _local_client(app): return TestClient(app, client=("127.0.0.1", 50000)) def _set_home(monkeypatch, tmp_path: Path) -> None: monkeypatch.setenv("HOME", str(tmp_path)) monkeypatch.setenv("USERPROFILE", str(tmp_path)) monkeypatch.chdir(tmp_path) def _clear_process_config(monkeypatch) -> None: from free_claude_code.config.admin.manifest import FIELD_BY_KEY # Clear every admin-managed env var plus fixed runtime-only keys so the # admin apply tests are independent of the ambient shell environment # (exported provider keys, debug/log flags, voice settings, ...). keys = { "FCC_ENV_FILE", "HOST", "PORT", "LOG_FILE", "ZAI_BASE_URL", "CLAUDE_WORKSPACE", "CLAUDE_CLI_BIN", "LOG_RAW_SSE_EVENTS", } keys.update(FIELD_BY_KEY) for key in keys: monkeypatch.delenv(key, raising=False) def test_admin_page_is_loopback_only(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) app = create_test_app() assert _local_client(app).get("/admin").status_code == 200 remote_client = TestClient(app, client=("203.0.113.10", 50000)) assert remote_client.get("/admin").status_code == 403 def test_admin_routes_are_reachable_remotely_inside_hf_space(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) space_settings = Settings.model_validate({"SPACE_ID": "sshinmen/shin"}) with patch( "free_claude_code.api.admin_routes.get_settings", return_value=space_settings, ): app = create_test_app() remote_client = TestClient(app, client=("203.0.113.10", 50000)) assert remote_client.get("/admin").status_code == 200 assert remote_client.get("/admin/api/config").status_code == 200 def _admin_asset_urls() -> tuple[str, ...]: """Every built admin asset URL, so cache policy covers hashed bundles.""" static_assets = Path("src/free_claude_code/api/admin_static/assets") return tuple( f"/admin/assets/{path.name}" for path in sorted(static_assets.glob("*")) ) @pytest.mark.parametrize( "path", ("/admin", "/admin/api/config", *_admin_asset_urls()), ) def test_admin_responses_are_never_cached(monkeypatch, tmp_path, path): _set_home(monkeypatch, tmp_path) response = _local_client(create_test_app()).get(path) assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" @pytest.mark.parametrize( ("path", "client_host", "expected_status"), ( ("/admin", "203.0.113.10", 403), ("/admin/assets/missing.js", "127.0.0.1", 404), ), ) def test_admin_http_errors_are_never_cached( monkeypatch, tmp_path, path, client_host, expected_status, ): _set_home(monkeypatch, tmp_path) client = TestClient(create_test_app(), client=(client_host, 50000)) response = client.get(path) assert response.status_code == expected_status assert response.headers["cache-control"] == "no-store" def test_admin_validation_errors_are_never_cached(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) response = _local_client(create_test_app()).post( "/admin/api/config/validate", content="{", headers={"Content-Type": "application/json"}, ) assert response.status_code == 422 assert response.headers["cache-control"] == "no-store" def test_admin_unexpected_errors_are_never_cached(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) client = TestClient( create_test_app(), client=("127.0.0.1", 50000), raise_server_exceptions=False, ) with patch( "free_claude_code.api.admin_routes.load_config_response", side_effect=RuntimeError("test error"), ): response = client.get("/admin/api/config") assert response.status_code == 500 assert response.headers["cache-control"] == "no-store" def test_admin_cache_policy_does_not_match_similar_public_paths(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) response = _local_client(create_test_app()).get("/administrator") assert response.status_code == 404 assert "cache-control" not in response.headers def test_admin_api_fetches_bypass_browser_cache(): script = Path("frontend/src/lib/api.ts").read_text(encoding="utf-8") assert 'cache: "no-store"' in script def test_admin_connected_account_login_preopens_sign_in_window(): app_source = Path("frontend/src/App.tsx").read_text(encoding="utf-8") cards = Path("frontend/src/components/ProviderCards.tsx").read_text( encoding="utf-8" ) assert 'window.open("about:blank", "_blank")' in app_source assert "popup.location.replace(target)" in app_source assert "if (popup) popup.close()" in app_source assert "Reconnect" in cards assert "Copy code" in cards assert "Restart your agent to refresh its model picker." in cards assert 'window.confirm("Disconnect this ChatGPT account from FCC?")' in app_source class _FakeConnectedAccount: def __init__(self) -> None: self.connected = False self.revision = 0 self.cancelled = False def is_connected(self) -> bool: return self.connected def status(self) -> ConnectedAccountStatus: return ConnectedAccountStatus( provider_id="openai", state=( ConnectedAccountState.CONNECTED if self.connected else ConnectedAccountState.DISCONNECTED ), connected=self.connected, revision=self.revision, email="safe@example.com" if self.connected else None, ) async def start_login( self, mode: ConnectedAccountLoginMode ) -> ConnectedAccountStatus: return ConnectedAccountStatus( provider_id="openai", state=ConnectedAccountState.CONNECTING, connected=False, revision=self.revision, attempt_id="login_safe", mode=mode, authorization_url="https://auth.openai.com/safe", ) async def cancel_login(self) -> ConnectedAccountStatus: self.cancelled = True return self.status() async def disconnect(self) -> ConnectedAccountStatus: self.connected = False self.revision += 1 return self.status() async def close(self) -> None: return None def test_admin_connected_account_routes_are_safe_loopback_only_and_uncached( monkeypatch, tmp_path ): _set_home(monkeypatch, tmp_path) account = _FakeConnectedAccount() app = create_test_app(connected_accounts={"openai": account}) client = _local_client(app) status_response = client.get("/admin/api/providers/openai/auth") login_response = client.post( "/admin/api/providers/openai/auth/login", json={"mode": "browser"}, ) cancel_response = client.post("/admin/api/providers/openai/auth/cancel") assert status_response.status_code == 200 assert status_response.headers["cache-control"] == "no-store" assert status_response.json()["state"] == "disconnected" assert login_response.status_code == 200 assert login_response.json() == { "provider_id": "openai", "state": "connecting", "connected": False, "revision": 0, "attempt_id": "login_safe", "mode": "browser", "authorization_url": "https://auth.openai.com/safe", } assert "token" not in login_response.text.lower() assert cancel_response.status_code == 200 assert account.cancelled is True remote = TestClient(app, client=("203.0.113.10", 50000)) assert remote.get("/admin/api/providers/openai/auth").status_code == 403 def test_admin_rejects_auth_routes_for_non_connected_provider(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) response = _local_client(create_test_app()).get( "/admin/api/providers/nvidia_nim/auth" ) assert response.status_code == 404 def test_admin_provider_cards_support_non_key_configuration(): script = Path("frontend/src/App.tsx").read_text(encoding="utf-8") cards = Path("frontend/src/components/ProviderCards.tsx").read_text( encoding="utf-8" ) assert '"missing_config"' in script assert "provider.configuration" in cards def test_admin_page_no_longer_renders_generated_env_panel(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) app = create_test_app() response = _local_client(app).get("/admin") assert response.status_code == 200 assert "Generated Env" not in response.text assert "envPreview" not in response.text def test_admin_page_no_longer_renders_global_status_header(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) app = create_test_app() response = _local_client(app).get("/admin") assert response.status_code == 200 assert "Local Admin" not in response.text assert "serverStatus" not in response.text assert "modelBadge" not in response.text def test_admin_static_no_longer_fetches_global_status_header(): api_source = Path("frontend/src/lib/api.ts").read_text(encoding="utf-8") app_source = Path("frontend/src/App.tsx").read_text(encoding="utf-8") assert 'api("/admin/api/status")' not in api_source + app_source assert "updateHeader" not in api_source + app_source assert '"Running"' not in api_source + app_source assert "serverStatus" not in api_source + app_source assert "modelBadge" not in api_source + app_source def test_admin_static_drops_managed_source_label(): source = Path("frontend/src/components/FieldControl.tsx").read_text( encoding="utf-8" ) assert "managed_env" not in source assert "hasOwnProperty.call(labels," in source assert 'parts.push("locked")' in source assert "labels[field.source]" in source def test_admin_static_places_reasoning_fields_in_model_config(): script = Path("frontend/src/App.tsx").read_text(encoding="utf-8") assert 'sections: ["models", "reasoning", "web_tools"]' in script assert 'sections: ["models", "thinking", "web_tools"]' not in script def test_admin_static_model_combobox_owns_dropdown_and_search_behavior(): script = Path("frontend/src/components/ModelCombobox.tsx").read_text( encoding="utf-8" ) control = Path("frontend/src/components/FieldControl.tsx").read_text( encoding="utf-8" ) assert "fetchModelOptions(true)" in Path("frontend/src/App.tsx").read_text( encoding="utf-8" ) assert 'case "model":' in control assert 'case "optional_model":' in control assert 'role="combobox"' in script assert 'role="listbox"' in script assert "aria-haspopup" in script assert "ArrowDown" in script assert "ArrowUp" in script assert "event.key === " in script assert "Enter" in script assert "Escape" in script assert "datalist" not in script assert "visible = useMemo" in script def test_admin_static_model_combobox_preserves_custom_slugs_and_none_semantics(): script = Path("frontend/src/components/ModelCombobox.tsx").read_text( encoding="utf-8" ) app_source = Path("frontend/src/App.tsx").read_text(encoding="utf-8") control = Path("frontend/src/components/FieldControl.tsx").read_text( encoding="utf-8" ) assert '["None", ...models]' in script assert "Press Enter to use this model id." in script assert "Use "{draft}"" in script assert "commitDraft" in script assert 'case "optional_model":' in control assert 'fieldType === "optional_model"' in script assert "hydrateModelOptions" in app_source assert "Model fields remain editable" in app_source assert "failed_providers" in app_source assert '"warn"' in app_source def test_admin_openai_compatible_view_supports_manual_model_entries(): view = Path("frontend/src/components/OpenAICompatibleView.tsx").read_text( encoding="utf-8" ) app_source = Path("frontend/src/App.tsx").read_text(encoding="utf-8") assert "Add model id, comma-separated" in view assert 'split(",")' in view assert "onAddModels" in view assert "Use as default" in view assert "handleAddEndpointModels" in app_source assert "onAddModels={handleAddEndpointModels}" in app_source def test_admin_config_masks_secrets_and_exposes_manifest(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).get("/admin/api/config") assert response.status_code == 200 body = response.json() assert isinstance(body["version"], str) and body["version"] keys = {field["key"] for field in body["fields"]} assert "MODEL_FABLE" in keys assert "REASONING_FABLE" in keys assert "ANTHROPIC_AUTH_TOKEN" in keys assert "OPENROUTER_API_KEY" in keys assert "AWS_BEARER_TOKEN_BEDROCK" in keys assert "BEDROCK_BASE_URL" in keys assert "FIREWORKS_API_KEY" in keys assert "CLOUDFLARE_API_TOKEN" in keys assert "CLOUDFLARE_ACCOUNT_ID" in keys assert "GITHUB_MODELS_TOKEN" in keys assert "GEMINI_API_KEY" in keys assert "GROQ_API_KEY" in keys assert "SAMBANOVA_API_KEY" in keys assert "TELEGRAM_PROXY_URL" in keys assert "CEREBRAS_API_KEY" in keys assert "OLLAMA_API_KEY" in keys assert "FCC_OPEN_BROWSER" in keys assert "ZAI_BASE_URL" not in keys assert "CLAUDE_WORKSPACE" not in keys assert "CLAUDE_CLI_BIN" not in keys assert "LOG_FILE" not in keys auth_field = next( field for field in body["fields"] if field["key"] == "ANTHROPIC_AUTH_TOKEN" ) assert auth_field["secret"] is True assert auth_field["value"] == MASKED_SECRET assert auth_field["source"] == "template" telegram_proxy_field = next( field for field in body["fields"] if field["key"] == "TELEGRAM_PROXY_URL" ) assert telegram_proxy_field["secret"] is True open_browser_field = next( field for field in body["fields"] if field["key"] == "FCC_OPEN_BROWSER" ) assert open_browser_field["type"] == "boolean" assert open_browser_field["value"] == "true" assert open_browser_field["restart_required"] is False model_field_types = { field["key"]: field["type"] for field in body["fields"] if field["key"] in {"MODEL", "MODEL_FABLE", "MODEL_OPUS", "MODEL_SONNET", "MODEL_HAIKU"} } assert model_field_types == { "MODEL": "model", "MODEL_FABLE": "optional_model", "MODEL_OPUS": "optional_model", "MODEL_SONNET": "optional_model", "MODEL_HAIKU": "optional_model", } reasoning_policy = next( field for field in body["fields"] if field["key"] == "REASONING_POLICY" ) assert reasoning_policy["section"] == "reasoning" assert reasoning_policy["type"] == "select" assert reasoning_policy["value"] == "client" assert reasoning_policy["options"] == [ {"value": "off", "label": "Off"}, {"value": "client", "label": "From client"}, {"value": "low", "label": "Low"}, {"value": "medium", "label": "Medium"}, {"value": "high", "label": "High"}, {"value": "xhigh", "label": "X-High"}, {"value": "max", "label": "Max"}, ] route_reasoning = next( field for field in body["fields"] if field["key"] == "REASONING_FABLE" ) assert route_reasoning["options"] == [ {"value": "inherit", "label": "Inherit"}, *reasoning_policy["options"], ] restart_required = { field["key"] for field in body["fields"] if field["restart_required"] is True } assert { "ANTHROPIC_AUTH_TOKEN", "DEBUG_PLATFORM_EDITS", "DEBUG_SUBAGENT_STACK", "LOG_RAW_API_PAYLOADS", "LOG_API_ERROR_TRACEBACKS", "LOG_RAW_MESSAGING_CONTENT", "LOG_RAW_CLI_DIAGNOSTICS", "LOG_MESSAGING_ERROR_DETAILS", } <= restart_required def test_admin_models_include_configured_and_cached_canonical_slugs(): settings = Settings() settings.model = "nvidia_nim/configured-model" settings.model_opus = "open_router/anthropic/configured-opus" settings.open_router_api_key = "open-router-key" app = create_test_app(settings) provider_manager_for_app(app).cache_model_infos( "open_router", { ProviderModelInfo("anthropic/configured-opus"), ProviderModelInfo("meta/llama-3.3"), }, ) response = _local_client(app).get("/admin/api/models") assert response.status_code == 200 assert response.json() == { "models": [ "nvidia_nim/configured-model", "open_router/anthropic/configured-opus", "open_router/meta/llama-3.3", ], "failed_providers": [], } def test_admin_models_include_stored_openai_compatible_instance_models(): """Applied per-endpoint model ids are listed even without live discovery.""" settings = Settings() settings.model = "nvidia_nim/configured-model" settings.openai_compatible_instances = ( OpenAICompatibleInstance( base_url="https://a.example/v1", models=("gpt-4o", "deepseek-v3"), ), OpenAICompatibleInstance(base_url=""), OpenAICompatibleInstance( base_url="https://b.example", models=(" local-model ", ""), ), ) app = create_test_app(settings) response = _local_client(app).get("/admin/api/models") assert response.status_code == 200 assert response.json() == { "models": [ "deepseek-v3", "gpt-4o", "local-model", "nvidia_nim/configured-model", "openai_compatible_1/deepseek-v3", "openai_compatible_1/gpt-4o", "openai_compatible_3/local-model", ], "failed_providers": [], } def test_admin_model_refresh_returns_the_updated_canonical_catalog(): settings = Settings() settings.model = "deepseek/deepseek-chat" settings.deepseek_api_key = "deepseek-key" app = create_test_app(settings) runtime = app.state.services.admin async def refresh_models() -> ProviderModelRefreshResult: provider_manager_for_app(app).cache_model_infos( "deepseek", {ProviderModelInfo("deepseek-reasoner")}, ) return ProviderModelRefreshResult(refreshed_provider_ids=("deepseek",)) runtime.refresh_models = AsyncMock(side_effect=refresh_models) response = _local_client(app).post("/admin/api/models/refresh") assert response.status_code == 200 assert response.json() == { "models": ["deepseek/deepseek-chat", "deepseek/deepseek-reasoner"], "failed_providers": [], } runtime.refresh_models.assert_awaited_once_with() def test_admin_model_refresh_reports_partial_provider_failures(): settings = Settings() settings.model = "deepseek/deepseek-chat" app = create_test_app(settings) runtime = app.state.services.admin runtime.refresh_models = AsyncMock( return_value=ProviderModelRefreshResult( refreshed_provider_ids=("deepseek",), failed_provider_ids=("open_router",), ) ) response = _local_client(app).post("/admin/api/models/refresh") assert response.status_code == 200 assert response.json() == { "models": ["deepseek/deepseek-chat"], "failed_providers": ["open_router"], } def test_admin_config_preserves_repo_env_source_contract(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.write_text("MODEL=open_router/managed-model\n", encoding="utf-8") app = create_test_app() response = _local_client(app).get("/admin/api/config") assert response.status_code == 200 body = response.json() model_field = next(field for field in body["fields"] if field["key"] == "MODEL") assert model_field["source"] == "repo_env" assert model_field["locked"] is False def test_admin_apply_persists_open_browser_for_next_launch(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"FCC_OPEN_BROWSER": False}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert body["pending_fields"] == [] assert body["restart"] == { "required": False, "automatic": False, "admin_url": None, "fields": [], } managed_env = tmp_path / ".env" assert "FCC_OPEN_BROWSER=false" in managed_env.read_text(encoding="utf-8") def test_admin_apply_masks_telegram_proxy_credentials(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() proxy_url = "https://user:password@proxy.example:8443" response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"TELEGRAM_PROXY_URL": proxy_url}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "TELEGRAM_PROXY_URL=********" in body["env_preview"] assert proxy_url not in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert f"TELEGRAM_PROXY_URL={proxy_url}" in text def test_admin_validate_accepts_bare_model_and_rejects_bad_provider( monkeypatch, tmp_path ): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() # Bare model ids are valid: the provider prefix is optional and resolves # at runtime to whichever OpenAI-compatible endpoint advertises the id. response = _local_client(app).post( "/admin/api/config/validate", json={"values": {"MODEL": "missing-provider-prefix"}}, ) assert response.status_code == 200 assert response.json()["valid"] is True # A prefixed ref with an unknown provider is still rejected. response = _local_client(app).post( "/admin/api/config/validate", json={"values": {"MODEL": "bad_provider/some-model"}}, ) assert response.status_code == 200 body = response.json() assert body["valid"] is False assert any("Invalid provider" in error for error in body["errors"]) def test_admin_apply_writes_complete_managed_env_and_masks_preview( monkeypatch, tmp_path ): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "open_router/test-model", "OPENROUTER_API_KEY": "router-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "OPENROUTER_API_KEY=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text("utf-8") assert "MODEL=open_router/test-model" in text assert "OPENROUTER_API_KEY=router-secret" in text assert "ANTHROPIC_AUTH_TOKEN=" in text assert body["restart"] == { "required": False, "automatic": False, "admin_url": None, "fields": [], } def test_admin_apply_writes_fireworks_key_and_masks_preview(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "fireworks/test-model", "FIREWORKS_API_KEY": "fw-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "FIREWORKS_API_KEY=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=fireworks/test-model" in text assert "FIREWORKS_API_KEY=fw-secret" in text def test_admin_apply_writes_gemini_key_and_masks_preview(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "gemini/models/gemini-3.1-flash-lite", "GEMINI_API_KEY": "gm-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "GEMINI_API_KEY=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=gemini/models/gemini-3.1-flash-lite" in text assert "GEMINI_API_KEY=gm-secret" in text def test_admin_apply_writes_groq_key_and_masks_preview(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "groq/llama-3.3-70b-versatile", "GROQ_API_KEY": "gq-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "GROQ_API_KEY=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=groq/llama-3.3-70b-versatile" in text assert "GROQ_API_KEY=gq-secret" in text def test_admin_apply_writes_sambanova_key_and_masks_preview(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "sambanova/Meta-Llama-3.3-70B-Instruct", "SAMBANOVA_API_KEY": "sn-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "SAMBANOVA_API_KEY=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=sambanova/Meta-Llama-3.3-70B-Instruct" in text assert "SAMBANOVA_API_KEY=sn-secret" in text def test_admin_apply_writes_cerebras_key_and_masks_preview(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "cerebras/llama3.1-8b", "CEREBRAS_API_KEY": "cb-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "CEREBRAS_API_KEY=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=cerebras/llama3.1-8b" in text assert "CEREBRAS_API_KEY=cb-secret" in text def test_admin_apply_writes_bedrock_region_config_and_masks_key(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "bedrock/openai.gpt-oss-120b", "AWS_BEARER_TOKEN_BEDROCK": "bedrock-secret", "BEDROCK_BASE_URL": ("https://bedrock-mantle.us-west-2.api.aws/v1"), } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "AWS_BEARER_TOKEN_BEDROCK=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=bedrock/openai.gpt-oss-120b" in text assert "AWS_BEARER_TOKEN_BEDROCK=bedrock-secret" in text assert "BEDROCK_BASE_URL=https://bedrock-mantle.us-west-2.api.aws/v1" in text def test_admin_apply_writes_cloudflare_fields_and_masks_preview(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "cloudflare/@cf/moonshotai/kimi-k2.6", "CLOUDFLARE_API_TOKEN": "cf-secret", "CLOUDFLARE_ACCOUNT_ID": "cf-account", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "CLOUDFLARE_API_TOKEN=********" in body["env_preview"] assert "CLOUDFLARE_ACCOUNT_ID=cf-account" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=cloudflare/@cf/moonshotai/kimi-k2.6" in text assert "CLOUDFLARE_API_TOKEN=cf-secret" in text assert "CLOUDFLARE_ACCOUNT_ID=cf-account" in text def test_admin_apply_writes_huggingface_key_and_masks_preview(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "huggingface/openai/gpt-oss-120b:fastest", "HUGGINGFACE_API_KEY": "hf-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert body["pending_fields"] == [] assert "HUGGINGFACE_API_KEY=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=huggingface/openai/gpt-oss-120b:fastest" in text assert "HUGGINGFACE_API_KEY=hf-secret" in text @pytest.mark.parametrize( ("device", "credential_key"), [ ("nvidia_nim", "NVIDIA_NIM_API_KEY"), ("cpu", "HUGGINGFACE_API_KEY"), ], ) def test_admin_key_change_requires_restart_for_active_voice_backend( monkeypatch, tmp_path, device, credential_key, ): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( "\n".join( [ "VOICE_NOTE_ENABLED=true", f"WHISPER_DEVICE={device}", f"{credential_key}=old-key", "", ] ), encoding="utf-8", ) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {credential_key: "new-key"}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert body["pending_fields"] == [credential_key] assert body["restart"] == { "required": True, "automatic": False, "admin_url": None, "fields": [credential_key], } @pytest.mark.parametrize( ("key", "initial", "updated"), [ ("ANTHROPIC_AUTH_TOKEN", "old-token", "new-token"), ("DEBUG_PLATFORM_EDITS", "true", "false"), ("DEBUG_SUBAGENT_STACK", "true", "false"), ("LOG_RAW_API_PAYLOADS", "true", "false"), ("LOG_API_ERROR_TRACEBACKS", "true", "false"), ("LOG_RAW_MESSAGING_CONTENT", "true", "false"), ("LOG_RAW_CLI_DIAGNOSTICS", "true", "false"), ("LOG_MESSAGING_ERROR_DETAILS", "true", "false"), ], ) def test_admin_constructor_captured_setting_requires_restart( monkeypatch, tmp_path, key, initial, updated, ): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text(f"{key}={initial}\n", encoding="utf-8") app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {key: updated}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert body["pending_fields"] == [key] assert body["restart"] == { "required": True, "automatic": False, "admin_url": None, "fields": [key], } def test_admin_apply_writes_cohere_key_and_masks_preview(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "cohere/command-a-plus-05-2026", "COHERE_API_KEY": "cohere-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "COHERE_API_KEY=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=cohere/command-a-plus-05-2026" in text assert "COHERE_API_KEY=cohere-secret" in text def test_admin_apply_writes_github_models_token_and_masks_preview( monkeypatch, tmp_path ): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={ "values": { "MODEL": "github_models/openai/gpt-4.1", "GITHUB_MODELS_TOKEN": "github-secret", } }, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert "GITHUB_MODELS_TOKEN=********" in body["env_preview"] env_file = tmp_path / ".env" text = env_file.read_text(encoding="utf-8") assert "MODEL=github_models/openai/gpt-4.1" in text assert "GITHUB_MODELS_TOKEN=github-secret" in text def test_admin_apply_preserves_hidden_diagnostics_and_smoke_values( monkeypatch, tmp_path ): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( "\n".join( [ "MODEL=nvidia_nim/old-model", "LOG_RAW_API_PAYLOADS=true", "FCC_SMOKE_MODEL_ZAI=zai/smoke-model", "", ] ), encoding="utf-8", ) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"MODEL": "open_router/test-model"}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True text = env_file.read_text("utf-8") assert "MODEL=open_router/test-model" in text assert "LOG_RAW_API_PAYLOADS=true" in text assert "FCC_SMOKE_MODEL_ZAI=zai/smoke-model" in text def test_admin_apply_omits_stale_zai_base_url(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( "\n".join( [ "MODEL=zai/glm-5.2", "ZAI_API_KEY=zai-secret", "ZAI_BASE_URL=https://custom.zai.invalid/v1", "", ] ), encoding="utf-8", ) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"MODEL": "zai/glm-5.2"}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True text = env_file.read_text("utf-8") assert "ZAI_API_KEY=zai-secret" in text assert "ZAI_BASE_URL" not in text def test_admin_apply_omits_stale_fixed_claude_runtime_settings(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( "\n".join( [ "MODEL=open_router/test-model", "CLAUDE_WORKSPACE=C:/custom/workspace", "CLAUDE_CLI_BIN=claude-custom", "", ] ), encoding="utf-8", ) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"MODEL": "open_router/test-model"}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True text = env_file.read_text("utf-8") assert "MODEL=open_router/test-model" in text assert "CLAUDE_WORKSPACE" not in text assert "CLAUDE_CLI_BIN" not in text def test_admin_apply_restart_required_reports_automatic_restart(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) callbacks: list[str] = [] async def restart_callback() -> None: callbacks.append("restart") app = create_test_app(restart_callback=restart_callback) response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"PORT": "9090"}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert body["pending_fields"] == ["PORT"] assert body["restart"] == { "required": True, "automatic": True, "admin_url": "http://127.0.0.1:9090/admin", "fields": ["PORT"], } assert callbacks == ["restart"] def test_admin_apply_restart_required_reports_manual_fallback(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"PORT": "9091"}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True assert body["pending_fields"] == ["PORT"] assert body["restart"] == { "required": True, "automatic": False, "admin_url": None, "fields": ["PORT"], } def test_admin_process_env_values_are_locked_and_not_written(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) monkeypatch.setenv("MODEL", "open_router/process-model") app = create_test_app() config = _local_client(app).get("/admin/api/config").json() model_field = next(field for field in config["fields"] if field["key"] == "MODEL") assert model_field["locked"] is True assert model_field["source"] == "process" response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"MODEL": "deepseek/managed-model"}}, ) assert response.status_code == 200 env_file = tmp_path / ".env" assert "deepseek/managed-model" not in env_file.read_text("utf-8") def test_admin_first_apply_migrates_repo_env(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) monkeypatch.chdir(tmp_path) (tmp_path / ".env").write_text( "MODEL=deepseek/deepseek-chat\nDEEPSEEK_API_KEY=deepseek-secret\n", encoding="utf-8", ) app = create_test_app() config = _local_client(app).get("/admin/api/config").json() model_field = next(field for field in config["fields"] if field["key"] == "MODEL") assert model_field["value"] == "deepseek/deepseek-chat" assert model_field["source"] == "repo_env" response = _local_client(app).post( "/admin/api/config/apply", json={"values": {}}, ) assert response.status_code == 200 managed_text = (tmp_path / ".env").read_text("utf-8") assert "MODEL=deepseek/deepseek-chat" in managed_text assert "DEEPSEEK_API_KEY=deepseek-secret" in managed_text def test_admin_local_provider_status_reports_reachable(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() class FakeAsyncClient: def __init__(self, *args, **kwargs): pass async def __aenter__(self): return self async def __aexit__(self, *args): return None async def get(self, url: str): return httpx.Response(200, json={"data": []}) with patch("free_claude_code.api.admin_routes.httpx.AsyncClient", FakeAsyncClient): response = _local_client(app).get("/admin/api/providers/local-status") assert response.status_code == 200 providers = response.json()["providers"] assert {provider["status"] for provider in providers} == {"reachable"} def test_admin_launch_url_uses_loopback_for_wildcard_host(): settings = Settings.model_construct(host="0.0.0.0", port=8082) assert local_admin_url(settings) == "http://127.0.0.1:8082/admin" def test_admin_credential_fields_expose_key_pool_metadata(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( 'OPENROUTER_API_KEY="k1, k2, k3"\n', encoding="utf-8", ) app = create_test_app() response = _local_client(app).get("/admin/api/config") assert response.status_code == 200 fields = {field["key"]: field for field in response.json()["fields"]} credential = fields["OPENROUTER_API_KEY"] assert credential["pool_supported"] is True assert credential["key_count"] == 3 assert credential["keys"] == ["__fcc_key_0__", "__fcc_key_1__", "__fcc_key_2__"] non_credential = fields["TELEGRAM_PROXY_URL"] assert non_credential["pool_supported"] is False def test_admin_pool_reveal_returns_stored_keys(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( 'OPENROUTER_API_KEY="sk-a, sk-b, sk-c"\n', encoding="utf-8", ) app = create_test_app() response = _local_client(app).get("/admin/api/pools/OPENROUTER_API_KEY") assert response.status_code == 200 assert response.json() == { "key": "OPENROUTER_API_KEY", "keys": ["sk-a", "sk-b", "sk-c"], } assert response.headers["cache-control"] == "no-store" def test_admin_pool_reveal_rejects_non_pool_and_unknown_fields(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() for field_key in ("TELEGRAM_PROXY_URL", "NOT_A_FIELD"): response = _local_client(app).get(f"/admin/api/pools/{field_key}") assert response.status_code == 404 assert response.headers["cache-control"] == "no-store" def test_admin_pool_reveal_is_loopback_only(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = TestClient(app, client=("203.0.113.10", 50000)).get( "/admin/api/pools/OPENROUTER_API_KEY" ) assert response.status_code == 403 def test_admin_apply_edits_pool_key_in_place(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( 'OPENROUTER_API_KEY="a, b, c"\n', encoding="utf-8", ) app = create_test_app() # The pool editor reveals key 0, edits it in place, and keeps the others. response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"OPENROUTER_API_KEY": "edited-a,__fcc_key_1__,__fcc_key_2__"}}, ) assert response.status_code == 200 assert response.json()["applied"] is True text = (tmp_path / ".env").read_text("utf-8") assert "OPENROUTER_API_KEY=edited-a,b,c" in text assert "__fcc_key_" not in text def test_admin_apply_persists_initial_key_pool(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"OPENROUTER_API_KEY": "a", "OPENROUTER_API_KEY_MORE": "x"}}, ) # The extra unknown key is dropped; a single new key is persisted as a pool. assert response.status_code == 200 assert response.json()["applied"] is True def test_admin_apply_resolves_pool_tokens_against_stored_keys(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( 'OPENROUTER_API_KEY="a, b, c"\n', encoding="utf-8", ) app = create_test_app() # Remove key "b" (index 1) and keep a new key at the end. response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"OPENROUTER_API_KEY": "__fcc_key_0__,,__fcc_key_2__,new-key"}}, ) assert response.status_code == 200 body = response.json() assert body["applied"] is True text = (tmp_path / ".env").read_text("utf-8") assert "OPENROUTER_API_KEY=a,c,new-key" in text assert "OPENROUTER_API_KEY=a, b, c" not in text assert "__fcc_key_" not in text def test_admin_apply_keeps_pool_unchanged_when_single_mask_submitted( monkeypatch, tmp_path ): _set_home(monkeypatch, tmp_path) _clear_process_config(monkeypatch) env_file = tmp_path / ".env" env_file.parent.mkdir(parents=True, exist_ok=True) env_file.write_text( 'OPENROUTER_API_KEY="a, b"\n', encoding="utf-8", ) app = create_test_app() response = _local_client(app).post( "/admin/api/config/apply", json={"values": {"OPENROUTER_API_KEY": MASKED_SECRET}}, ) assert response.status_code == 200 text = (tmp_path / ".env").read_text("utf-8") assert 'OPENROUTER_API_KEY="a, b"' in text def test_admin_key_pool_editor_present_in_admin_script(): script = Path("frontend/src/components/PoolEditor.tsx").read_text(encoding="utf-8") control = Path("frontend/src/components/FieldControl.tsx").read_text( encoding="utf-8" ) assert "PoolEditor" in script assert "field.pool_supported" in control assert "Remove" in script assert '"Add API key"' in script assert "Add" in script assert "PoolEditor" in control # The hidden pool input must be attached to the editor so changedValues() # can submit tokens/raw keys; without it the UI silently drops edits. assert "item.token || item.raw" in script assert "items" in script assert "onChange" in script # Stored keys can be revealed and edited inline from the pool editor. assert "fetchPoolKeys" in script assert "Show API key" in script assert "Hide API key" in script assert "fieldKey" in script assert "/admin/api/pools/" in Path("frontend/src/lib/api.ts").read_text( encoding="utf-8" ) def test_admin_key_pool_editor_guards_locked_fields_and_reserved_values(): script = Path("frontend/src/components/PoolEditor.tsx").read_text(encoding="utf-8") # Locked (process/env-file) pools: adding is already disabled; removing must # be too so edits are not silently dropped on apply. assert "disabled={locked || disabled}" in script # Keys containing commas would be split into multiple pool entries on save. assert '"API keys cannot contain commas."' in script # The __fcc_key_N__ token pattern is reserved for UI masking tokens. assert "reserved; enter a real API key" in script # Single-key pools get accurate copy instead of the rotation message. assert "One key configured. Add extra keys for round-robin and failover." in script assert "Multiple keys are used in rotation. Add extra keys" in script def test_admin_usage_route_returns_empty_stats(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) response = _local_client(create_test_app()).get("/admin/api/usage") assert response.status_code == 200 payload = response.json() assert payload["stats"]["total_requests"] == 0 assert payload["records"] == [] def test_admin_usage_route_returns_captured_records(monkeypatch, tmp_path): from free_claude_code.core.usage_tracking import ( UsageRecord, get_buffer, reset_buffer, ) reset_buffer() _set_home(monkeypatch, tmp_path) app = create_test_app() buffer = get_buffer() assert buffer is not None buffer.push( UsageRecord( request_id="req-1", timestamp=1000.0, provider="openai_compatible_1", provider_model="gpt-4o", gateway_model="gpt-4o", wire_api="messages", input_tokens=10, output_tokens=20, cache_creation_tokens=5, cache_read_tokens=3, reasoning_tokens=2, duration_ms=100, status="success", error_type=None, prompt="full prompt text", ) ) response = _local_client(app).get("/admin/api/usage") assert response.status_code == 200 payload = response.json() assert payload["stats"]["total_requests"] == 1 assert payload["stats"]["total_input_tokens"] == 10 assert payload["stats"]["total_output_tokens"] == 20 record = payload["records"][0] assert record["request_id"] == "req-1" assert record["prompt"] == "full prompt text" assert record["gateway_model"] == "gpt-4o" def test_admin_usage_route_is_loopback_only(monkeypatch, tmp_path): _set_home(monkeypatch, tmp_path) app = create_test_app() remote_client = TestClient(app, client=("203.0.113.10", 50000)) assert remote_client.get("/admin/api/usage").status_code == 403