ducoterra
c4bb374ccc
Three dashboards served by one FastAPI app with a SQLite history store
(raw / 1m / 1h) and hand-written SVG charts: no Prometheus, no Grafana,
no CDN, no build step. Nothing is fetched from the internet at runtime.
* system: a stdlib-only agent (agent/host_agent.py) reports /proc and
/sys counters; the dashboard turns them into rates over a labelled
window, so every number says how it was measured.
* llm: gufo/llama.cpp and vLLM /metrics are parsed by an in-house
exposition parser. Measured delta/elapsed throughput beats a native
per-request gauge, and each value carries its provenance.
* ups: a NUT wire-protocol client plus health checks that name what
they looked at. Commands are gated by NUT_ALLOW_ACTIONS,
DASH_READ_ONLY and a DANGER confirmation, re-checked server-side.
Missing values render as n/a rather than 0, and an unconfigured source
answers 404 {"configured": false} without dialling anything, so "not
deployed here" never looks like an outage.
253 tests at 94% coverage against captured real fixtures; ruff and
pyright clean.
LLM-Generated: true
347 lines
15 KiB
Python
347 lines
15 KiB
Python
"""HTTP contract: what the pages may assume, and what must never leak.
|
|
|
|
The behaviours here are the ones a screenshot cannot show: an unconfigured source answering
|
|
404 `configured: false` instead of dialling a hostname that does not exist; a bearer token
|
|
that stops at the backend; the single write endpoint refusing on every gate; no external
|
|
asset the network can fail to deliver.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from dashboards.app import create_app
|
|
from dashboards.config import Settings, Target
|
|
from dashboards.store import Event, Sample, SampleStore
|
|
|
|
# refused immediately, and never dials out of the test environment
|
|
DEAD_URL = "http://127.0.0.1:1"
|
|
|
|
|
|
def build_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, simulate: bool = False,
|
|
**overrides: Any) -> TestClient:
|
|
"""A TestClient whose lifespan runs, against a private DB and no real network."""
|
|
if simulate:
|
|
monkeypatch.setenv("NUT_SIMULATE", "1")
|
|
else:
|
|
monkeypatch.delenv("NUT_SIMULATE", raising=False)
|
|
db = str(tmp_path / "api.sqlite3")
|
|
settings = Settings(db_path=db, **overrides)
|
|
store = SampleStore(db)
|
|
app = create_app(settings, store)
|
|
client = TestClient(app)
|
|
client.__enter__()
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
|
|
with build_client(tmp_path, monkeypatch) as test_client:
|
|
yield test_client
|
|
|
|
|
|
@pytest.fixture
|
|
def ups_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
|
|
"""NUT on the labelled simulator, actions allowed — the dev configuration."""
|
|
with build_client(tmp_path, monkeypatch, simulate=True, nut_host="127.0.0.1",
|
|
nut_allow_actions=True) as c:
|
|
yield c
|
|
|
|
|
|
def wait_for_payload(client: TestClient, path: str, *, timeout: float = 5.0) -> dict[str, Any]:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
response = client.get(path)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
time.sleep(0.1)
|
|
pytest.fail(f"{path} never became available")
|
|
|
|
|
|
# ---- unconfigured: say so, and do not dial -------------------------------- #
|
|
|
|
def test_config_reports_every_source_as_off(client) -> None:
|
|
body = client.get("/api/config").json()
|
|
assert body["system"]["enabled"] is False
|
|
assert body["system"]["targets"] == []
|
|
assert body["llm"]["gufo"]["enabled"] is False
|
|
assert body["llm"]["vllm"]["enabled"] is False
|
|
assert body["ups"]["enabled"] is False
|
|
assert body["ups"]["actions_enabled"] is False
|
|
assert body["ranges"], "the page builds its window buttons from this list"
|
|
|
|
|
|
def test_health_with_nothing_configured_is_not_a_failure(client) -> None:
|
|
"""Nothing configured is 'empty', not 'broken' — and the page must not go red."""
|
|
body = client.get("/api/health").json()
|
|
assert body["status"] in ("ok", "empty")
|
|
assert body["ok"] is True
|
|
assert body["sources"] == []
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["/api/system/snapshot", "/api/llm/snapshot",
|
|
"/api/ups/snapshot", "/api/ups/commands"])
|
|
def test_unconfigured_source_is_404_configured_false(client, path) -> None:
|
|
response = client.get(path)
|
|
assert response.status_code == 404
|
|
assert response.json()["configured"] is False
|
|
assert "detail" in response.json()
|
|
|
|
|
|
def test_summary_still_answers_with_no_sources(client) -> None:
|
|
body = client.get("/api/summary").json()
|
|
assert body["system"]["enabled"] is False
|
|
assert body["system"]["fleet"] is None
|
|
assert body["llm"]["enabled"] is False
|
|
assert body["ups"]["enabled"] is False
|
|
|
|
|
|
def test_unknown_route_is_json_not_html(client) -> None:
|
|
response = client.get("/api/nope")
|
|
assert response.status_code == 404
|
|
assert response.headers["content-type"].startswith("application/json")
|
|
assert "no such route" in response.json()["detail"]
|
|
|
|
|
|
# ---- tokens --------------------------------------------------------------- #
|
|
|
|
def test_agent_tokens_never_reach_the_browser(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("NUT_SIMULATE", "")
|
|
targets = [Target(name="driveripper", url="http://driveripper:9120",
|
|
headers={"Authorization": "Bearer SEKRIT-TOKEN"})]
|
|
db = str(tmp_path / "tokens.sqlite3")
|
|
store = SampleStore(db)
|
|
app = create_app(Settings(db_path=db, host_targets=tuple(targets)), store)
|
|
with TestClient(app) as client:
|
|
body = client.get("/api/config")
|
|
assert "SEKRIT-TOKEN" not in body.text
|
|
target = body.json()["system"]["targets"][0]
|
|
assert target["name"] == "driveripper"
|
|
assert "SEKRIT" not in target["url"] and "@" not in target["url"]
|
|
# and a host that never answered stays in the list, marked, with the fleet card intact
|
|
snapshot = wait_for_state(client, "/api/system/snapshot")
|
|
assert snapshot["hosts"][0]["name"] == "driveripper"
|
|
assert snapshot["hosts"][0].get("unreachable") is True
|
|
assert snapshot["fleet"]["hosts_total"] == 1
|
|
assert snapshot["fleet"]["hosts_live"] == 0
|
|
assert snapshot["fleet"]["memory_total_bytes"] is None
|
|
store.close()
|
|
|
|
|
|
def wait_for_state(client: TestClient, path: str, *, timeout: float = 5.0) -> dict[str, Any]:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
response = client.get(path)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
time.sleep(0.1)
|
|
pytest.fail(f"{path} never answered")
|
|
|
|
|
|
def test_target_problems_are_surfaced_not_swallowed(client) -> None:
|
|
body = client.get("/api/config").json()
|
|
assert body["system"]["target_problems"] == []
|
|
|
|
|
|
# ---- pages ---------------------------------------------------------------- #
|
|
|
|
@pytest.mark.parametrize("path,title", [
|
|
("/", "overview"), ("/system", "system"), ("/llm", "llm"), ("/ups", "ups"),
|
|
])
|
|
def test_pages_serve_and_declare_themselves(client, path, title) -> None:
|
|
response = client.get(path)
|
|
assert response.status_code == 200
|
|
assert "text/html" in response.headers["content-type"]
|
|
assert title in response.text.lower()
|
|
|
|
|
|
def test_no_page_loads_anything_from_the_internet(client) -> None:
|
|
"""No CDN, no webfonts: this dashboard has to work with no route to the internet."""
|
|
for path in ("/", "/system", "/llm", "/ups"):
|
|
html = client.get(path).text
|
|
for fragment in ("//cdn.", "https://fonts.", "//unpkg.com", "//cdnjs"):
|
|
assert fragment not in html, f"{path} references {fragment}"
|
|
|
|
|
|
def test_static_assets_are_served(client) -> None:
|
|
for asset in ("/static/dash.css", "/static/app.js", "/static/charts.js",
|
|
"/static/page-system.js", "/static/page-llm.js", "/static/page-ups.js",
|
|
"/static/page-index.js"):
|
|
assert client.get(asset).status_code == 200, asset
|
|
|
|
|
|
def test_api_responses_are_not_cacheable(client) -> None:
|
|
for path in ("/api/config", "/api/health", "/api/summary"):
|
|
assert "no-store" in client.get(path).headers.get("cache-control", "")
|
|
|
|
|
|
# ---- history -------------------------------------------------------------- #
|
|
|
|
def test_history_returns_seeded_samples(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
db = str(tmp_path / "hist.sqlite3")
|
|
store = SampleStore(db)
|
|
now = time.time()
|
|
for i in range(30):
|
|
store.add("host", "driveripper", "cpu.usage_pct", 10.0 + i, ts=now - 300 + i)
|
|
app = create_app(Settings(db_path=db, host_targets=(Target("driveripper", DEAD_URL),)), store)
|
|
with TestClient(app) as client:
|
|
body = client.get("/api/history", params={"source": "host", "host": "driveripper",
|
|
"metric": "cpu.usage_pct", "range": "15m"})
|
|
assert response_ok(body)
|
|
points = body.json()["points"]
|
|
assert len(points) >= 25
|
|
assert points[-1]["value"] == pytest.approx(39.0)
|
|
store.close()
|
|
|
|
|
|
def test_history_rejects_an_unknown_range(client) -> None:
|
|
response = client.get("/api/history", params={"source": "host", "host": "h",
|
|
"metric": "m", "range": "banana"})
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_history_requires_the_series_coordinates(client) -> None:
|
|
assert client.get("/api/history").status_code == 422
|
|
|
|
|
|
def test_metrics_endpoint_lists_what_actually_has_history(client) -> None:
|
|
body = client.get("/api/metrics").json()
|
|
assert body["metrics"] == []
|
|
|
|
|
|
# ---- events --------------------------------------------------------------- #
|
|
|
|
def test_events_are_returned_in_json(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
db = str(tmp_path / "events.sqlite3")
|
|
store = SampleStore(db)
|
|
store.log_event(Event(ts=time.time(), source="host", host="deskwork", kind="down",
|
|
severity="error", message="deskwork is not answering"))
|
|
app = create_app(Settings(db_path=db), store)
|
|
with TestClient(app) as client:
|
|
events = client.get("/api/events").json()["events"]
|
|
assert events[0]["host"] == "deskwork"
|
|
assert events[0]["severity"] == "error"
|
|
store.close()
|
|
|
|
|
|
# ---- the one write endpoint ------------------------------------------------ #
|
|
|
|
def test_command_endpoint_is_post_only(ups_client) -> None:
|
|
assert ups_client.post("/api/ups/command", json={}).status_code in (400, 403, 404)
|
|
assert ups_client.put("/api/ups/command", json={}).status_code == 405
|
|
assert ups_client.delete("/api/ups/command").status_code == 405
|
|
|
|
|
|
def test_commands_are_offered_when_actions_are_enabled(ups_client) -> None:
|
|
body = wait_for_payload(ups_client, "/api/ups/snapshot")
|
|
assert body["simulated"] is True, "the simulator must label itself"
|
|
assert body["actions_enabled"] is True
|
|
ups = body["ups"][0]
|
|
enabled = [item for item in ups["commands"] if item["enabled"]]
|
|
assert enabled, "with NUT_ALLOW_ACTIONS=1 and a simulator, safe commands are sendable"
|
|
assert all(item["risk"] in ("safe", "caution", "dangerous") for item in ups["commands"])
|
|
|
|
|
|
def test_a_safe_command_actually_reaches_the_driver(ups_client) -> None:
|
|
body = wait_for_payload(ups_client, "/api/ups/snapshot")
|
|
name = body["ups"][0]["name"]
|
|
response = ups_client.post("/api/ups/command",
|
|
json={"ups": name, "command": "test.panel.start"})
|
|
assert response.status_code == 200, response.text
|
|
payload = response.json()
|
|
assert payload["ok"] is True and payload["simulated"] is True
|
|
events = ups_client.get("/api/events").json()["events"]
|
|
assert any("test.panel.start" in event["message"] or "Panel test" in event["message"]
|
|
for event in events), "every sent command must leave an audit trail"
|
|
|
|
|
|
def test_a_dangerous_command_needs_the_confirmation_word(ups_client) -> None:
|
|
name = wait_for_payload(ups_client, "/api/ups/snapshot")["ups"][0]["name"]
|
|
deep = "test.battery.start.deep"
|
|
refusal = ups_client.post("/api/ups/command", json={"ups": name, "command": deep})
|
|
assert refusal.status_code == 422
|
|
assert "DANGER" in refusal.json()["detail"]
|
|
assert ups_client.post("/api/ups/command",
|
|
json={"ups": name, "command": deep, "confirm": "DANGER"}).status_code == 200
|
|
|
|
|
|
def test_unknown_commands_are_refused_even_with_actions_on(ups_client) -> None:
|
|
name = wait_for_payload(ups_client, "/api/ups/snapshot")["ups"][0]["name"]
|
|
for command in ("load.format", "rm -rf /", "INSTCMD x y"):
|
|
response = ups_client.post("/api/ups/command", json={"ups": name, "command": command})
|
|
assert response.status_code == 400, command
|
|
|
|
|
|
def test_commands_for_an_unknown_ups_are_404(ups_client) -> None:
|
|
wait_for_payload(ups_client, "/api/ups/snapshot")
|
|
response = ups_client.post("/api/ups/command",
|
|
json={"ups": "not-real", "command": "beeper.mute"})
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_missing_fields_are_a_400(ups_client) -> None:
|
|
assert ups_client.post("/api/ups/command", json={"ups": ""}).status_code == 400
|
|
|
|
|
|
def test_actions_flag_off_blocks_every_command(tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch) -> None:
|
|
with build_client(tmp_path, monkeypatch, simulate=True, nut_host="127.0.0.1") as client:
|
|
body = wait_for_payload(client, "/api/ups/snapshot")
|
|
name = body["ups"][0]["name"]
|
|
assert body["actions_enabled"] is False
|
|
response = client.post("/api/ups/command",
|
|
json={"ups": name, "command": "beeper.mute"})
|
|
assert response.status_code == 403
|
|
assert "NUT_ALLOW_ACTIONS" in response.json()["detail"]
|
|
|
|
|
|
def test_read_only_beats_the_actions_flag(tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch) -> None:
|
|
with build_client(tmp_path, monkeypatch, simulate=True, nut_host="127.0.0.1",
|
|
nut_allow_actions=True, read_only=True) as client:
|
|
body = wait_for_payload(client, "/api/ups/snapshot")
|
|
name = body["ups"][0]["name"]
|
|
response = client.post("/api/ups/command",
|
|
json={"ups": name, "command": "beeper.mute"})
|
|
assert response.status_code == 403
|
|
commands = wait_for_payload(client, "/api/ups/commands")
|
|
assert commands["read_only"] is True
|
|
assert all(item["enabled"] is False for ups in commands["ups"]
|
|
for item in ups["commands"])
|
|
|
|
|
|
def test_ups_history_answers_for_simulated_ups(ups_client) -> None:
|
|
wait_for_payload(ups_client, "/api/ups/snapshot")
|
|
time.sleep(1.2)
|
|
body = ups_client.get("/api/ups/history", params={"metrics": "ups.charge_pct",
|
|
"range": "1h"}).json()
|
|
assert body["ups"], "the UPS page needs a name to chart before any history exists"
|
|
|
|
|
|
def response_ok(response) -> bool:
|
|
assert response.status_code == 200, response.text
|
|
return True
|
|
|
|
|
|
def test_samples_written_by_the_poller_are_queryable(tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""The write path (poller -> store) and the read path (/api/history) must agree on names."""
|
|
db = str(tmp_path / "roundtrip.sqlite3")
|
|
store = SampleStore(db)
|
|
store.add_many([Sample("llm", "gufo", "llm.decode_tokens_per_second", 40.0, time.time() - 60),
|
|
Sample("llm", "gufo", "llm.decode_tokens_per_second", 45.0, time.time() - 30)])
|
|
app = create_app(Settings(db_path=db), store)
|
|
with TestClient(app) as client:
|
|
metrics = client.get("/api/metrics", params={"source": "llm"}).json()["metrics"]
|
|
assert "llm.decode_tokens_per_second" in metrics
|
|
points = client.get("/api/history", params={"source": "llm", "host": "gufo",
|
|
"metric": "llm.decode_tokens_per_second",
|
|
"range": "15m"}).json()["points"]
|
|
assert [p["value"] for p in points] == [40.0, 45.0]
|
|
store.close()
|