6krrt/tests/test_admin_providers.py

"""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