87 lines
No EOL
2.9 KiB
Python
87 lines
No EOL
2.9 KiB
Python
"""GET /api/stats HUD counter contract (counts only, small, never 500)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
from datetime import timezone
|
|
|
|
import httpx
|
|
|
|
from main import app, _stats_counts
|
|
|
|
BASE = "http://test"
|
|
|
|
EXPECTED_KEYS = ("aircraft", "vessels", "trains", "cameras",
|
|
"fires", "quakes", "alerts", "timestamp")
|
|
|
|
|
|
async def _get(path: str) -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
|
|
return await client.get(path)
|
|
|
|
|
|
def test_stats_200_all_keys_present():
|
|
resp = asyncio.run(_get("/api/stats"))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
for key in EXPECTED_KEYS:
|
|
assert key in body, f"missing key {key}"
|
|
assert "max-age" in (resp.headers.get("cache-control") or "").lower()
|
|
|
|
|
|
def test_stats_counters_are_ints():
|
|
body = asyncio.run(_get("/api/stats")).json()
|
|
for key in EXPECTED_KEYS:
|
|
if key == "timestamp":
|
|
continue
|
|
assert isinstance(body[key], int), f"{key} is not an int: {body[key]!r}"
|
|
|
|
|
|
def test_stats_timestamp_is_iso8601_z():
|
|
body = asyncio.run(_get("/api/stats")).json()
|
|
ts = body["timestamp"]
|
|
# ISO8601 with a trailing Z (we normalize +00:00 -> Z).
|
|
assert isinstance(ts, str) and ts.endswith("Z")
|
|
assert re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}", ts)
|
|
|
|
|
|
def test_stats_payload_is_tiny():
|
|
resp = asyncio.run(_get("/api/stats"))
|
|
assert len(resp.content) < 2048, "stats payload must be counts-only, not GeoJSON"
|
|
|
|
|
|
def test_stats_counts_reflect_last_known(monkeypatch):
|
|
"""aircraft/vessels/trains/alerts come from in-memory last-known state."""
|
|
import live_layers
|
|
|
|
monkeypatch.setattr(live_layers, "aircraft_last_known", {str(i): {} for i in range(7)})
|
|
monkeypatch.setattr(live_layers, "vessel_last_known", {str(i): {} for i in range(3)})
|
|
monkeypatch.setattr(live_layers, "train_count", 11)
|
|
monkeypatch.setattr(live_layers, "nws_alert_count", 5)
|
|
|
|
# _stats_counts imports the dicts/counters inside the function from live_layers,
|
|
# so monkeypatching the module attributes is what it observes.
|
|
from main import _stats_counts as fn
|
|
|
|
body = asyncio.run(fn())
|
|
assert body["aircraft"] == 7
|
|
assert body["vessels"] == 3
|
|
assert body["trains"] == 11
|
|
assert body["alerts"] == 5
|
|
|
|
|
|
def test_stats_db_failure_degrades_to_zero(monkeypatch):
|
|
"""A down DB yields zeros for the SQL-backed counters, never a 500."""
|
|
# Make the session factory raise synchronously so the try/except in
|
|
# _stats_counts degrades the SQL counters to zero (no dangling coroutine).
|
|
def _raise(*args, **kwargs):
|
|
raise RuntimeError("db down")
|
|
|
|
monkeypatch.setattr("main.async_session", _raise)
|
|
body = asyncio.run(_stats_counts())
|
|
assert body["cameras"] == 0
|
|
assert body["fires"] == 0
|
|
assert body["quakes"] == 0
|
|
assert isinstance(body["timestamp"], str) |