264 lines
9.1 KiB
Python
264 lines
9.1 KiB
Python
"""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 / "schema.sql").read_text()
|
|
CFG = load_config(str(ROOT / "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
|
|
) 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)
|
|
""",
|
|
(_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 /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.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",
|
|
"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_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_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.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 "generated_at" in data
|
|
|
|
|
|
def _sse_frame(line: str | bytes) -> dict:
|
|
"""Parse one ``data: <json>`` 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()
|
|
|
|
|
|
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 = next(stream)
|
|
assert retry.startswith("retry:")
|
|
replayed = next(stream)
|
|
assert _sse_frame(replayed)["id"] == 1
|
|
|
|
events.publish_decision(
|
|
{"id": 2, "selected_model": "tiny", "task_category": "debugging"}
|
|
)
|
|
live = next(stream)
|
|
assert _sse_frame(live)["id"] == 2
|
|
stream.close()
|
|
finally:
|
|
events.clear()
|