"""The only tests that touch a network: `uv run pytest -m live --live`. Everything else in this suite is hermetic on purpose (see conftest.py). These are the bridge to reality — they re-run the *production* code paths (`collectors.fetch_*`, `NutMonitor.collect`) against real targets, which is what catches an upstream that renamed a series or a driver that started reporting a temperature in kelvin. They read exactly the env vars the app reads, so the dev loop is direct dial — the same addresses the deployed dashboard dials, nothing forwarded: set -a; . .env.dev; set +a uv run pytest -m live --live Each test skips when its source is not configured, so running the whole set with half a fleet reachable is a normal outcome, not a failure. **Nothing in here may write.** The UPS test asserts the command gating is honest; it never sends an instant command — a battery test on hardware someone else is plugged into is not a unit test. """ from __future__ import annotations import asyncio import ipaddress import json import os import sqlite3 from collections.abc import Iterator from typing import Any import httpx import pytest from card_contract import assert_card_contract from dashboards import certs, collectors from dashboards.config import Settings, load from dashboards.rates import RateEstimator from dashboards.severity import classify from dashboards.sources import gateway, nut from dashboards.sources.disks import SMART_SERIES, DisksCollector from dashboards.sources.host import HostCollector from slots_contract import assert_slot_shape from ups_contract import assert_ups_hero_contract pytestmark = pytest.mark.live def endpoint_identity(out: Any) -> tuple[Any, ...] | None: """One endpoint in the kernel's currency, derived **here** rather than by the source. An address endpoint is its own address; a name endpoint is the addresses the agent's resolver published for it (`out.resolved`), normalised so `::ffff:10.1.2.64` and `10.1.2.64` are one key. `None` is "nothing can name this far end", which is counted nowhere. Phase 32 made this helper necessary: a live router can have two route lines reaching one address through two different names, and a total keyed on the config's text would book it twice. """ port = out.get("port") if not isinstance(port, int): return None spellings = ([out["host"]] if out.get("address") is True else list(out.get("resolved") or [])) keys: list[str] = [] for spelling in spellings: try: address = ipaddress.ip_address(str(spelling)) except ValueError: continue mapped = getattr(address, "ipv4_mapped", None) keys.append(str(mapped if mapped is not None else address)) return (min(keys), port) if keys else None @pytest.fixture(scope="module") def settings() -> Settings: """Live env, not the import-time SETTINGS singleton.""" return load() class _StoredAuthority: """`certs.MetaStore` over an existing database, opened **read-only**. The live suite has to dial the fleet the way the app does, and since the phase-24 rollout that means presenting this dashboard's own client certificate. The authority lives in the `meta` table of the store the dev server owns, so this is the smallest honest reader of it: `mode=ro` on the URI, `get_meta` the only method that does anything, and the two write methods a *refusal* rather than a no-op — this file's docstring says nothing in here may write, and minting or repairing an authority from a test would be the worst write it could make. """ def __init__(self, path: str) -> None: self._conn = sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True) def close(self) -> None: self._conn.close() def get_meta(self, key: str) -> str | None: row = self._conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone() return None if row is None else str(row[0]) def set_meta(self, key: str, value: str) -> None: raise AssertionError(f"live tests are read-only: refused to write {key!r}") def delete_meta(self, key: str) -> int: raise AssertionError(f"live tests are read-only: refused to delete {key!r}") @pytest.fixture(scope="module") def agent_tls(settings: Settings) -> Iterator[dict[str, collectors.AgentTls]]: """One :class:`collectors.AgentTls` per `https://` host, from the authority in the store. Empty when no `HOST_TARGETS` entry is https — every host is then dialled on the pre-phase-15 token path, which is what an `http://` fleet still is — and skipped when the store or its `certs.*` rows are absent, because a box running the live suite without the dev database is a normal partial environment rather than a failure (the same rule that lets half a fleet be reachable). `certs.read_state` is the request-time read and never `ensure_certs`: a live test may not mint a root CA, and the material this writes is removed on teardown. """ targets = settings.host_targets if not any(target.url.startswith("https://") for target in targets): yield {} return if not os.path.exists(settings.db_path): pytest.skip(f"no store at {settings.db_path} to read the CA from") authority = _StoredAuthority(settings.db_path) try: state = certs.read_state(authority) except (RuntimeError, sqlite3.Error) as exc: authority.close() pytest.skip(f"no readable authority in {settings.db_path}: {type(exc).__name__}") material = certs.write_material(state) try: yield collectors.agent_tls_map(material, authority, targets) finally: certs.remove_material(material) authority.close() @pytest.fixture def host_collector() -> HostCollector: return HostCollector(rates=RateEstimator(span=10.0)) async def test_every_host_agent_answers_and_parses(settings: Settings, host_collector: HostCollector, agent_tls: dict[str, collectors.AgentTls]) -> None: """Each HOST_TARGETS entry must answer /snapshot and yield the metrics the page renders.""" if not settings.host_targets: pytest.skip("HOST_TARGETS is unset") required = {"cpu.usage_pct", "mem.used_pct", "disk.used_pct"} async with httpx.AsyncClient(timeout=settings.host_timeout) as client: for target in settings.host_targets: tls = agent_tls.get(target.name) # Poll twice: usage_pct is a Δ/jiffies rate, and a rate has no second point yet # on the first scrape. That is the design (the agent never remembers who polled # it), so a live test that scraped once would be asserting a false failure. await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls) await asyncio.sleep(1.2) result = await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls) metrics = {sample.metric for sample in result.samples} assert result.payload["info"]["hostname"], f"{target.name}: no hostname" missing = required - metrics assert not missing, f"{target.name}: snapshot parsed but no {sorted(missing)}" # a section the agent could not read is a `degraded` note, never a silent zero assert result.payload.get("degraded") in (None, []), ( f"{target.name}: agent reported {result.payload.get('degraded')}" ) async def test_every_host_serves_the_hero_card_contract(settings: Settings, host_collector: HostCollector, agent_tls: dict[str, collectors.AgentTls]) -> None: """The phase-03 `/system` card paths exist on the **real** agents, not just the captures. Values may honestly be `n/a` on real hardware (gateway reports no CPU sensor and no GPU at all); the *shape* is what phase 03 renders, so this is the test that catches an agent upgrade renaming a field. Hermetic twin: `tests/test_system_contract.py`. """ if not settings.host_targets: pytest.skip("HOST_TARGETS is unset") async with httpx.AsyncClient(timeout=settings.host_timeout) as client: for target in settings.host_targets: tls = agent_tls.get(target.name) await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls) await asyncio.sleep(1.2) result = await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls) assert_card_contract(result.payload) assert result.payload["cpu"]["temp_c"] == result.payload["temperatures"]["cpu_c"], ( f"{target.name}: the card and /api/summary would report different CPU temps" ) async def test_agent_disk_rows_carry_the_smart_shape( settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None: """Phase-09: a redeployed agent answers `smart` per physical disk of `/snapshot`. Agents are redeployed by hand, so an agent whose rows carry no `smart` key at all SKIPS this host rather than failing it. `smart: null` is a documented state (no smartctl binary, or the agent user lacks device read access) — what must hold is the shape of a non-null block and that unknowns are null, never 0. Read-only: nothing here sends `smartctl -t`; a self-test on hardware something is plugged into is the operator's call, not a unit test. """ if not settings.host_targets: pytest.skip("HOST_TARGETS is unset") seen_smart = 0 async with httpx.AsyncClient(timeout=settings.host_timeout) as client: for target in settings.host_targets: try: # through `fetch_snapshot`, not a hand-rolled `get`, because that is the function # that owns the certificate dial: an `https://` agent needs this app's own CA and # client leaf, and a test dialling it with the platform trust store would fail on # a healthy fleet (`CERTIFICATE_VERIFY_FAILED: self-signed certificate in chain`). response = await collectors.fetch_snapshot(client, target, tls=agent_tls.get(target.name)) except collectors.UpstreamError: continue # reachability has its own live test devices = (response.get("disk_io") or {}).get("devices") or [] if devices and not all("smart" in device for device in devices): continue # agent predates the smart pass — redeploy it to test this for device in devices: seen_smart += 1 smart = device.get("smart") if smart is None: continue # degraded by design; the errors[] entry says why # the literal set is the pin: a rename on the agent must fail here too assert set(smart) == {"model", "serial", "overall_health", "temp_c", "power_on_hours", "reallocated_sector_ct", "current_pending_sector_ct", "offline_uncorrectable", "percentage_used"}, \ f"{target.name}/{device['name']}" if smart["power_on_hours"] is not None: assert smart["power_on_hours"] > 0 if smart["temp_c"] is not None: assert -20.0 < smart["temp_c"] < 125.0 assert smart["overall_health"] in (None, "PASSED", "FAILED", "UNKNOWN") if not seen_smart: pytest.skip("no agent redeployed with the smart pass yet (deploy/install-agent.sh)") async def test_disks_source_charts_the_physical_drives_the_agent_reports( settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None: """Phase 09 task 02: the `disks` source over a real agent snapshot. This is the production code path — the fetch the poller runs, not a hand-built dict — so what fails here is a rename between the agent and the page, not a typo in a fixture. The assertions hold whether or not this fleet can read SMART yet: one row per whole disk, `disk.*@` series only for values that exist, and no invented numbers. """ if not settings.host_targets: pytest.skip("HOST_TARGETS is unset") collector = DisksCollector() checked = 0 async with httpx.AsyncClient(timeout=settings.disks_timeout) as client: for target in settings.host_targets: try: result = await collectors.fetch_disk_snapshot( client, target, collector, timeout=settings.disks_timeout, tls=agent_tls.get(target.name)) except collectors.UpstreamError: continue # reachability has its own live test checked += 1 payload = result.payload assert payload["name"] == target.name assert payload["disks"], f"{target.name}: no physical disks at all?" for row in payload["disks"]: assert row["path"] == f"/dev/{row['device']}" assert "/" not in row["device"], f"partition {row['device']} in a disk list" assert row["health"] in (None, "ok", "failed") devices = {row["device"] for row in payload["disks"]} bases = {base for _field, base in SMART_SERIES} for sample in result.samples: assert sample.source == "disks", "the wrong family would chart nowhere" base, _, device = sample.metric.partition("@") assert base in bases and device in devices, ( f"{target.name}: {sample.metric} is not a pinned disk series") assert sample.value == sample.value and abs(sample.value) != float("inf"), \ "NaN must never reach the store" if any(row["smart"] for row in payload["disks"]): assert payload["smartctl_available"] is True if not checked: pytest.skip("no host agent answered") async def test_the_gateway_source_joins_the_live_ingress(settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None: """Phase 29/30: `sources/gateway.py` over the **real** gateway agent, read-only. The production path — the same `fetch_snapshot` the poller will run, through the stored mTLS authority — so what fails here is a drift between the agent's sections and the join, not a typo in a fixture. The assertions are the ones that hold whatever the traffic is: the agent answers the three sections, every configured frontend and `listen` port is a trunk, and the inbound count **re-derived here** from the snapshot's own ESTABLISHED rows equals the sum the payload prints — the identity of phase 30's simplified join, checked against the kernel rather than against a fixture. The totals are an inequality by design (`sum(listeners) + sum(endpoints) <= established`): a socket neither config file describes is counted nowhere, and on this router that is most of the table. Phase 32 widened **which** routes can be counted — an endpoint the config wrote as an address, or one the gateway's own resolver published for the name it wrote — and the two assertions near the end are stated in those terms, with the endpoint identity derived here so the outbound total is not simply the source's own arithmetic re-added. """ target = next((item for item in settings.host_targets if item.name == settings.gateway_host), None) if target is None: pytest.skip(f"GATEWAY_HOST={settings.gateway_host!r} is not in HOST_TARGETS") async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client: try: snapshot = await collectors.fetch_snapshot(client, target, timeout=settings.gateway_timeout, tls=agent_tls.get(target.name)) except collectors.UpstreamError as exc: pytest.skip(f"{target.name} did not answer: {exc}") payload = gateway.parse(snapshot, target.name) if not payload["supported"]: pytest.fail(f"{target.name} runs agent {payload.get('agent_version')} — the gateway page " "needs 1.2.0's haproxy/nginx/sockets sections (AGENTS.md gap 11's rollout)") assert payload["host"] == target.name assert payload["sockets"]["established"] == len(snapshot["sockets"]["established"]) assert payload["listeners"], "no frontend on either config would leave the page empty" assert payload["connections"], "the two configs declare no route at all" for trunk in payload["listeners"]: assert isinstance(trunk["connections"], int), "the socket table answered, so a count is known" # the join, re-derived from the agent's own rows rather than from the payload's arithmetic configured = {trunk["port"] for trunk in payload["listeners"]} assert configured, "a gateway with no configured port has nothing to reconcile" inbound = [row for row in snapshot["sockets"]["established"] if ((row.get("local") or {}).get("port")) in configured] assert sum(t["connections"] for t in payload["listeners"]) == len(inbound), ( "the trunk counts are the ESTABLISHED rows whose local port a config names — no more, " "no less") endpoints: dict[tuple[Any, Any], Any] = {} joinable: set[str] = set() for route in payload["connections"]: assert route["in"]["port"] in configured, "a route hangs off no trunk the page draws" out = route["out"] placed = out["address"] is True or bool(out.get("resolved")) assert (route["connections"] is None) is (not placed), ( f"{route['id']}: only an endpoint whose address is known — written, or resolved by the " "gateway itself — can carry a count") key = endpoint_identity(out) if key is not None: joinable.add(str(route["metric"])) endpoints.setdefault(key, route["connections"]) else: assert route["metric"] is None, "a route nobody can count publishes no series" live_out = sum(value for value in endpoints.values() if isinstance(value, int)) assert sum(t["connections"] for t in payload["listeners"]) + live_out \ <= payload["sockets"]["established"], "the page may under-count the kernel, never exceed it" # and nothing outside the two config files reached the payload at all — this app's own agent # port is the case phase 30 deleted the reporting of text = json.dumps(payload) assert ":9120" not in text and "@9120" not in text, "the agent's own port leaked into the map" stored = gateway.samples(payload, 0.0) assert all(sample.source == gateway.SOURCE for sample in stored) assert {sample.metric for sample in stored if "out_conns" in sample.metric} <= joinable, ( "only an endpoint whose address the join knows publishes an outbound gauge — written as an " "address, or resolved by the gateway itself (phase 32); an unplaced name publishes nothing") async def test_a_resolved_target_on_the_live_ingress_is_never_a_dash( settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None: """Phase 32 task 03 against the real ingress: `ok` in the resolver means a number on the page. The operator's complaint was that one line out of thirty-three moved, and the acceptance line for removing the dash is that it goes because the gateway published the address — not because a column was hidden. So every route the resolver placed carries a **finite** `connections` (`0` allowed: resolved and idle is a reading), a non-empty `out.resolved`, the agent's own `via`, the timestamp of the lookup, and the gauge that goes with it. Skips when the ingress runs an agent with no `resolver` section yet — which is every host until task 05 re-stages it. """ target = next((item for item in settings.host_targets if item.name == settings.gateway_host), None) if target is None: pytest.skip(f"GATEWAY_HOST={settings.gateway_host!r} is not in HOST_TARGETS") async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client: try: snapshot = await collectors.fetch_snapshot(client, target, timeout=settings.gateway_timeout, tls=agent_tls.get(target.name)) except collectors.UpstreamError as exc: pytest.skip(f"{target.name} did not answer: {exc}") section = snapshot.get("resolver") if not isinstance(section, dict): pytest.skip(f"{target.name} publishes no resolver section — agent 1.3.0 is task 05's " "re-stage; until then every name-written route is a dash by design") payload = gateway.parse(snapshot, target.name) if section.get("available") is not True: pytest.skip(f"{target.name}'s resolver did not answer " f"(available={section.get('available')!r}): nothing is placed, so nothing " "is expected to be counted") placed = [route for route in payload["connections"] if route["out"].get("resolved")] assert placed, (f"{target.name} published {len(section.get('entries') or [])} resolver " "entries and the join placed none of them") for route in placed: out = route["out"] # The count lives on the **route** (`connections[]`), not on its `out` endpoint — the same # shape `test_the_gateway_source_joins_the_live_ingress` reads. This path only ever ran once # agent 1.3.0 was installed (before that the whole test skipped for want of a `resolver` # section), which is how a wrong key survived task 03: a KeyError is not a dash. assert isinstance(route["connections"], int) and not isinstance(route["connections"], bool), ( f"{route['id']}: resolved to {out['resolved']} and the line is still a dash") assert out["via"] == section.get("via") == "libc", ( "the vantage is the agent's string, copied, never one this module invented") assert isinstance(out["resolved_ts"], int | float), ( f"{route['id']}: a count obtained through a lookup must say when the lookup happened") assert route["metric"] is not None, "a counted route always has its gauge" stored = {sample.metric for sample in gateway.samples(payload, 0.0)} assert {route["metric"] for route in placed} <= stored, ( "a resolved route's count must reach the store, or its chart stays empty forever") @pytest.mark.parametrize("engine", ["gufo", "vllm"]) async def test_engine_still_publishes_the_series_we_parse(engine: str, settings: Settings) -> None: """Metric names are verified against the deployed engine, never guessed. The assertion is that at least one cumulative counter per family *resolves* — that is what the Δ rates, the throughput cells, and the history rows all hang off. If an upgrade renames it, this fails here instead of printing 0 tok/s on the dashboard. """ base_url = settings.gufo_url if engine == "gufo" else settings.vllm_url api_key = settings.gufo_api_key if engine == "gufo" else settings.vllm_api_key if not base_url: pytest.skip(f"{engine.upper()}_BASE_URL is unset — that engine is not deployed here") counters = { "gufo": ("tokens_generated_total", "tokens_prompt_total"), "vllm": ("generation_tokens_total", "prompt_tokens_total"), }[engine] # The span the LLM jobs really get: phase 23 spans the LLM estimator by the poll # interval, not by the host's `RATE_WINDOW_SECONDS`. rates = RateEstimator(span=settings.llm_interval) async with httpx.AsyncClient(timeout=settings.llm_timeout) as client: result = await collectors.fetch_engine(client, engine, base_url, api_key, rates, with_slots=engine == "gufo") metrics = result.payload["metrics"] resolved = [key for key in counters if metrics[key]["source"] is not None] assert resolved, ( f"{engine}: none of {counters} resolved at {base_url} — upstream renamed its " f"counters, update {engine.upper()}_SERIES in sources/llm.py" ) sessions = result.payload["sessions"] assert sessions["active"] is None or sessions["active"] >= 0 assert result.samples, f"{engine}: nothing to store, so the charts would stay empty" async def test_gufo_slots_answer_the_pinned_shape(settings: Settings) -> None: """The `/slots` rows behind the phase-07 slot grid, asserted against the running gufo. Hermetic twin: `tests/test_llm_slots_contract.py`, driven from `tests/fixtures/slots_gufo.json`. This is the half that catches gufo renaming a slot field — the old table rendered dashes because nobody had ever pinned the shape, and the assertion here goes through the production path (`collectors.fetch_engine`, which owns the `/slots` → `/v1/slots` fallback), not a hand-rolled request. """ if not settings.gufo_url: pytest.skip("GUFO_BASE_URL is unset — no gufo to ask for /slots") # The span the LLM jobs really get: phase 23 spans the LLM estimator by the poll # interval, not by the host's `RATE_WINDOW_SECONDS`. rates = RateEstimator(span=settings.llm_interval) async with httpx.AsyncClient(timeout=settings.llm_timeout) as client: result = await collectors.fetch_engine(client, "gufo", settings.gufo_url, settings.gufo_api_key, rates, with_slots=True) rows = result.payload["sessions"]["list"] assert rows, ( "gufo answered /slots with no rows — even an idle engine reports one row per " "parallel slot (the idle shape pinned in tests/fixtures/slots_gufo.json); " f"detail: {result.payload['sessions']['detail']!r}" ) assert_slot_shape(rows) def require_upsd(settings: Settings) -> None: """Skip, not fail, when there is no real upsd to talk to.""" if not settings.nut_enabled: pytest.skip("NUT_HOST is unset — no upsd to talk to") if os.environ.get("NUT_SIMULATE", "").lower() in ("1", "true", "yes", "on"): pytest.skip("NUT_SIMULATE=1 — point NUT_HOST at the real upsd for this run") async def test_upsd_reports_both_upses_with_sane_units(settings: Settings) -> None: """Real batteries: plausible volts/percent, and the health checks are never silent.""" require_upsd(settings) result = await collectors.NutMonitor(settings).collect() upses = result.payload["ups"] assert upses, "upsd answered but exported no UPS (check ups.conf driver instances)" for card in upses: name = card["name"] assert card["status"], f"{name}: ups.status is empty" assert card["health"]["checks"], f"{name}: assess() produced no health checks" numbers = card["numbers"] charge = numbers.get("charge_pct") assert charge is None or 0.0 <= charge <= 100.0, f"{name}: charge_pct={charge}" # kelvin-as-celsius is the classic unit bug here; 273 would sail past a bare >0 temp = numbers.get("battery_temperature") assert temp is None or -20.0 <= temp <= 80.0, f"{name}: battery temp {temp} is not °C" async def test_ups_hero_fields_are_present_on_the_real_batteries(settings: Settings) -> None: """Phase 05's hero row and battery glyph read these paths on the real UPSes. Hermetic twin: `tests/test_ups_contract.py`, driven from the verbatim captures in `tests/fixtures/nut_*.vars`. This is the half that catches a driver update renaming a variable, adding `battery.charger.status`, or starting to report a temperature in a unit the page has never seen — the assertion is on the shape and the vocabulary, never on a number that moves between polls. """ require_upsd(settings) result = await collectors.NutMonitor(settings).collect() cards = result.payload["ups"] assert cards, "upsd answered but exported no UPS — nothing for the hero row to render" for card in cards: assert_ups_hero_contract(card) # includes: charger_status is None or a token we label runtime = card["numbers"]["battery_runtime"] assert card["severity"]["battery_runtime"] == (None if runtime is None else classify("ups.battery_runtime", runtime)) assert not (card["on_battery"] and card["status"] and "OL" in card["status"].split()), ( f"{card['name']}: on_battery and OL at once — the glyph would show the wrong state" ) async def test_ups_command_gating_is_honest_without_sending_anything(settings: Settings) -> None: """Every advertised command is either enabled or carries the reason it is not. Deliberately does not POST anything: `POST /api/ups/command` behaviour is covered by tests/test_api.py against a fake client, because running a deep battery test on live hardware to prove a button works is exactly what this app is careful about. """ require_upsd(settings) result = await collectors.NutMonitor(settings).collect() seen = False for card in result.payload["ups"]: for entry in card["commands"]: seen = True assert isinstance(entry["enabled"], bool), f"{card['name']}: {entry} has no gate state" if not entry["enabled"]: assert entry.get("reason"), ( f"{card['name']}: {entry['command']} disabled with no reason — the UI " f"would show a dead button with no explanation" ) if entry.get("risk") == nut.RISK_DANGEROUS: assert entry.get("confirm_text"), ( f"{card['name']}: {entry['command']} can cut power without a typed confirmation" ) if not seen: pytest.skip("the UPS advertises no commands we model (see ACTION_CATALOG)") def test_target_config_is_clean_and_leak_free(settings: Settings) -> None: """Every configured target parses, and none of them leak a userinfo token.""" names = [target.name for target in settings.host_targets] assert len(names) == len(set(names)), f"duplicate target names: {names}" for target in settings.host_targets: assert target.url.startswith(("http://", "https://")), f"{target.name}: {target.url}" assert "@" not in target.display_url, f"{target.name}: display_url leaks credentials" for problem in settings.host_target_problems: assert isinstance(problem, str) and problem, f"empty problem: {problem!r}"