ddeebc7fdd
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
260 lines
8.8 KiB
Python
260 lines
8.8 KiB
Python
"""Tests for the admin provider CRUD endpoints and /admin/providers page route.
|
|
|
|
The page fixture creates a temporary ``admin/frontend/providers.html`` if the
|
|
file does not yet exist, and removes it after the test so no HTML file is left
|
|
in the repo as part of this change.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from starlette.testclient import TestClient
|
|
|
|
from admin import build_router
|
|
from config import load_config
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def _make_db(tmp_path: Path) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(str(tmp_path / "providers.db"))
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path):
|
|
"""A TestClient for an isolated admin router over a temp config.yaml copy."""
|
|
(tmp_path / "config").mkdir(parents=True, exist_ok=True)
|
|
config_yaml = tmp_path / "config" / "config.yaml"
|
|
shutil.copyfile(ROOT / "config" / "config.yaml", config_yaml)
|
|
|
|
schema_sql = (ROOT / "config" / "schema.sql").read_text()
|
|
conn = _make_db(tmp_path)
|
|
conn.executescript(schema_sql)
|
|
conn.close()
|
|
|
|
cfg = load_config(str(config_yaml))
|
|
|
|
def _db_factory() -> sqlite3.Connection:
|
|
return _make_db(tmp_path)
|
|
|
|
router = build_router(cfg, _db_factory, base_dir=str(tmp_path))
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/admin")
|
|
|
|
page = ROOT / "admin" / "frontend" / "providers.html"
|
|
created_page = False
|
|
if not page.exists():
|
|
page.write_text("<html><body>Providers</body></html>")
|
|
created_page = True
|
|
|
|
try:
|
|
yield TestClient(app)
|
|
finally:
|
|
if created_page:
|
|
page.unlink(missing_ok=True)
|
|
|
|
|
|
_VALID_PROVIDER = {
|
|
"base_url": "https://api.example.com/v1",
|
|
"api_key_env": "EXAMPLE_API_KEY",
|
|
"has_energy_telemetry": True,
|
|
"enabled": True,
|
|
}
|
|
|
|
|
|
_SENTINEL = "# SENTINEL_PRESERVED_12345"
|
|
|
|
|
|
def _insert_header_marker(config_yaml: Path) -> None:
|
|
"""Prepend a unique marker comment just above the ``logging:`` map."""
|
|
text = config_yaml.read_text()
|
|
assert "logging:" in text
|
|
config_yaml.write_text(text.replace("logging:", f"{_SENTINEL}\nlogging:", 1))
|
|
|
|
|
|
def test_providers_page_returns_html(client):
|
|
"""GET /admin/providers serves the providers HTML page."""
|
|
resp = client.get("/admin/providers")
|
|
assert resp.status_code == 200
|
|
assert resp.headers["content-type"].startswith("text/html")
|
|
assert "Providers" in resp.text
|
|
|
|
|
|
def test_provider_crud_round_trip(client, tmp_path):
|
|
"""Create, read, and delete a provider through the overlay."""
|
|
local_yaml = tmp_path / "config" / "config.local.yaml"
|
|
|
|
resp = client.post("/admin/api/providers/testprov", json=_VALID_PROVIDER)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["provider"]["name"] == "testprov"
|
|
assert body["provider"]["base_url"] == _VALID_PROVIDER["base_url"]
|
|
|
|
assert local_yaml.exists()
|
|
text = local_yaml.read_text()
|
|
assert "testprov:" in text
|
|
assert "dispatch_providers:" in text
|
|
|
|
resp = client.get("/admin/api/providers/testprov")
|
|
assert resp.status_code == 200
|
|
detail = resp.json()
|
|
assert detail["name"] == "testprov"
|
|
assert detail["base_url"] == _VALID_PROVIDER["base_url"]
|
|
assert detail["api_key_env"] == _VALID_PROVIDER["api_key_env"]
|
|
assert detail["has_energy_telemetry"] is True
|
|
assert detail["enabled"] is True
|
|
|
|
resp = client.get("/admin/api/providers")
|
|
assert resp.status_code == 200
|
|
names = [p["name"] for p in resp.json()]
|
|
assert "testprov" in names
|
|
|
|
resp = client.delete("/admin/api/providers/testprov")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["deleted"] == "testprov"
|
|
|
|
resp = client.get("/admin/api/providers/testprov")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_provider_post_invalid_url_returns_422(client, tmp_path):
|
|
"""A non-URL base_url is rejected without writing to the overlay."""
|
|
local_yaml = tmp_path / "config" / "config.local.yaml"
|
|
resp = client.post(
|
|
"/admin/api/providers/badurl",
|
|
json={
|
|
"base_url": "not-a-url",
|
|
"api_key_env": "BAD_API_KEY",
|
|
"has_energy_telemetry": False,
|
|
"enabled": True,
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
assert not local_yaml.exists()
|
|
|
|
|
|
def test_provider_delete_default_returns_422_and_leaves_config(
|
|
client, tmp_path
|
|
):
|
|
"""Deleting dispatch_settings.default_provider is rejected and config is untouched."""
|
|
config_yaml = tmp_path / "config" / "config.yaml"
|
|
local_yaml = tmp_path / "config" / "config.local.yaml"
|
|
base_before = config_yaml.read_bytes()
|
|
local_before = local_yaml.read_bytes() if local_yaml.exists() else b""
|
|
|
|
resp = client.delete("/admin/api/providers/neuralwatt")
|
|
assert resp.status_code == 422
|
|
detail = resp.json()["detail"]
|
|
assert "neuralwatt" in detail
|
|
assert "default_provider" in detail
|
|
|
|
assert config_yaml.read_bytes() == base_before
|
|
assert not local_yaml.exists()
|
|
assert local_before == b""
|
|
|
|
|
|
def test_provider_delete_base_configured_returns_403_and_explains_provenance(
|
|
client, tmp_path
|
|
):
|
|
"""Deleting a base-configured non-default provider returns 403 naming config/config.yaml."""
|
|
config_yaml = tmp_path / "config" / "config.yaml"
|
|
base_before = config_yaml.read_bytes()
|
|
|
|
resp = client.delete("/admin/api/providers/openrouter")
|
|
assert resp.status_code == 403
|
|
detail = resp.json()["detail"]
|
|
assert "openrouter" in detail
|
|
assert "config/config.yaml" in detail
|
|
assert "read-only" in detail.lower()
|
|
|
|
assert config_yaml.read_bytes() == base_before
|
|
|
|
|
|
def test_provider_delete_unknown_returns_404(client, tmp_path):
|
|
"""Deleting a provider that does not exist returns 404."""
|
|
config_yaml = tmp_path / "config" / "config.yaml"
|
|
base_before = config_yaml.read_bytes()
|
|
|
|
resp = client.delete("/admin/api/providers/nonexistent")
|
|
assert resp.status_code == 404
|
|
assert "provider not found" in resp.json()["detail"]
|
|
|
|
assert config_yaml.read_bytes() == base_before
|
|
|
|
|
|
def test_provider_post_writes_overlay_and_preserves_base_backup_created(
|
|
client, tmp_path
|
|
):
|
|
"""POSTing a valid provider writes only the overlay, leaves base alone, and backs up."""
|
|
config_yaml = tmp_path / "config" / "config.yaml"
|
|
local_yaml = tmp_path / "config" / "config.local.yaml"
|
|
_insert_header_marker(config_yaml)
|
|
base_before = config_yaml.read_bytes()
|
|
|
|
resp = client.post("/admin/api/providers/testprov", json=_VALID_PROVIDER)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["provider"]["name"] == "testprov"
|
|
assert body["provider"]["base_url"] == _VALID_PROVIDER["base_url"]
|
|
|
|
assert config_yaml.read_bytes() == base_before
|
|
assert local_yaml.exists()
|
|
local_text = local_yaml.read_text()
|
|
assert "dispatch_providers:" in local_text
|
|
assert "testprov:" in local_text
|
|
assert "base_url: https://api.example.com/v1" in local_text
|
|
|
|
backups = sorted(tmp_path.glob("config/config.local.yaml.bak.*"))
|
|
assert len(backups) == 1
|
|
assert backups[0].read_text().strip() != ""
|
|
|
|
|
|
def test_provider_list_includes_base_and_overlay_providers(client, tmp_path):
|
|
"""GET /admin/api/providers includes both base-configured and overlay providers."""
|
|
config_yaml = tmp_path / "config" / "config.yaml"
|
|
local_yaml = tmp_path / "config" / "config.local.yaml"
|
|
|
|
local_yaml.write_text(
|
|
"# Machine-local overlay -- test fixture\n"
|
|
"dispatch_providers:\n"
|
|
" overlayprov:\n"
|
|
" base_url: https://overlay.example.com/v1\n"
|
|
" api_key_env: OVERLAY_API_KEY\n"
|
|
" has_energy_telemetry: false\n"
|
|
" enabled: true\n"
|
|
)
|
|
|
|
resp = client.get("/admin/api/providers")
|
|
assert resp.status_code == 200
|
|
names = {p["name"] for p in resp.json()}
|
|
assert names >= {"neuralwatt", "openrouter", "overlayprov"}
|
|
|
|
config_before = config_yaml.read_bytes()
|
|
assert config_yaml.read_bytes() == config_before
|
|
|
|
by_name = {p["name"]: p for p in resp.json()}
|
|
assert by_name["neuralwatt"]["base_url"] is not None
|
|
assert by_name["openrouter"]["base_url"] is not None
|
|
assert by_name["overlayprov"]["base_url"] == "https://overlay.example.com/v1"
|
|
|
|
assert by_name["neuralwatt"]["in_base"] is True
|
|
assert by_name["neuralwatt"]["in_overlay"] is False
|
|
assert by_name["openrouter"]["in_base"] is True
|
|
assert by_name["openrouter"]["in_overlay"] is False
|
|
assert by_name["overlayprov"]["in_base"] is False
|
|
assert by_name["overlayprov"]["in_overlay"] is True
|
|
|
|
detail = client.get("/admin/api/providers/neuralwatt").json()
|
|
assert detail["in_base"] is True
|
|
assert detail["in_overlay"] is False
|
|
detail = client.get("/admin/api/providers/overlayprov").json()
|
|
assert detail["in_base"] is False
|
|
assert detail["in_overlay"] is True
|