250 lines
9.0 KiB
Python
250 lines
9.0 KiB
Python
"""Tests for the /admin/api/history bucketed time-series endpoint.
|
|
|
|
``/api/history`` runs a bucketed GROUP BY over two existing tables
|
|
(``energy_observations`` and ``route_decisions``), keyed by a ``range`` query
|
|
param that selects a (span_seconds, bucket_seconds) pair. These tests drive a
|
|
real TestClient GET against a temp DB seeded at controlled timestamps, exactly
|
|
like ``test_admin_health.py`` — never a mock-call assertion.
|
|
"""
|
|
|
|
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"))
|
|
|
|
# The 1h bucket size used by range=24h; expected bucket ts are whole-second
|
|
# hour boundaries, computed the same way the SQL floors to a bucket index.
|
|
BUCKET_24H = 3600
|
|
|
|
|
|
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_energy(
|
|
conn: sqlite3.Connection,
|
|
at: datetime,
|
|
energy_kwh: float,
|
|
cost_usd: float,
|
|
carbon_g_co2eq: float,
|
|
) -> 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, ?, ?, ?, 0.25, ?)",
|
|
(energy_kwh, cost_usd, carbon_g_co2eq, at.isoformat()),
|
|
)
|
|
|
|
|
|
def _seed_decision(conn: sqlite3.Connection, at: datetime) -> 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)
|
|
""",
|
|
(at.isoformat(),),
|
|
)
|
|
|
|
|
|
def _hour_floor(dt: datetime) -> datetime:
|
|
"""The start of the hour containing *dt*, on a whole second."""
|
|
return dt.replace(second=0, microsecond=0).replace(minute=0)
|
|
|
|
|
|
@pytest.fixture
|
|
def seeded_client(tmp_path, monkeypatch):
|
|
"""A TestClient at /admin wired to a temp DB.
|
|
|
|
Seeds, relative to the current hour, three energy rows across two
|
|
one-hour buckets and three decisions across the same two buckets — all
|
|
inside a 24h window:
|
|
|
|
- bucket A (current hour): R1 energy=0.001 cost=0.01 carbon=0.5,
|
|
R2 energy=0.002 cost=0.02 carbon=1.0
|
|
- bucket B (previous hour): R3 energy=0.004 cost=0.04 carbon=2.0
|
|
- decisions: D1+D2 in bucket A, D3 in bucket B
|
|
"""
|
|
conn = _make_db(tmp_path)
|
|
_seed_models(conn)
|
|
|
|
now = _now()
|
|
bucket_a = _hour_floor(now)
|
|
bucket_b = bucket_a - timedelta(hours=1)
|
|
|
|
_seed_energy(conn, bucket_a + timedelta(seconds=60), 0.001, 0.01, 0.5)
|
|
_seed_energy(conn, bucket_a + timedelta(seconds=120), 0.002, 0.02, 1.0)
|
|
_seed_energy(conn, bucket_b + timedelta(seconds=30), 0.004, 0.04, 2.0)
|
|
|
|
_seed_decision(conn, bucket_a + timedelta(seconds=60))
|
|
_seed_decision(conn, bucket_a + timedelta(seconds=200))
|
|
_seed_decision(conn, bucket_b + timedelta(seconds=30))
|
|
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.setenv("NEURALWATT_API_KEY", "test-key")
|
|
|
|
with TestClient(dispatcher.app) as client:
|
|
yield client
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _bucket_ts(dt: datetime) -> int:
|
|
"""The unix second of the 1h bucket containing *dt*."""
|
|
return int(dt.timestamp() // BUCKET_24H * BUCKET_24H)
|
|
|
|
|
|
def test_admin_history_24h_returns_series(seeded_client):
|
|
"""range=24h returns all five series, each a list of [ts, value] pairs."""
|
|
resp = seeded_client.get("/admin/api/history", params={"range": "24h"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert set(data) == {
|
|
"decisions_per_bucket",
|
|
"requests_per_bucket",
|
|
"cost_per_bucket",
|
|
"energy_per_bucket",
|
|
"carbon_per_bucket",
|
|
}
|
|
for series in data.values():
|
|
assert isinstance(series, list)
|
|
for point in series:
|
|
assert len(point) == 2
|
|
assert isinstance(point[0], (int, float))
|
|
|
|
|
|
def test_admin_history_24h_decisions_per_bucket(seeded_client):
|
|
"""decisions_per_bucket counts route_decisions per 1h bucket."""
|
|
now = _now()
|
|
bucket_a = _bucket_ts(_hour_floor(now))
|
|
bucket_b = bucket_a - BUCKET_24H
|
|
series = seeded_client.get("/admin/api/history", params={"range": "24h"}).json()[
|
|
"decisions_per_bucket"
|
|
]
|
|
assert series == [[bucket_b, 1], [bucket_a, 2]]
|
|
|
|
|
|
def test_admin_history_24h_requests_per_bucket(seeded_client):
|
|
"""requests_per_bucket counts energy_observations per 1h bucket."""
|
|
now = _now()
|
|
bucket_a = _bucket_ts(_hour_floor(now))
|
|
bucket_b = bucket_a - BUCKET_24H
|
|
series = seeded_client.get("/admin/api/history", params={"range": "24h"}).json()[
|
|
"requests_per_bucket"
|
|
]
|
|
assert series == [[bucket_b, 1], [bucket_a, 2]]
|
|
|
|
|
|
def test_admin_history_24h_cost_per_bucket(seeded_client):
|
|
"""cost_per_bucket sums cost_usd per 1h bucket."""
|
|
now = _now()
|
|
bucket_a = _bucket_ts(_hour_floor(now))
|
|
bucket_b = bucket_a - BUCKET_24H
|
|
series = seeded_client.get("/admin/api/history", params={"range": "24h"}).json()[
|
|
"cost_per_bucket"
|
|
]
|
|
assert series == [[bucket_b, 0.04], [bucket_a, 0.03]]
|
|
|
|
|
|
def test_admin_history_24h_energy_per_bucket(seeded_client):
|
|
"""energy_per_bucket sums energy_kwh per 1h bucket."""
|
|
now = _now()
|
|
bucket_a = _bucket_ts(_hour_floor(now))
|
|
bucket_b = bucket_a - BUCKET_24H
|
|
series = seeded_client.get("/admin/api/history", params={"range": "24h"}).json()[
|
|
"energy_per_bucket"
|
|
]
|
|
assert series == [[bucket_b, 0.004], [bucket_a, 0.003]]
|
|
|
|
|
|
def test_admin_history_24h_carbon_per_bucket(seeded_client):
|
|
"""carbon_per_bucket sums carbon_g_co2eq per 1h bucket."""
|
|
now = _now()
|
|
bucket_a = _bucket_ts(_hour_floor(now))
|
|
bucket_b = bucket_a - BUCKET_24H
|
|
series = seeded_client.get("/admin/api/history", params={"range": "24h"}).json()[
|
|
"carbon_per_bucket"
|
|
]
|
|
assert series == [[bucket_b, 2.0], [bucket_a, 1.5]]
|
|
|
|
|
|
def test_admin_history_bogus_range_returns_400(seeded_client):
|
|
"""An unlisted range value is rejected with 400."""
|
|
resp = seeded_client.get("/admin/api/history", params={"range": "bogus"})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.fixture
|
|
def empty_client(tmp_path, monkeypatch):
|
|
"""A TestClient at /admin wired to an empty (schema-only) temp DB."""
|
|
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:
|
|
yield client
|
|
|
|
|
|
def test_admin_history_empty_tables_return_empty_series(empty_client):
|
|
"""Empty tables yield 200 with every series as an empty list."""
|
|
resp = empty_client.get("/admin/api/history", params={"range": "24h"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
for series in data.values():
|
|
assert series == []
|