dashboards/tests/test_live_fleet.py

"""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 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


@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.*@<dev>` 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.
    """
    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"]
        assert (route["connections"] is None) is (out["address"] is not True), (
            f"{route['id']}: only an endpoint the config wrote as an address can be counted")
        if out["address"] is True and isinstance(out["port"], int):
            joinable.add(str(route["metric"]))
            endpoints.setdefault((out["host"], out["port"]), 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 address endpoint publishes an outbound gauge — a domain-keyed series is the one "
        "phase 30 stopped writing, because the figure it carried belonged to the port")


@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}"