6krrt/tests/test_admin_classifier_config.py

"""GET/POST /admin/api/classifier-config.

The lesson this reuses rather than relearns (see the profiles zero-admit
fix): the admin surface must report what dispatch actually does, using the
same functions dispatch uses. The GET side's "resolved_primary" for
cloud_primary_auto calls routing.cheapest_classifier_candidate directly --
these tests assert the reported value is provably that function's answer
against a seeded catalog, not a static echo of the config.

POST validates the same way config load does (RouterConfig(**merged) inside
_persist_many_to), so these mirror test_admin_profiles.py's validator tests:
one case proving each invalid combination is rejected, one proving the
matching case is accepted.
"""
from __future__ import annotations

import sqlite3
from pathlib import Path

import pytest
import yaml
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
SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
ADMIN_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS admin_model_overrides (
    model_id         TEXT NOT NULL,
    provider         TEXT NOT NULL,
    availability     TEXT NOT NULL,
    reason           TEXT,
    updated_at       TEXT NOT NULL,
    PRIMARY KEY (model_id, provider)
);
"""


def _model_row(model_id: str, *, tier: int = 1, cost_completion: float = 1.0) -> dict:
    return {
        "model_id": model_id,
        "provider": "neuralwatt",
        "base_model_id": model_id,
        "display_name": model_id,
        "cost_per_1m_prompt": cost_completion * 0.5,
        "cost_per_1m_completion": cost_completion,
        "context_window": 131072,
        "effective_context_window": 65536,
        "max_output_tokens": 8192,
        "tier": tier,
        "supports_tools": 1,
        "supports_json_mode": 1,
        "supports_vision": 1,
        "supports_reasoning": 1,
        "reasoning_default_enabled": 1,
        "latency_class": "standard",
        "reasoning_mode": "default",
        "context_variant": "full",
        "access_level": "public",
        "deprecated": 0,
        "availability": "active",
        "last_updated": "2026-09-01T00:00:00+00:00",
    }


def _seed_models(conn: sqlite3.Connection, rows: list[dict]) -> None:
    cols = list(rows[0].keys())
    placeholders = ",".join(["?"] * len(cols))
    for r in rows:
        conn.execute(
            f"INSERT INTO models ({','.join(cols)}) VALUES ({placeholders})",
            [r[c] for c in cols],
        )
    conn.commit()


def _client(tmp_path, model_rows: list[dict] | None = None, overlay_yaml: str | None = None):
    (tmp_path / "config").mkdir(parents=True, exist_ok=True)
    config_yaml = tmp_path / "config" / "config.yaml"
    local_yaml = tmp_path / "config" / "config.local.yaml"
    config_yaml.write_text((ROOT / "config" / "config.yaml").read_text())
    if overlay_yaml is not None:
        local_yaml.write_text(overlay_yaml)

    cfg = load_config(str(config_yaml))

    conn = sqlite3.connect(str(tmp_path / "admin.db"))
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    if model_rows:
        _seed_models(conn, model_rows)
    conn.close()

    def _db_factory() -> sqlite3.Connection:
        c = sqlite3.connect(str(tmp_path / "admin.db"))
        c.row_factory = sqlite3.Row
        return c

    router = build_router(cfg, _db_factory, base_dir=str(tmp_path))
    app = FastAPI()
    app.include_router(router, prefix="/admin")
    return TestClient(app), config_yaml, local_yaml


# --- GET ---------------------------------------------------------------


def test_get_reports_the_default_mode_sourced_from_base(tmp_path):
    client, _config_yaml, _local_yaml = _client(tmp_path)
    body = client.get("/admin/api/classifier-config").json()
    assert body["mode"] == "local_llm"
    assert body["mode_source"] == "base"
    assert body["resolved_primary"] is None


def test_get_reports_overlay_provenance_after_a_write(tmp_path):
    client, _config_yaml, _local_yaml = _client(
        tmp_path, model_rows=[_model_row("m1")]
    )
    client.post(
        "/admin/api/classifier-config",
        json={"mode": "cloud_llm", "cloud_primary_auto": True},
    )
    body = client.get("/admin/api/classifier-config").json()
    assert body["mode"] == "cloud_llm"
    assert body["mode_source"] == "overlay"


def test_get_resolves_the_live_cheapest_candidate_for_auto_mode(tmp_path):
    """The core claim: resolved_primary is NOT a static echo, it is
    provably routing.cheapest_classifier_candidate's own answer."""
    client, _config_yaml, local_yaml = _client(
        tmp_path,
        model_rows=[
            _model_row("expensive", cost_completion=10.0),
            _model_row("cheap", cost_completion=0.2),
        ],
    )
    local_yaml.write_text(
        "classifier:\n  mode: cloud_llm\n  cloud_primary_auto: true\n"
    )

    body = client.get("/admin/api/classifier-config").json()

    assert body["mode"] == "cloud_llm"
    assert body["cloud_primary_auto"] is True
    assert body["resolved_primary"]["model_id"] == "cheap"


def test_get_reports_no_resolution_when_pinned_not_auto(tmp_path):
    client, _config_yaml, local_yaml = _client(
        tmp_path, model_rows=[_model_row("m1")]
    )
    local_yaml.write_text(
        "classifier:\n"
        "  mode: cloud_llm\n"
        "  cloud_primary:\n"
        "    base_url: https://api.neuralwatt.com/v1\n"
        "    model: deepseek-v4-flash\n"
    )
    body = client.get("/admin/api/classifier-config").json()
    assert body["resolved_primary"] is None
    assert body["cloud_primary"]["model"] == "deepseek-v4-flash"


