dashboards/tests/conftest.py

"""Shared fixtures: real captures for shape, fakes for behaviour.

Anything that must match an upstream (an agent snapshot, an engine's /metrics text, the NUT
wire protocol) is asserted against a verbatim capture in tests/fixtures/. Everything that
needs a *failure* — a dead host, a refused command, a counter reset — is a fake here.

The `live` marker is the bridge to real hardware: those tests dial the fleet, the engines,
or upsd, so they SKIP unless you pass `--live` and have set the env vars they read (see
tests/test_live_fleet.py). That keeps `uv run pytest` hermetic and offline while still
letting `uv run pytest -m live --live` catch an upstream that renamed something.
"""
from __future__ import annotations

import copy
import json
import sys
from pathlib import Path
from typing import Any

import pytest


def pytest_addoption(parser: pytest.Parser) -> None:
    parser.addoption(
        "--live",
        action="store_true",
        default=False,
        help="run tests marked `live` against the real fleet/engines/upsd (they skip otherwise)",
    )


def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
    if config.getoption("--live"):
        return
    skip = pytest.mark.skip(reason="needs --live and a reachable target (see tests/test_live_fleet.py)")
    for item in items:
        if "live" in item.keywords:
            item.add_marker(skip)

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
if str(SRC) not in sys.path:
    sys.path.insert(0, str(SRC))

from dashboards.rates import RateEstimator  # noqa: E402
from dashboards.sources.host import HostCollector, parse_host_snapshot  # noqa: E402
from dashboards.store import SampleStore  # noqa: E402

FIXTURES = Path(__file__).resolve().parent / "fixtures"


@pytest.fixture
def db(tmp_path: Path) -> Any:
    store = SampleStore(tmp_path / "t.sqlite3")
    try:
        yield store
    finally:
        store.close()


@pytest.fixture
def store(db: Any) -> Any:
    """Alias: some tests read better with `store`, others with `db`. Same object."""
    return db


@pytest.fixture
def agent_snapshot() -> dict[str, Any]:
    """A real driveripper snapshot (Threadripper + R9700, 48 logical CPUs)."""
    return json.loads((FIXTURES / "agent_driveripper.json").read_text())


@pytest.fixture
def agent_snapshots() -> dict[str, dict[str, Any]]:
    """Every captured host, keyed by name — fleet code paths need >1 host to mean anything."""
    out: dict[str, dict[str, Any]] = {}
    for path in sorted(FIXTURES.glob("agent_*.json")):
        out[path.stem.removeprefix("agent_")] = json.loads(path.read_text())
    return out


@pytest.fixture
def host_scrape() -> Any:
    """`host_scrape(snapshot, cpu_busy=10.0, mutate=None)` -> one parsed host payload.

    Two scrapes are needed before any rate exists (the agent reports counters; the
    dashboard owns the window), and the payload the API serves is the second one. `mutate`
    edits the snapshot before both scrapes so a test can put a value on a chosen side of
    `severity.THRESHOLDS`; `cpu_busy` sets the CPU percentage over the window directly.
    Driven from the real capture so a schema drift breaks here rather than in a browser.
    """
    def scrape(snapshot: dict[str, Any], *, name: str = "driveripper", cpu_busy: float = 10.0,
               mutate: Any = None) -> dict[str, Any]:
        snap = copy.deepcopy(snapshot)
        if mutate is not None:
            mutate(snap)
        collector = HostCollector(RateEstimator(span=60.0))
        parse_host_snapshot(name, snap, collector, now=1_000.0)
        second = copy.deepcopy(snap)
        second["ts"] = 1_010.0
        modes = second["cpu"]["modes"]
        # busy = (Δtotal - Δidle)/Δtotal, and iowait is left untouched so it rates at 0
        modes["total"] += 1_000.0
        modes["idle"] += 1_000.0 - (cpu_busy * 10.0)
        return parse_host_snapshot(name, second, collector, now=1_010.0).payload

    return scrape


# ---- phase 09: physical disks (the /disks source and page) ----------------- #

