"""Tests for the GET /metrics endpoint on dispatcher. Seeds a throwaway temp DB and asserts on the actual JSON returned by a real TestClient GET (never a mock-call assertion), to defeat ``misleading_success_output``. Also covers the ``/events/decisions`` Server-Sent Events endpoint, which streams routing decisions to the TUI without polling. """ from __future__ import annotations import json import sqlite3 from datetime import datetime, timedelta, timezone from pathlib import Path import pytest from starlette.testclient import TestClient import dispatcher import events 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, extra_sql: str = "") -> sqlite3.Connection: conn = sqlite3.connect(str(tmp_path / "test.db")) conn.row_factory = sqlite3.Row conn.executescript(SCHEMA_SQL + extra_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() 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): """A TestClient wired to a seeded temp DB, at /metrics.""" conn = _make_db(tmp_path) _seed_models(conn) for i in range(60): # exceed the 50 cap _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.setattr(dispatcher.cfg.local_energy, "enabled", False) monkeypatch.setenv("NEURALWATT_API_KEY", "test-key") with TestClient(dispatcher.app) as client: yield client def test_metrics_endpoint_has_all_top_level_keys(seeded_client): """GET /metrics returns 200 with every required top-level key.""" resp = seeded_client.get("/metrics") assert resp.status_code == 200 data = resp.json() for key in ( "quota", "coverage", "recent_decisions", "per_model", "verdict_mix", "top_proficiency", "local_energy", "pinch", "cost_calibration", "latency", "generated_at", ): assert key in data, f"missing top-level key {key!r}" def test_metrics_recent_decisions_capped_and_carries_provider(seeded_client): """recent_decisions is capped at 50 and each row has selected_provider.""" resp = seeded_client.get("/metrics") data = resp.json() decisions = data["recent_decisions"] assert isinstance(decisions, list) assert len(decisions) <= 50 assert len(decisions) > 0 for row in decisions: assert "selected_provider" in row # DESC order: the first decision has the largest id. first_id = decisions[0]["id"] for row in decisions[1:]: assert row["id"] <= first_id def test_metrics_aggregations_are_populated(seeded_client): """per_model / verdict_mix / top_proficiency reflect seeded data.""" resp = seeded_client.get("/metrics") data = resp.json() assert isinstance(data["per_model"], list) assert any(r["model_id"] == "cheap" for r in data["per_model"]) assert data["per_model"][0]["calls"] == 1 assert data["verdict_mix"]["ok"] == 1 assert isinstance(data["top_proficiency"], list) assert data["top_proficiency"][0]["model_id"] == "cheap" assert data["quota"] is not None def test_metrics_recent_decisions_carry_flex_columns(seeded_client): """Each recent_decisions row carries the three flex telemetry columns.""" resp = seeded_client.get("/metrics") data = resp.json() decisions = data["recent_decisions"] assert decisions for row in decisions: assert "flex_preference" in row assert "flex_swapped" in row assert "flex_forced" in row assert decisions[0]["flex_preference"] == "auto" assert decisions[0]["flex_swapped"] == 0 assert decisions[0]["flex_forced"] == 1 def test_metrics_exposes_flex_default(seeded_client): """The configured flex default is reported (under coverage).""" resp = seeded_client.get("/metrics") data = resp.json() assert "flex_default" in data["coverage"] assert data["coverage"]["flex_default"] in ("no-flex", "auto", "prefer-flex", "force-flex") def test_metrics_contains_no_session_dir(seeded_client): """The JSON must never name session_dir or expose conversation text.""" body = seeded_client.get("/metrics").text assert "session_dir" not in body def test_metrics_quota_carries_accounts_shape(seeded_client): """GET /metrics quota carries period/accounts/spend/alarm blocks.""" resp = seeded_client.get("/metrics") assert resp.status_code == 200 data = resp.json() assert data["quota"] is not None quota = data["quota"] for key in ("period", "accounts", "spend", "alarm"): assert key in quota, f"missing quota key {key!r}" period = quota["period"] for key in ("start", "next_reset", "elapsed_fraction", "source"): assert key in period, f"missing period key {key!r}" accounts = quota["accounts"] assert isinstance(accounts, list) for acc in accounts: for key in ("provider", "shape", "spend_usd"): assert key in acc, f"missing account key {key!r}" assert acc["shape"] in ("metered_plan", "prepaid_credit", "self_hosted", "unmetered") spend = quota["spend"] for key in ("by_provider_usd", "total_usd", "estimated_usd", "estimate_ratio"): assert key in spend, f"missing spend key {key!r}" def test_metrics_quota_survives_missing_provider_balance_table(tmp_path, monkeypatch): """A live DB predating provider_balance_observations must not 500 /metrics.""" conn = _make_db(tmp_path) _seed_models(conn) _seed_decision(conn) conn.execute("DROP TABLE provider_balance_observations") conn.commit() 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: resp = client.get("/metrics") assert resp.status_code == 200 quota = resp.json()["quota"] assert "accounts" in quota for acc in quota["accounts"]: # Should not crash: missing table means pool is None/burn is None assert acc["shape"] in ("metered_plan", "prepaid_credit", "self_hosted", "unmetered") def test_metrics_local_energy_omitted_when_disabled(seeded_client): """With metering off the section is omitted — independent of deployment config.""" resp = seeded_client.get("/metrics") data = resp.json() assert "local_energy" in data assert data["local_energy"] is None @pytest.mark.parametrize("local_energy_enabled", [False, True]) def test_metrics_local_energy_follows_enabled_flag( tmp_path, monkeypatch, local_energy_enabled ): """The suite must not couple to deployment config: both flag values are exercised.""" conn = _make_db(tmp_path) _seed_models(conn) if local_energy_enabled: _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", local_energy_enabled ) monkeypatch.setattr(dispatcher.cfg.local_energy, "tariff_usd_per_kwh", 0.12) monkeypatch.setenv("NEURALWATT_API_KEY", "test-key") with TestClient(dispatcher.app) as client: resp = client.get("/metrics") assert resp.status_code == 200 data = resp.json() if local_energy_enabled: assert isinstance(data["local_energy"], dict) assert data["local_energy"]["calls_30d"] == 1 assert "by_type" in data["local_energy"] else: assert data["local_energy"] is None def test_metrics_local_energy_enabled_reflects_seeded_row( tmp_path, monkeypatch ): """When enabled and a row is seeded, local_energy carries expected totals.""" conn = _make_db(tmp_path) _seed_models(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", True) monkeypatch.setattr(dispatcher.cfg.local_energy, "tariff_usd_per_kwh", 0.12) monkeypatch.setenv("NEURALWATT_API_KEY", "test-key") with TestClient(dispatcher.app) as client: resp = client.get("/metrics") assert resp.status_code == 200 data = resp.json() le = data["local_energy"] assert isinstance(le, dict) assert "metered_kwh_30d" in le assert "metered_cost_usd_30d" in le assert "calls_30d" in le assert "by_type" in le assert "reset_date" in le assert le["metered_kwh_30d"] == round(0.00015, 5) assert le["metered_cost_usd_30d"] == round(1.8e-05, 5) assert le["calls_30d"] == 1 assert "classify" in le["by_type"] assert le["by_type"]["classify"]["calls"] == 1 assert le["by_type"]["classify"]["kwh"] == round(0.00015, 5) assert le["by_type"]["classify"]["cost_usd"] == round(1.8e-05, 5) def test_metrics_empty_db_returns_200(monkeypatch, tmp_path): """Fresh empty temp DB: 200 with empty arrays, no exception.""" 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.setattr(dispatcher.cfg.local_energy, "enabled", False) monkeypatch.setenv("NEURALWATT_API_KEY", "test-key") with TestClient(dispatcher.app) as client: resp = client.get("/metrics") assert resp.status_code == 200 data = resp.json() assert data["recent_decisions"] == [] assert data["per_model"] == [] assert data["verdict_mix"] == {} assert data["top_proficiency"] == [] assert "local_energy" in data assert data["local_energy"] is None assert "generated_at" in data # admin_model_overrides is NOT in schema.sql — dispatcher creates it at # startup via ensure_route_decisions-style migration, so endpoint tests that # need it build it with extra_sql (pattern from tests/test_metrics.py). _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) ); CREATE INDEX IF NOT EXISTS idx_admin_model_overrides_availability ON admin_model_overrides (availability); """ def test_metrics_coverage_warnings_include_rejection_and_capability( tmp_path, monkeypatch ): """GET /metrics surfaces the new rejection and capability warnings. Seeds the 2026-09-04 incident state — kimi-k3 and kimi-k3-fast (the only vision-capable tier-1 models) deprecated via admin overrides, plus two image rejections — into the endpoint's temp DB, and asserts the warnings reach the same coverage.warnings list the admin bell and TUI render. The rejection warning is the deterministic assertion; on this exact seed the capability warning fires too. """ now = _now().isoformat() conn = _make_db(tmp_path, extra_sql=_ADMIN_TABLE_SQL) for model_id, vision, eff_ctx in ( ("kimi-k3", 1, 782324), ("kimi-k3-fast", 1, 782324), ("deepseek-v4-flash", 0, 262128), ): 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', ?, 1, ?, ?, 16384, 0.30, 0.10, ?, 1, 'standard', 'default', 'full', 'public', 'active', ?) """, (model_id, model_id, eff_ctx, eff_ctx, vision, now), ) for model_id in ("kimi-k3", "kimi-k3-fast"): conn.execute( "INSERT INTO admin_model_overrides " "(model_id, provider, availability, reason, updated_at) " "VALUES (?, 'neuralwatt', 'deprecated', 'cost', ?)", (model_id, now), ) for tokens in (242486, 195000): conn.execute( """ INSERT INTO route_decisions ( observed_at, kind, task_category, task_tier, required_context_tokens, latency_tolerance, selected_model, selected_provider, rejected_reason, tools, images, json_mode, streamed ) VALUES (?, 'chat', 'coding_general', 1, 200000, 'interactive', NULL, NULL, ?, 0, 1, 0, 0) """, ( now, f"context >= {tokens} tokens; " f"vision-capable model (request carries image(s))", ), ) conn.commit() 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: resp = client.get("/metrics") assert resp.status_code == 200 data = resp.json() warnings = data["coverage"]["warnings"] assert len(warnings) > 0 assert any("rejection" in w for w in warnings) # The capability warning is secondary per the plan (it depends on the # seeded models) but this seed reproduces the incident, so it fires. assert any("vision" in w for w in warnings) def _sse_frame(line: str | bytes) -> dict: """Parse one ``data: `` SSE line and return the JSON payload.""" if isinstance(line, bytes): line = line.decode() assert line.startswith("data: "), f"unexpected SSE frame: {line!r}" return json.loads(line[len("data: ") :]) def test_events_decisions_returns_sse_headers(seeded_client, monkeypatch): """The endpoint announces text/event-stream and no-cache headers. The real stream is unbounded (it blocks for heartbeats), so the generator is stubbed to a bounded one to let TestClient read the whole body without hanging. The live replay/stream behaviour is covered by the deterministic generator test below. """ events.clear() try: def _bounded(): yield "retry: 3000\n\n" yield "data: {\"id\": 9}\n\n" monkeypatch.setattr(dispatcher, "_decision_event_stream", _bounded) resp = seeded_client.get("/events/decisions") assert resp.status_code == 200 assert "text/event-stream" in resp.headers["content-type"] assert "no-cache" in resp.headers.get("cache-control", "") assert "data: {\"id\": 9}" in resp.text finally: events.clear() @pytest.mark.anyio async def test_decision_event_stream_replays_then_streams_live(): """Drive the generator directly: replay first, then a live publish arrives as the next frame. No HTTP client or threads, so it is deterministic.""" events.clear() try: events.publish_decision( {"id": 1, "selected_model": "cheap", "task_category": "coding"} ) stream = dispatcher._decision_event_stream() retry = await anext(stream) assert retry.startswith("retry:") replayed = await anext(stream) assert _sse_frame(replayed)["id"] == 1 events.publish_decision( {"id": 2, "selected_model": "tiny", "task_category": "debugging"} ) live = await anext(stream) assert _sse_frame(live)["id"] == 2 finally: events.clear() @pytest.mark.anyio async def test_persist_route_decision_publishes_profile_in_sse_payload( tmp_path, monkeypatch, ): import asyncio events.clear() db_path = tmp_path / "profile-sse.db" conn = sqlite3.connect(db_path) conn.executescript(SCHEMA_SQL) conn.close() monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path)) monkeypatch.setattr(dispatcher.cfg.logging, "log_route_decisions", True) sse_queue: asyncio.Queue[dict] = asyncio.Queue() try: events.subscribe_sse(sse_queue, replay=True) dispatcher.persist_route_decision( "route", classification=dispatcher.RouteResponse( classification=dispatcher.Classification( task_category="coding_general", task_tier=2, required_context_tokens=100, confidence=0.9, ), latency_tolerance="interactive", profile="batch", selected=dispatcher.Candidate( model_id="cheap", provider="neuralwatt", tier=2, latency_class="standard", reasoning_mode="default", context_variant="full", effective_context_window=128000, composite=0.5, cost_score=0.5, proficiency_score=0.9, ), candidates_considered=1, ), latency_tolerance="interactive", selected_model="cheap", selected_provider="neuralwatt", ) decision = await asyncio.wait_for(sse_queue.get(), timeout=2.0) assert decision["profile"] == "batch" finally: events.unsubscribe_sse(sse_queue) events.clear() # --------------------------------------------------------------------------- # content_fault_warnings unit tests # --------------------------------------------------------------------------- from types import SimpleNamespace as _SN # noqa: E402 def _cfw_cfg(): return _SN( objective=_SN( rejection_warning_window_hours=1, rejection_warning_baseline_hours=24, rejection_warning_min_count=6, ) ) def _cfw_conn(tmp_path): conn = sqlite3.connect(str(tmp_path / "cfw_test.db")) conn.row_factory = sqlite3.Row conn.execute(""" CREATE TABLE verifications ( id INTEGER PRIMARY KEY AUTOINCREMENT, model_id TEXT NOT NULL, provider TEXT NOT NULL, request_id TEXT, task_category TEXT, kind TEXT NOT NULL, verdict TEXT NOT NULL, detail TEXT, completion_tokens INTEGER, observed_at TEXT NOT NULL, model_attributable INTEGER DEFAULT 1 ) """) conn.commit() return conn def _cfw_insert(conn, model, provider, observed_at, attributable=1): conn.execute( "INSERT INTO verifications " "(model_id, provider, task_category, request_id, kind, verdict, " " detail, completion_tokens, model_attributable, observed_at) " "VALUES (?, ?, 'coding_general', ?, 'structural', 'malformed', " " 'empty response', 0, ?, ?)", (model, provider, f"req-{model}", attributable, observed_at), ) def test_content_fault_no_malformed_returns_empty(tmp_path): from metrics import content_fault_warnings conn = _cfw_conn(tmp_path) try: assert content_fault_warnings(conn, _cfw_cfg()) == [] finally: conn.close() def test_content_fault_high_rate_triggers_known_warning(tmp_path): from metrics import content_fault_warnings conn = _cfw_conn(tmp_path) try: now = datetime.now(timezone.utc) for i in range(7): _cfw_insert(conn, "bad-model", "openrouter", (now - timedelta(minutes=i)).isoformat()) conn.commit() warnings = content_fault_warnings(conn, _cfw_cfg()) assert len(warnings) >= 1 assert any("bad-model" in w for w in warnings) assert any("content fault:" in w for w in warnings) finally: conn.close() def test_content_fault_novel_pattern_triggers_new_warning(tmp_path): from metrics import content_fault_warnings conn = _cfw_conn(tmp_path) try: now = datetime.now(timezone.utc) for i in range(3): _cfw_insert(conn, "novel-model", "openrouter", (now - timedelta(minutes=i)).isoformat()) conn.commit() warnings = content_fault_warnings(conn, _cfw_cfg()) assert len(warnings) >= 1 assert any("new content fault:" in w for w in warnings) finally: conn.close() def test_content_fault_low_count_does_not_trigger(tmp_path): from metrics import content_fault_warnings conn = _cfw_conn(tmp_path) try: now = datetime.now(timezone.utc) _cfw_insert(conn, "low-model", "openrouter", (now - timedelta(hours=20)).isoformat()) for i in range(3): _cfw_insert(conn, "low-model", "openrouter", (now - timedelta(minutes=i)).isoformat()) conn.commit() assert content_fault_warnings(conn, _cfw_cfg()) == [] finally: conn.close() def test_content_fault_non_attributable_excluded(tmp_path): from metrics import content_fault_warnings conn = _cfw_conn(tmp_path) try: now = datetime.now(timezone.utc) for i in range(10): _cfw_insert(conn, "tool-model", "openrouter", (now - timedelta(minutes=i)).isoformat(), attributable=0) conn.commit() assert content_fault_warnings(conn, _cfw_cfg()) == [] finally: conn.close()