334 lines
12 KiB
Python
334 lines
12 KiB
Python
"""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"
|
|
|
|
|
|
def test_get_classifier_config_includes_decision_block(tmp_path):
|
|
"""GET /admin/api/classifier-config includes decision block when mode is local_decision."""
|
|
client, _config_yaml, _local_yaml = _client(tmp_path)
|
|
resp = client.post(
|
|
"/admin/api/classifier-config",
|
|
json={"mode": "local_decision", "decision": {"coverage_min": 0.42}},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
resp = client.get("/admin/api/classifier-config")
|
|
body = resp.json()
|
|
assert body["mode"] == "local_decision"
|
|
assert body.get("decision") is not None
|
|
assert body["decision"]["coverage_min"] == 0.42
|
|
|
|
|
|
# --- 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()
|
|
|
|
|
|
def test_post_local_decision_mode_returns_200(tmp_path):
|
|
"""local_decision mode with an empty decision block is accepted."""
|
|
client, _config_yaml, local_yaml = _client(tmp_path)
|
|
resp = client.post(
|
|
"/admin/api/classifier-config",
|
|
json={"mode": "local_decision", "decision": {}},
|
|
)
|
|
assert resp.status_code == 200
|
|
written = yaml.safe_load(local_yaml.read_text())
|
|
assert written["classifier"]["mode"] == "local_decision"
|
|
assert written["classifier"]["decision"] == {}
|
|
|
|
|
|
def test_post_invalid_mode_returns_422(tmp_path):
|
|
"""A mode that is not in the Literal is rejected."""
|
|
client, _config_yaml, _local_yaml = _client(tmp_path)
|
|
resp = client.post(
|
|
"/admin/api/classifier-config",
|
|
json={"mode": "invalid_mode"},
|
|
)
|
|
assert resp.status_code == 422
|