def _load_host_agent() -> Any:
    """The stdlib agent module, loaded by path: it is not an installed package (and must
    not become one — it ships as a single file to every host). Same loader as
    tests/test_agent.py, kept private here because only the disk fixtures need it."""
    import importlib.util  # noqa: PLC0415 - only the disk fixtures load the agent

    path = FIXTURES.parent.parent / "agent" / "host_agent.py"
    spec = importlib.util.spec_from_file_location("host_agent", path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules.setdefault("host_agent", module)
    spec.loader.exec_module(module)
    return module


#: The two `smartctl` shapes the fleet has: NVMe (health-log fields) and ATA (the vendor
#: attribute table). `tests/fixtures/smartctl_reference_*.txt` — labels verbatim from the
#: deployed smartmontools 7.5, values synthetic (see the header of each file).
SMART_SHAPES = ("nvme", "ata_hdd")

#: The agent's own words for the way SMART goes missing that the page must name: the
#: binary is not installed. Mirrors `host_agent.Collector._smart_pass()`; the agent test
#: `test_missing_smartctl_binary_is_one_error_for_the_whole_pass` pins that the line it
#: really writes still contains the word `sources.disks.MISSING_BINARY_WORD` looks for, so
#: a rewording fails a test instead of turning "install smartmontools" into "unknown".
SMART_MISSING_BINARY = ("smart: /usr/sbin/smartctl not found — install smartmontools and "
                        "give this user device read access (disk group) for SMART data")


@pytest.fixture
def smart_shape() -> Any:
    """`smart_shape('nvme')` / `smart_shape('ata_hdd')` -> the capture, parsed by the agent.

    Fixture-driven on purpose (AGENTS.md rule 7): the `smart` blocks a `/disks` test hands
    to the source are the recorded `smartctl` text run through the agent's own parser, so a
    field rename on either side of the wire fails a test instead of printing a dash.
    """
    agent = _load_host_agent()

    def shape(kind: str = "nvme") -> dict[str, Any]:
        text = (FIXTURES / f"smartctl_reference_{kind}.txt").read_text()
        return agent.parse_smart_output(text)

    return shape


@pytest.fixture
def disk_snapshot(agent_snapshots: dict[str, dict[str, Any]], smart_shape: Any) -> Any:
    """`build(host="driveripper", mode="healthy")` — a real capture with SMART attached.

    The fleet captures predate the agent's SMART pass, so there is no single capture with
    both halves. This assembles the two that exist: the physical disk rows come from the
    host capture verbatim, the `smart` blocks from the recorded smartmontools shapes, and
    the failure modes are the ones `host_agent.smart()` actually produces.

    Modes:

    * `healthy` — every disk reads, alternating the NVMe and ATA shapes
    * `no_binary` — every `smart` is null plus the agent's single "not found" line
    * `denied` — the last disk is null with its own permission line (SMART is installed)
    * `silent` — every `smart` is null with no error line (`--no-smart`, or an agent that
      predates the pass: the dashboard cannot tell why, and must not guess)
    * `failed` — one drive answering `FAILED`, the rest healthy
    """
    def build(host: str = "driveripper", *, mode: str = "healthy",
              mutate: Any = None) -> dict[str, Any]:
        import copy  # noqa: PLC0415 - every build starts from the untouched capture

        snapshot = copy.deepcopy(agent_snapshots[host])
        devices = (snapshot.get("disk_io") or {}).get("devices") or []
        errors: list[str] = list(snapshot.get("errors") or [])
        for index, device in enumerate(devices):
            if mode == "no_binary":
                device["smart"] = None
                continue
            if mode == "silent":
                device["smart"] = None
                continue
            if mode == "denied" and index == len(devices) - 1:
                device["smart"] = None
                continue
            shape = smart_shape(SMART_SHAPES[index % len(SMART_SHAPES)])
            block = dict(shape)
            # the identity in the capture is the disk the row is about; the shape files
            # describe one drive each, so the row keeps its own model and the counters
            # come from the shape
            block["model"] = block.get("model") or device.get("model")
            if mode == "failed" and index == 0:
                block["overall_health"] = "FAILED"
            device["smart"] = block
        if mode == "no_binary":
            errors.append(SMART_MISSING_BINARY)
        if mode == "denied" and devices:
            name = str(devices[-1].get("name"))
            errors.append(f"smart: {name}: Smartctl open device: /dev/{name} failed: "
                          "Permission denied")
        snapshot["errors"] = errors
        if mutate is not None:
            mutate(snapshot)
        return snapshot

    return build


@pytest.fixture
def metrics_text() -> Any:
    """metrics_text('gufo') / metrics_text('vllm') -> the captured exposition verbatim."""
    def load(engine: str) -> str:
        path = FIXTURES / f"metrics_{engine}.txt"
        if not path.exists():
            pytest.skip(f"missing fixture {path.name} — run scripts/capture-fixtures.sh --with-llm")
        return path.read_text()

    return load


@pytest.fixture
def rates() -> RateEstimator:
    return RateEstimator(span=10.0)