6krrt/tests/test_admin_health.py

"""Tests for the /admin/api health and snapshot endpoints.

``admin.py`` is a standalone FastAPI router that mirrors ``metrics.py``'s
contract — never import ``dispatcher``, take ``(conn, cfg)`` explicitly — and
is mounted onto the dispatcher app under the ``/admin`` prefix. These tests
drive a real TestClient GET against the seeded temp DB (never a mock-call
assertion), mirroring ``test_metrics_endpoint.py``.
"""

from __future__ import annotations

import os
import sqlite3
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest
from starlette.testclient import TestClient

import dispatcher
from config import load_config

ROOT = Path(__file__).resolve().parent.parent
SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
CFG = load_config(str(ROOT / "config" / "config.yaml"))


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _make_db(tmp_path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(str(tmp_path / "test.db"))
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA_SQL)
    return conn


def _seed_models(conn: sqlite3.Connection) -> None:
    for model_id, tier, context, cost, vision in (
        ("cheap", 2, 262128, 0.30, 1),
        ("dear", 2, 262128, 9.00, 0),
        ("tiny", 1, 131072, 0.10, 1),
    ):
        conn.execute(
            """
            INSERT INTO models (
                model_id, provider, base_model_id, tier, context_window,
                effective_context_window, max_output_tokens,
                cost_per_1m_prompt, cost_per_1m_completion,
                supports_vision, supports_json_mode,
                latency_class, reasoning_mode, context_variant,
                access_level, availability, last_updated
            ) VALUES (?, 'neuralwatt', ?, ?, ?, 192500, 16384, ?, ?,
                      ?, 1, 'standard', 'default', 'full', 'public', 'active',
                      '2026-08-22T00:00:00+00:00')
            """,
            (model_id, model_id, tier, context, cost, cost / 3, vision),
        )
    conn.commit()


def _seed_decision(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        INSERT INTO route_decisions (
            observed_at, kind, task_category, task_tier, required_context_tokens,
            confidence, classifier_ms, classification_source, latency_tolerance,
            candidates_considered, selected_model, selected_provider,
            runner_up_models, est_cost_usd, est_proficiency,
            session_key, tools, images, json_mode, streamed,
            flex_preference, flex_swapped, flex_forced
        ) VALUES (?, 'route', 'coding_general', 2, 100, 0.95, 200,
                  'classifier', 'interactive', 5, 'cheap', 'neuralwatt',
                  '[{"model_id":"dear","provider":"neuralwatt"}]',
                  0.001, 0.9, 'abc123', 0, 0, 0, 0,
                  'auto', 0, 1)
        """,
        (_now().isoformat(),),
    )
    conn.commit()


def _seed_energy(conn: sqlite3.Connection) -> None:
    now = _now()
    conn.execute(
        "INSERT INTO energy_observations "
        "(model_id, provider, task_category, completion_tokens, energy_kwh, "
        "cost_usd, carbon_g_co2eq, attribution_ratio, observed_at) "
        "VALUES ('cheap', 'neuralwatt', 'coding_general', 100, 5.0e-05, 0.001, "
        "2.4e-03, 0.25, ?)",
        ((now - timedelta(days=2)).isoformat(),),
    )
    conn.commit()


def _seed_verification(conn: sqlite3.Connection) -> None:
    conn.execute(
        "INSERT INTO verifications (model_id, provider, kind, verdict, observed_at) "
        "VALUES ('cheap', 'neuralwatt', 'structural', 'ok', ?)",
        (_now().isoformat(),),
    )
    conn.commit()


def _seed_proficiency(conn: sqlite3.Connection) -> None:
    conn.execute(
        "INSERT INTO proficiency (model_id, provider, category, blended_score, "
        "source, last_updated) "
        "VALUES ('cheap', 'neuralwatt', 'coding_general', 0.9, "
        "'self_eval_thin', '2026-01-01T00:00:00+00:00')",
    )
    conn.commit()


@pytest.fixture
def seeded_client(tmp_path, monkeypatch):
    """A TestClient wired to a seeded temp DB, at /admin."""
    conn = _make_db(tmp_path)
    _seed_models(conn)
    for _ in range(3):
        _seed_decision(conn)
    _seed_energy(conn)
    _seed_verification(conn)
    _seed_proficiency(conn)
    conn.close()

    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

    with TestClient(dispatcher.app) as client:
        yield client


def test_admin_health_returns_ok(seeded_client):
    """GET /admin/api/health returns 200 with {"status": "ok"}."""
    resp = seeded_client.get("/admin/api/health")
    assert resp.status_code == 200
    assert resp.json() == {"status": "ok"}


def test_admin_snapshot_has_all_top_level_keys(seeded_client):
    """GET /admin/api/snapshot returns 200 with every required key."""
    resp = seeded_client.get("/admin/api/snapshot")
    assert resp.status_code == 200
    data = resp.json()
    for key in (
        "quota",
        "coverage",
        "recent_decisions",
        "per_model",
        "verdict_mix",
        "decision_outcomes",
        "top_proficiency",
        "local_energy",
        "health",
        "pinch",
        "generated_at",
    ):
        assert key in data, f"missing top-level key {key!r}"


def test_admin_snapshot_health_has_expected_shape(seeded_client):
    """Snapshot's health block carries the same shape as /health."""
    data = seeded_client.get("/admin/api/snapshot").json()
    health = data["health"]
    assert "status" in health
    assert "counts" in health
    assert "classifier_reachable" in health and isinstance(
        health["classifier_reachable"], bool
    )
    assert "tiers" in health
    assert "providers" in health
    assert "api_keys_present" in health


def test_admin_snapshot_returns_seeded_data(seeded_client):
    """quota / per_model / verdict_mix / top_proficiency reflect the seed."""
    data = seeded_client.get("/admin/api/snapshot").json()
    assert data["quota"] is not None
    assert isinstance(data["per_model"], list)
    assert any(r["model_id"] == "cheap" for r in data["per_model"])
    assert data["verdict_mix"]["ok"] == 1
    assert isinstance(data["top_proficiency"], list)
    assert data["top_proficiency"][0]["model_id"] == "cheap"
    assert data["health"]["counts"]["models"] == 3


def test_admin_snapshot_contains_no_session_dir(seeded_client):
    """The JSON must never name session_dir or expose conversation text."""
    body = seeded_client.get("/admin/api/snapshot").text
    assert "session_dir" not in body


def test_admin_does_not_import_dispatcher():
    """Importing admin.py must not pull dispatcher into sys.modules."""
    probe = (
        "import admin; import sys; "
        "assert 'dispatcher' not in sys.modules, "
        "'import admin transitively imported dispatcher'"
    )
    subprocess.run(
        [sys.executable, "-c", probe],
        check=True,
        cwd=str(ROOT),
        capture_output=True,
        env={**os.environ, "PYTHONPATH": str(ROOT / "src")},
    )