# --- POST: validated the same way config load is ------------------------


def test_post_cloud_llm_without_either_primary_is_rejected(tmp_path):
    client, _config_yaml, _local_yaml = _client(tmp_path)
    resp = client.post("/admin/api/classifier-config", json={"mode": "cloud_llm"})
    assert resp.status_code == 422
    assert "cloud_primary" in resp.json()["detail"]


def test_post_cloud_llm_with_blank_pinned_fields_is_rejected(tmp_path):
    """A present-but-empty cloud_primary is a truthy dict -- a bare
    `if not body.cloud_primary` would miss this. Found live: the admin form
    sent {"base_url": "", "model": ""} for empty inputs and it silently saved."""
    client, _config_yaml, local_yaml = _client(tmp_path)
    resp = client.post(
        "/admin/api/classifier-config",
        json={
            "mode": "cloud_llm",
            "cloud_primary": {"base_url": "", "model": ""},
        },
    )
    assert resp.status_code == 422
    assert not local_yaml.exists(), "a rejected save must not touch the overlay"


def test_post_cloud_llm_with_only_model_blank_is_rejected(tmp_path):
    client, _config_yaml, _local_yaml = _client(tmp_path)
    resp = client.post(
        "/admin/api/classifier-config",
        json={
            "mode": "cloud_llm",
            "cloud_primary": {"base_url": "https://api.neuralwatt.com/v1", "model": ""},
        },
    )
    assert resp.status_code == 422


def test_post_cloud_llm_pinned_writes_both_keys_atomically(tmp_path):
    client, _config_yaml, local_yaml = _client(tmp_path)
    resp = client.post(
        "/admin/api/classifier-config",
        json={
            "mode": "cloud_llm",
            "cloud_primary": {
                "base_url": "https://api.neuralwatt.com/v1",
                "model": "deepseek-v4-flash",
            },
        },
    )
    assert resp.status_code == 200
    written = yaml.safe_load(local_yaml.read_text())
    assert written["classifier"]["mode"] == "cloud_llm"
    assert written["classifier"]["cloud_primary"]["model"] == "deepseek-v4-flash"
    assert written["classifier"]["cloud_primary_auto"] is False


def test_post_cloud_llm_auto_is_accepted(tmp_path):
    client, _config_yaml, local_yaml = _client(tmp_path)
    resp = client.post(
        "/admin/api/classifier-config",
        json={"mode": "cloud_llm", "cloud_primary_auto": True},
    )
    assert resp.status_code == 200
    written = yaml.safe_load(local_yaml.read_text())
    assert written["classifier"]["cloud_primary_auto"] is True
    assert "cloud_primary" not in written["classifier"]


def test_post_switching_to_auto_clears_a_leftover_pinned_primary(tmp_path):
    """Without this, RouterConfig's own validator would reject the save --
    both cloud_primary and cloud_primary_auto set is exactly what it rejects."""
    client, _config_yaml, local_yaml = _client(tmp_path)
    client.post(
        "/admin/api/classifier-config",
        json={
            "mode": "cloud_llm",
            "cloud_primary": {
                "base_url": "https://api.neuralwatt.com/v1",
                "model": "deepseek-v4-flash",
            },
        },
    )
    resp = client.post(
        "/admin/api/classifier-config",
        json={"mode": "cloud_llm", "cloud_primary_auto": True},
    )
    assert resp.status_code == 200
    written = yaml.safe_load(local_yaml.read_text())
    assert written["classifier"]["cloud_primary_auto"] is True
    assert "cloud_primary" not in written["classifier"]


def test_post_local_encoder_without_encoder_block_still_loads_with_defaults(tmp_path):
    """The body's encoder field is optional; LocalEncoderConfig fills in
    every field, so an empty block is a valid save."""
    client, _config_yaml, local_yaml = _client(tmp_path)
    resp = client.post("/admin/api/classifier-config", json={"mode": "local_encoder"})
    assert resp.status_code == 200
    written = yaml.safe_load(local_yaml.read_text())
    assert written["classifier"]["mode"] == "local_encoder"
    assert written["classifier"]["encoder"] == {}


def test_post_local_encoder_with_bad_device_is_rejected(tmp_path):
    client, _config_yaml, _local_yaml = _client(tmp_path)
    resp = client.post(
        "/admin/api/classifier-config",
        json={"mode": "local_encoder", "encoder": {"device": "tpu"}},
    )
    assert resp.status_code == 422


def test_post_local_llm_is_always_accepted(tmp_path):
    client, _config_yaml, local_yaml = _client(tmp_path)
    resp = client.post("/admin/api/classifier-config", json={"mode": "local_llm"})
    assert resp.status_code == 200
    written = yaml.safe_load(local_yaml.read_text())
    assert written["classifier"]["mode"] == "local_llm"


def test_post_response_names_a_restart_is_required(tmp_path):
    client, _config_yaml, _local_yaml = _client(tmp_path)
    resp = client.post("/admin/api/classifier-config", json={"mode": "local_llm"})
    assert "restart" in resp.json()["message"].lower()