mirror of
https://github.com/Alishahryar1/free-claude-code.git
synced 2026-05-20 17:40:50 +00:00
246 lines
7.6 KiB
Python
246 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
from fastapi.testclient import TestClient
|
|
|
|
from api.admin_config import MASKED_SECRET
|
|
from api.admin_urls import local_admin_url
|
|
from api.app import create_app
|
|
from config.settings import Settings
|
|
|
|
|
|
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:
|
|
for key in (
|
|
"MODEL",
|
|
"NVIDIA_NIM_API_KEY",
|
|
"OPENROUTER_API_KEY",
|
|
"ANTHROPIC_AUTH_TOKEN",
|
|
"FCC_ENV_FILE",
|
|
"HOST",
|
|
"PORT",
|
|
"LOG_FILE",
|
|
):
|
|
monkeypatch.delenv(key, raising=False)
|
|
|
|
|
|
def test_admin_page_is_loopback_only(monkeypatch, tmp_path):
|
|
_set_home(monkeypatch, tmp_path)
|
|
app = create_app(lifespan_enabled=False)
|
|
|
|
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_config_masks_secrets_and_exposes_manifest(monkeypatch, tmp_path):
|
|
_set_home(monkeypatch, tmp_path)
|
|
_clear_process_config(monkeypatch)
|
|
app = create_app(lifespan_enabled=False)
|
|
|
|
response = _local_client(app).get("/admin/api/config")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
keys = {field["key"] for field in body["fields"]}
|
|
assert "ANTHROPIC_AUTH_TOKEN" in keys
|
|
assert "OPENROUTER_API_KEY" 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"
|
|
|
|
|
|
def test_admin_validate_rejects_bad_model_shape(monkeypatch, tmp_path):
|
|
_set_home(monkeypatch, tmp_path)
|
|
_clear_process_config(monkeypatch)
|
|
app = create_app(lifespan_enabled=False)
|
|
|
|
response = _local_client(app).post(
|
|
"/admin/api/config/validate",
|
|
json={"values": {"MODEL": "missing-provider-prefix"}},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["valid"] is False
|
|
assert any("provider type" 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_app(lifespan_enabled=False)
|
|
|
|
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 / ".fcc" / ".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_restart_required_reports_automatic_restart(monkeypatch, tmp_path):
|
|
_set_home(monkeypatch, tmp_path)
|
|
_clear_process_config(monkeypatch)
|
|
app = create_app(lifespan_enabled=False)
|
|
callbacks: list[str] = []
|
|
|
|
async def restart_callback() -> None:
|
|
callbacks.append("restart")
|
|
|
|
app.state.admin_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_app(lifespan_enabled=False)
|
|
|
|
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_app(lifespan_enabled=False)
|
|
|
|
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 / ".fcc" / ".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_app(lifespan_enabled=False)
|
|
|
|
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 / ".fcc" / ".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_app(lifespan_enabled=False)
|
|
|
|
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("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"
|