| 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 |
|
|
| |
| |
| |
| 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() |
|
|
| |
| |
| 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 |
|
|
| |
| 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() |
|
|
| |
| 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"}}, |
| ) |
|
|
| |
| 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() |
|
|
| |
| 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 |
| |
| |
| assert "item.token || item.raw" in script |
| assert "items" in script |
| assert "onChange" in script |
| |
| 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") |
|
|
| |
| |
| assert "disabled={locked || disabled}" in script |
| |
| assert '"API keys cannot contain commas."' in script |
| |
| assert "reserved; enter a real API key" in script |
| |
| 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 |
|
|