"""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 dashboards import collectors from dashboards.config import Settings, load from dashboards.rates import RateEstimator from dashboards.sources import nut from dashboards.sources.host import HostCollector 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')}" ) @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_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}"