shin / tests /api /test_admin.py
shinmentakezo07
feat: theme dashboard charts and expose admin UI on HF Spaces
8bdd0d7
Raw
History Blame Contribute Delete
51.4 kB
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