6krrt/tests/test_admin_snapshot.py

"""Tests for GET /admin/api/snapshot: the full top-level key contract.

The snapshot is the admin dashboard's main data payload. This file locks the
complete key set (not just a hand-picked subset) so a dropped or renamed key
fails the suite instead of silently missing from the UI. It drives a real
TestClient against a seeded temp DB, mirroring ``test_admin_health.py``.
"""

from __future__ import annotations

import sqlite3
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"))

EXHAUSTIVE_KEYS = (
    "quota",
    "coverage",
    "recent_decisions",
    "per_model",
    "verdict_mix",
    "top_proficiency",
    "local_energy",
    "health",
    "pinch",
    "generated_at",
)


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),
    ):
        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:
    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()


def _seed_local_energy(conn: sqlite3.Connection) -> None:
    conn.execute(
        "INSERT INTO local_energy_observations "
        "(observed_at, model_id, call_type, avg_power_watts, duration_seconds, "
        "energy_kwh, cost_usd, carbon_g_co2eq, meter) "
        "VALUES (?, 'mistral-nemo:12b', 'classify', 150.0, 1.0, "
        "0.00015, 1.8e-05, 0.06, 'nvidia-smi')",
        (_now().isoformat(),),
    )
    conn.commit()


@pytest.fixture
def seeded_client(tmp_path, monkeypatch):
    conn = _make_db(tmp_path)
    _seed_models(conn)
    for _ in range(3):
        _seed_decision(conn)
    _seed_energy(conn)
    _seed_verification(conn)
    _seed_proficiency(conn)
    _seed_local_energy(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.setattr(dispatcher.cfg.local_energy, "enabled", False)
    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")

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


def test_admin_snapshot_returns_exhaustive_top_level_keys(seeded_client):
    """Every documented top-level snapshot key is present in the response."""
    resp = seeded_client.get("/admin/api/snapshot")
    assert resp.status_code == 200
    data = resp.json()
    for key in EXHAUSTIVE_KEYS:
        assert key in data, f"missing top-level snapshot key {key!r}"
    assert data["local_energy"] is None


def test_admin_snapshot_generated_at_is_iso(seeded_client):
    """generated_at is an ISO-8601 timestamp and freshly generated."""
    data = seeded_client.get("/admin/api/snapshot").json()
    generated = datetime.fromisoformat(data["generated_at"])
    assert generated.tzinfo is not None
    delta = abs((_now() - generated).total_seconds())
    assert delta < 60, "snapshot generated_at is stale"


def test_admin_snapshot_empty_db_still_returns_all_keys(tmp_path, monkeypatch):
    """An unseeded DB returns the same key set, with empty collections."""
    conn = _make_db(tmp_path)
    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:
        data = client.get("/admin/api/snapshot").json()
    for key in EXHAUSTIVE_KEYS:
        assert key in data, f"empty-DB snapshot missing key {key!r}"
    assert data["per_model"] == []
    assert data["recent_decisions"] == []
    assert data["top_proficiency"] == []