"""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: ./scripts/dev-tunnels.sh start # agents + upsd forwarded to localhost 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 os import httpx import pytest from card_contract import assert_card_contract from dashboards import collectors from dashboards.config import Settings, load from dashboards.rates import RateEstimator from dashboards.severity import classify from dashboards.sources import nut from dashboards.sources.host import HostCollector from ups_contract import assert_ups_hero_contract pytestmark = pytest.mark.live @pytest.fixture(scope="module") def settings() -> Settings: """Live env, not the import-time SETTINGS singleton.""" return load() @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) -> 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 — start ./scripts/dev-tunnels.sh") 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: # 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) await asyncio.sleep(1.2) result = await collectors.fetch_host_snapshot(client, target, host_collector) 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) -> 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 — start ./scripts/dev-tunnels.sh") async with httpx.AsyncClient(timeout=settings.host_timeout) as client: for target in settings.host_targets: await collectors.fetch_host_snapshot(client, target, host_collector) await asyncio.sleep(1.2) result = await collectors.fetch_host_snapshot(client, target, host_collector) 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" ) @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] rates = RateEstimator(span=settings.rate_window) 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" 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}"