"""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)", ) parser.addoption( "--store-perf", action="store_true", default=False, help="run tests marked `store_perf`: the phase-14 timing measurements against the " "seeded production-shaped store (they skip otherwise — they cost minutes, not ms)", ) def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: # Two opt-ins, one shape: a marker whose tests are real but expensive, skipped by default # so `uv run pytest` stays fast and hermetic. `--store-perf` is the phase-14 half of the # rule `--live` established; `scripts/store_perf_probe.py` is the CLI for the same numbers. skips = { "live": pytest.mark.skip( reason="needs --live and a reachable target (see tests/test_live_fleet.py)"), "store_perf": pytest.mark.skip( reason="needs --store-perf (a seeded ~6 M-row store; see tests/store_perf.py)"), } enabled = { "live": bool(config.getoption("--live")), "store_perf": bool(config.getoption("--store-perf")), } for item in items: for marker, skip in skips.items(): if not enabled[marker] and marker 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(scope="session") def host_agent_module() -> Any: """The stdlib agent as a module, for the tests that run it outside `tests/test_agent.py`. Loaded by path because it is one file that ships to every host and must not become a package (see `_load_host_agent`). The phase-15 collector tests need one module object for both the bundle they load and the handler they serve, so the TLS dial runs the artifact the fleet really runs rather than a look-alike. """ return _load_host_agent() @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) # ---- phase 14: the production-shaped store the hot-path budgets measure ---- # # # Here rather than in tests/test_store.py because two files price the same budgets — the store # paths in `test_store.py`, the HTTP paths in `test_api.py` — and a second seeded 6.4 M-row file # (or, worse, a second *differently shaped* one) would make the two sets of numbers describe two # databases. Session-scoped because seeding costs ~25 s, and cached on disk (`data/perf/`, see # `store_perf.cache_dir`) because the before/after has to be the same physical file both times. @pytest.fixture(scope="session") def perf_store_path() -> Path: """The cached seeded store, built on first use and re-verified on every reuse. Opt-in tests only: nothing in the default run asks for this fixture, so `uv run pytest` never pays for the seed. `DASH_PERF_STORE_DIR` relocates the file (see `store_perf.cache_dir`). """ import store_perf # noqa: PLC0415 - only the phase-14 tests seed a store try: return store_perf.ensure_seeded(log=lambda line: print(line, flush=True)) except OSError as exc: # pragma: no cover - only a read-only checkout hits this pytest.skip(f"cannot build the seeded store ({exc}) — set DASH_PERF_STORE_DIR " "to a writable directory to run the phase-14 measurements") @pytest.fixture def perf_store(perf_store_path: Path) -> Any: """The seeded store, opened per test and closed after (one handle per test, one file).""" import store_perf # noqa: PLC0415 - see perf_store_path store = store_perf.open_store(perf_store_path) try: yield store finally: store.close()