"""The host agent: stdlib-only, read-only by construction, and wrong on no box. Three things are asserted here: * parsing of /proc and /sys files exactly as Linux writes them (fake roots, so the weird cases are testable on any machine); * the collector produces a complete, non-crashing snapshot against the *real* /proc of whatever runs the tests, because that is the code that ships; * the HTTP surface answers GETs, needs a token when it is exposed, and has no write path. """ from __future__ import annotations import importlib.util import json import re import sys import threading import time import urllib.error import urllib.request from http.server import ThreadingHTTPServer from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] AGENT_PATH = ROOT / "agent" / "host_agent.py" spec = importlib.util.spec_from_file_location("host_agent", AGENT_PATH) assert spec and spec.loader host_agent = importlib.util.module_from_spec(spec) sys.modules["host_agent"] = host_agent spec.loader.exec_module(host_agent) linux_only = pytest.mark.skipif(not sys.platform.startswith("linux"), reason="reads /proc and /sys") # ---- source-level helpers ------------------------------------------------- # @pytest.mark.parametrize("lspci,expected", [ ("Advanced Micro Devices, Inc. [AMD/ATI] Navi 48 [Radeon AI PRO R9700]", "Radeon AI PRO R9700"), ("Advanced Micro Devices, Inc. [AMD/ATI] Strix Halo [Radeon Graphics / Radeon 8050S " "Graphics / Radeon 8060S Graphics]", "Radeon 8060S Graphics"), ("Advanced Micro Devices, Inc. [AMD/ATI] Phoenix1", "Phoenix1"), ("00:02.0 VGA compatible controller: Intel Corporation UHD Graphics 620", "UHD Graphics 620"), ("NVIDIA GB200", "NVIDIA GB200"), ("Vendor 1af1 Device 0001", "Vendor 1af1 Device 0001"), (None, None), ("", None), ]) def test_gpu_model_is_shortened_for_legends_without_losing_identity(lspci, expected) -> None: """`ATI` on a legend is worse than no name: it is true of every AMD GPU on the box.""" assert host_agent.short_gpu_model(lspci) == expected def test_gpu_model_is_capped_so_a_legend_cannot_blow_up() -> None: long_name = "X" * 200 assert len(host_agent.short_gpu_model(long_name)) <= 41 # 40 + ellipsis # ---- /proc and /sys parsing (fake roots) ---------------------------------- # def write_file(root: Path, relative: str, text: str) -> None: path = root / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text) def test_read_number_takes_the_first_field(tmp_path: Path) -> None: """/proc/uptime is `210036.18 9613750.86`; float() of the whole line is None.""" write_file(tmp_path, "proc/uptime", "210036.18 9613750.86\n") collector = host_agent.Collector(str(tmp_path)) assert collector.read_number("proc/uptime") == 210036.18 def test_read_number_handles_missing_and_junk_files(tmp_path: Path) -> None: collector = host_agent.Collector(str(tmp_path)) assert collector.read_number("proc/nope") is None write_file(tmp_path, "proc/junk", "\n") assert collector.read_number("proc/junk") is None def test_core_counts_dedupe_socket_and_core(tmp_path: Path) -> None: """One /proc/cpuinfo block per *thread*: counting 'core id' lines said '1 core'.""" blocks = [] for thread in range(8): # 2 sockets x 2 cores x 2 threads socket, core = thread % 2, (thread // 2) % 2 blocks.append(f"processor\t: {thread}\nphysical id\t: {socket}\ncore id\t\t: {core}\n") write_file(tmp_path, "proc/cpuinfo", "\n".join(blocks)) collector = host_agent.Collector(str(tmp_path)) assert collector.core_counts() == (8, 4) def test_core_counts_falls_back_to_smt_width_on_sysfs(tmp_path: Path) -> None: """Some kernels/vms give no physical id/core id; cpu0's sibling list still says SMT=2.""" write_file(tmp_path, "proc/cpuinfo", "processor\t: 0\nprocessor\t: 1\n") write_file(tmp_path, "sys/devices/system/cpu/cpu0/topology/thread_siblings_list", "0,1\n") collector = host_agent.Collector(str(tmp_path)) logical, physical = collector.core_counts() assert (logical, physical) == (2, 1) def test_core_counts_survive_a_kernel_with_no_hints(tmp_path: Path) -> None: collector = host_agent.Collector(str(tmp_path)) assert collector.core_counts()[1] is None, "unknown must be None, never 1" def test_cpu_modes_total_excludes_guest_ticks(tmp_path: Path) -> None: """/proc/stat's idle/iowait are already inside 'total'; guest is inside user on Linux.""" write_file(tmp_path, "proc/stat", "cpu 100 10 20 500 5 3 2 1 0 0\nintr 1\nctxt 2\nprocesses 3\n") collector = host_agent.Collector(str(tmp_path)) sample = collector.cpu()["modes"] assert sample["user"] == 100 and sample["idle"] == 500 assert sample["total"] == pytest.approx(100 + 10 + 20 + 500 + 5 + 3 + 2 + 1) # ---- SMART (smartctl) ------------------------------------------------------ # FIXTURES = Path(__file__).resolve().parent / "fixtures" NVME_CAPTURE = (FIXTURES / "smartctl_reference_nvme.txt").read_text() ATA_HDD_CAPTURE = (FIXTURES / "smartctl_reference_ata_hdd.txt").read_text() def read_fixture(text: str) -> str: """Strip the `#` provenance header the capture script writes above a fixture.""" return "\n".join(line for line in text.splitlines() if not line.startswith("#")) @pytest.mark.parametrize("path", sorted(FIXTURES.glob("smartctl_*.txt"))) def test_every_smartctl_capture_parses_without_inventing_numbers(path: Path) -> None: """A capture the parser cannot read must fail here, not on a `/disks` card. A serial in a fixture is a serial in the browser, so every captured serial is scrubbed.""" parsed = host_agent.parse_smart_output(read_fixture(path.read_text())) assert parsed["model"], f"{path.name}: no model identity parsed" # a recorded FAILED drive is a real drive failing, not a test to loosen: replace the # drive, or capture its truth into a purpose-built fixture and say why here. assert parsed["overall_health"] == "PASSED", f"{path.name}: health is {parsed['overall_health']!r}" assert parsed["power_on_hours"] is not None, f"{path.name}: no power-on hours parsed" assert parsed["serial"] is None or parsed["serial"].startswith("SCRUBBEDSERIAL"), path.name for key in host_agent.SMART_FIELDS: assert key in parsed, f"{path.name}: {key} missing from the parsed shape" def test_smart_output_parser_reads_the_nvme_health_log() -> None: parsed = host_agent.parse_smart_output(read_fixture(NVME_CAPTURE)) assert parsed["model"] == "EXAMPLE NVMe SSD 2TB" assert parsed["serial"] == "SCRUBBEDSERIAL1" assert parsed["temp_c"] == 38.0 assert parsed["power_on_hours"] == 8760.0, "the capture writes 8,760 — commas are notation" assert parsed["percentage_used"] == 1.0 # NVMe's media-error counter IS this drive family's offline-uncorrectable fact assert parsed["offline_uncorrectable"] == 0.0 # and what an NVMe does not publish stays None — reallocated/pending are ATA attributes assert parsed["reallocated_sector_ct"] is None assert parsed["current_pending_sector_ct"] is None def test_smart_output_parser_reads_the_ata_attribute_table() -> None: parsed = host_agent.parse_smart_output(read_fixture(ATA_HDD_CAPTURE)) assert parsed["model"] == "EXAMPLE-ST8000VN004" assert parsed["overall_health"] == "PASSED" assert parsed["temp_c"] == 31.0, "raw value (last column), not the normalised VALUE column" assert parsed["power_on_hours"] == 17520.0 assert parsed["reallocated_sector_ct"] == 8.0 assert parsed["current_pending_sector_ct"] == 3.0 assert parsed["offline_uncorrectable"] == 2.0 assert parsed["percentage_used"] is None, "this HDD publishes no wear percentage" def test_smart_output_parser_keeps_the_drives_own_verdict_words() -> None: assert host_agent.parse_smart_output( "SMART overall-health self-assessment test result: FAILED!\n")["overall_health"] == "FAILED" assert host_agent.parse_smart_output( "SMART overall-health self-assessment test result: UNKNOWN!\n")["overall_health"] == "UNKNOWN" def test_smart_failure_note_quotes_the_error_not_the_version_banner() -> None: """Caught on a real box: smartctl prints its banner first and the failure later, so 'first line of output' reported the copyright instead of Permission denied.""" banner = ("smartctl 7.5 2025-04-30 r5714 [x86_64-linux] (local build)\n" "Copyright (C) 2002-25, Bruce Allen, Christian Franke\n\n") denied = "Smartctl open device: /dev/nvme0 failed: Permission denied\n" assert "Permission denied" in host_agent.smart_failure_line(2, banner + denied, "") assert host_agent.smart_failure_line(7, "", "") == "exit status 7" assert host_agent.smart_failure_line(4, "some odd note\n", "") == "some odd note" # stderr counts too — smartctl puts some failures there and nowhere else assert "scsi hangup" in host_agent.smart_failure_line(2, banner, "scsi hangup") def test_smart_output_parser_returns_all_none_for_output_it_cannot_read() -> None: junk = ("smartctl 7.5 2025-04-30 r5714 (local build)\n" "Smartctl open device: /dev/sdb failed: Permission denied\n") parsed = host_agent.parse_smart_output(junk) assert set(parsed) == set(host_agent.SMART_FIELDS) assert all(value is None for value in parsed.values()), "a denied read is n/a, never 0" for empty in (None, ""): assert all(value is None for value in host_agent.parse_smart_output(empty).values()) def test_smart_output_parser_refuses_to_decode_vendor_hex_packed_raws() -> None: """Some vendors pack temperature into a hex raw whose byte layout is drive-specific. Decoding it from memory would be an invented number, so it stays None.""" parsed = host_agent.parse_smart_output( "194 Temperature_Celsius 0x0022 073 059 000 Old_age ALWAYS - 0x001f003c3430\n") assert parsed["temp_c"] is None def test_smart_attribute_rows_are_read_from_the_raw_column_not_the_line_end() -> None: """Every shape the deployed binary writes into RAW_VALUE, captured off real drives. Reading the last token of the row — the obvious way to read a column table — loses the temperature of every spinning disk on driveripper, because a Seagate ends its row with `35 (Min/Max 34/44)` and the token there is `44)`. A Western Digital packs its hours as `43894h+43m+32.519s`, which is hours first, minutes and seconds after: the drive's own number is the leading group, and the value the page prints is that number.""" parsed = host_agent.parse_smart_output( "190 Airflow_Temperature_Cel 0x0022 065 049 040 Old_age Always" " In_the_past 35 (Min/Max 34/44)\n" "194 Temperature_Celsius 0x0022 035 046 000 Old_age Always" " - 35 (0 13 0 0 0)\n" " 9 Power_On_Hours 0x0032 055 055 000 Old_age Always" " - 43894h+43m+32.519s\n" "197 Current_Pending_Sector 0x0012 100 100 000 Old_age Always" " - 2 (of 4 failed when written)\n") assert parsed["temp_c"] == 35.0 assert parsed["power_on_hours"] == 43894.0 assert parsed["current_pending_sector_ct"] == 2.0 def test_the_real_drives_answer_every_number_the_disks_page_prints() -> None: """`smartctl_driveripper_*.txt` is hardware truth, 16 drives, both shapes. The values move on every re-capture, so this pins that each number `/disks` prints is there and plausible rather than a particular figure — the check that a fleet which looks healthy in a synthetic fixture is also healthy when read from a real device. A drive that stops answering health fails here, and the honest response is to replace the drive or re-capture, never to loosen the assertion.""" real = sorted(FIXTURES.glob("smartctl_driveripper_*.txt")) assert real, "no real captures on disk: SMART_HOSTS= ./scripts/capture-fixtures.sh" for path in real: parsed = host_agent.parse_smart_output(read_fixture(path.read_text())) assert parsed["model"], f"{path.name}: no identity" hours = parsed["power_on_hours"] assert hours is not None and hours > 0 and hours == int(hours), ( f"{path.name}: power-on hours read as {hours!r}") temp = parsed["temp_c"] assert temp is not None and 10 <= temp <= 85, ( f"{path.name}: temperature read as {temp!r}") # The two shapes disagree the way the page renders them: no ATA drive on this fleet # publishes a wear percentage, every NVMe does, and neither answer is a 0. if "_sd" in path.name: assert parsed["percentage_used"] is None, f"{path.name}: an HDD has no wear percent" assert parsed["reallocated_sector_ct"] is not None, f"{path.name}: no ATA counters" else: assert parsed["percentage_used"] is not None, f"{path.name}: an NVMe must have wear" assert parsed["offline_uncorrectable"] is not None, ( f"{path.name}: the media-error log is an NVMe's uncorrectable fact") def make_fake_smartctl(tmp_path: Path, outputs: dict[str, str], exit_codes: dict[str, int] | None = None, slow: tuple[str, ...] = ()) -> str: """A stdlib fake `smartctl`: per-device stdout/exit code, plus an audit log of which devices were asked — so 'virtual disks are skipped' is an assertion, not a hope.""" data = tmp_path / "smart-out" data.mkdir(exist_ok=True) for device, text in outputs.items(): (data / device).write_text(text) for device, code in (exit_codes or {}).items(): (data / f"{device}.rc").write_text(str(code)) script = tmp_path / "bin" / "smartctl" script.parent.mkdir(exist_ok=True) script.write_text(f"""#!{sys.executable} import os, sys, time device = sys.argv[-1].rsplit("/", 1)[-1] data = os.environ["FAKE_SMARTCTL_DIR"] with open(os.path.join(data, "_called"), "a") as handle: handle.write(device + "\\n") if device in {list(slow)!r}: time.sleep(30) rc_path = os.path.join(data, device + ".rc") code = int(open(rc_path).read()) if os.path.exists(rc_path) else 0 out_path = os.path.join(data, device) text = open(out_path).read() if os.path.exists(out_path) else "" if code and not text: # real smartctl shape: banner first, failure later — the error note must quote # the failure, never the banner text = (f"smartctl 7.5 2025-04-30 r5714 [x86_64-linux] (local build)\\n" f"Copyright (C) 2002-25, Bruce Allen, Christian Franke\\n\\n" f"Smartctl open device: /dev/{{device}} failed: Permission denied\\n") sys.stdout.write(text) sys.exit(code) """) script.chmod(0o755) return str(script) def write_fake_disk(root: Path, name: str, model: str | None = None) -> None: write_file(root, f"sys/block/{name}/stat", "1 2 3 4 5 6 7 8 9 10 11") write_file(root, f"sys/block/{name}/queue/rotational", "0\n") write_file(root, f"sys/block/{name}/size", "2048\n") if model: write_file(root, f"sys/block/{name}/device/model", f"{model}\n") def fake_calls(tmp_path: Path) -> list[str]: called = tmp_path / "smart-out" / "_called" return called.read_text().split() if called.exists() else [] def test_smart_pass_attaches_parsed_health_per_disk(tmp_path, monkeypatch) -> None: root = tmp_path / "root" write_fake_disk(root, "nvme0n1", model="SYSFS nvme0n1") write_fake_disk(root, "sdb", model="SYSFS sdb") write_fake_disk(root, "loop0") # virtual: must never be asked write_fake_disk(root, "dm-0") monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector(str(root), smartctl=make_fake_smartctl(tmp_path, { "nvme0n1": read_fixture(NVME_CAPTURE), "sdb": read_fixture(ATA_HDD_CAPTURE), })) snap = collector.snapshot() assert snap["errors"] == [], snap["errors"] by_name = {device["name"]: device for device in snap["disk_io"]["devices"]} assert by_name["nvme0n1"]["smart"]["model"] == "EXAMPLE NVMe SSD 2TB" assert by_name["nvme0n1"]["smart"]["power_on_hours"] == 8760.0 assert by_name["sdb"]["smart"]["reallocated_sector_ct"] == 8.0 called = fake_calls(tmp_path) assert "loop0" not in called and "dm-0" not in called, "virtual entries are not physical disks" assert sorted(called) == ["nvme0n1", "sdb"] def test_smart_identity_falls_back_to_the_agents_own_disk_model(tmp_path, monkeypatch) -> None: """Some bridges refuse the identity block while still answering the health log.""" root = tmp_path / "root" write_fake_disk(root, "nvme0n1", model="SYSFS DISK MODEL") monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector( str(root), smartctl=make_fake_smartctl(tmp_path, {"nvme0n1": "Power On Hours: 42\n"})) device = collector.snapshot()["disk_io"]["devices"][0] assert device["smart"]["model"] == "SYSFS DISK MODEL" assert device["smart"]["power_on_hours"] == 42.0 assert device["smart"]["overall_health"] is None def test_missing_smartctl_binary_is_one_error_for_the_whole_pass(tmp_path, monkeypatch) -> None: root = tmp_path / "root" for device in ("nvme0n1", "sdb"): write_fake_disk(root, device) monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector(str(root), smartctl="/nonexistent/smartctl") snap = collector.snapshot() smart_errors = [e for e in snap["errors"] if e.startswith("smart:")] assert len(smart_errors) == 1, "one missing binary is one fact, not one per disk" assert "smartmontools" in smart_errors[0] # cross-module contract: `sources/disks.py` reads this word to answer # "is smartctl installed?" (three-way: yes / no / the agent does not say). Reword the # line here and that test fails, instead of the /disks page losing its explanation. assert "not found" in smart_errors[0] assert all(device["smart"] is None for device in snap["disk_io"]["devices"]) # the rest of the snapshot is intact — SMART is an add-on, not a dependency assert snap["disk_io"]["device_names"] == ["nvme0n1", "sdb"] def test_one_denied_device_nulls_only_that_disk(tmp_path, monkeypatch) -> None: root = tmp_path / "root" write_fake_disk(root, "nvme0n1") write_fake_disk(root, "sdb") monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector( str(root), smartctl=make_fake_smartctl(tmp_path, {"nvme0n1": read_fixture(NVME_CAPTURE)}, exit_codes={"sdb": 2})) snap = collector.snapshot() by_name = {device["name"]: device for device in snap["disk_io"]["devices"]} assert by_name["nvme0n1"]["smart"] is not None assert by_name["sdb"]["smart"] is None smart_errors = [e for e in snap["errors"] if e.startswith("smart:")] assert len(smart_errors) == 1 and "sdb" in smart_errors[0] and "Permission denied" in smart_errors[0] def test_a_wedged_device_costs_its_timeout_not_the_snapshot(tmp_path, monkeypatch) -> None: root = tmp_path / "root" write_fake_disk(root, "nvme0n1") write_fake_disk(root, "sdb") monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector( str(root), smartctl=make_fake_smartctl(tmp_path, {"nvme0n1": read_fixture(NVME_CAPTURE)}, slow=("sdb",))) collector.smart_timeout_seconds = 0.5 snap = collector.snapshot() by_name = {device["name"]: device for device in snap["disk_io"]["devices"]} assert by_name["nvme0n1"]["smart"] is not None assert by_name["sdb"]["smart"] is None assert any("timed out" in e for e in snap["errors"]) def test_the_pass_budget_skips_the_rest_with_one_error(tmp_path, monkeypatch) -> None: root = tmp_path / "root" for device in ("nvme0n1", "sdb", "sdc"): write_fake_disk(root, device) monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector( str(root), smartctl=make_fake_smartctl(tmp_path, {})) collector.smart_budget_seconds = 0.0 snap = collector.snapshot() assert all(device["smart"] is None for device in snap["disk_io"]["devices"]) assert any("budget" in e and "3 disk(s) skipped" in e for e in snap["errors"]) def test_unexecutable_smartctl_degrades_like_a_missing_one(tmp_path, monkeypatch) -> None: root = tmp_path / "root" write_fake_disk(root, "nvme0n1") monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) blocked = tmp_path / "not-executable" blocked.write_text("not a program\n") collector = host_agent.Collector(str(root), smartctl=str(blocked)) snap = collector.snapshot() assert all(device["smart"] is None for device in snap["disk_io"]["devices"]) assert any(e.startswith("smart:") for e in snap["errors"]) def wait_for_calls(tmp_path, expected: int, *, timeout: float = 5.0) -> list[str]: """Wait for the background refresher to have run `expected` passes.""" deadline = time.time() + timeout while time.time() < deadline: calls = fake_calls(tmp_path) if len(calls) >= expected: return calls time.sleep(0.02) return fake_calls(tmp_path) def test_smart_results_are_cached_and_an_expired_cache_refreshes_off_the_request_path( tmp_path, monkeypatch) -> None: """The system page polls every few seconds; smartctl must not re-run every poll. And when the cache *does* expire, the refresh runs on a background thread while the snapshot answers from the previous result: `SMARTCTL_BUDGET_SECONDS` (20) is five times the dashboard's `HOST_TIMEOUT_SECONDS` (4), so an inline refresh would turn the whole `system` source red every four minutes because of an add-on it does not use. """ root = tmp_path / "root" write_fake_disk(root, "nvme0n1") monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector( str(root), smartctl=make_fake_smartctl(tmp_path, {"nvme0n1": read_fixture(NVME_CAPTURE)})) first = collector.snapshot() second = collector.snapshot() assert first["disk_io"]["devices"][0]["smart"]["temp_c"] == 38.0 assert second["disk_io"]["devices"][0]["smart"] == first["disk_io"]["devices"][0]["smart"] assert fake_calls(tmp_path) == ["nvme0n1"], "one pass, then the cache" collector.smart_cache_seconds = 0.0 # an expired cache re-runs the pass… served = collector.snapshot() assert served["disk_io"]["devices"][0]["smart"] == first["disk_io"]["devices"][0]["smart"] assert wait_for_calls(tmp_path, 2) == ["nvme0n1", "nvme0n1"], "…in the background" def test_an_expired_cache_never_stalls_the_snapshot_on_a_wedged_disk(tmp_path, monkeypatch) -> None: """The regression that matters: a disk that takes 30 s must cost a poll nothing. Asserted on wall-clock because the whole point is the request's latency — the agent answers with the last SMART it had while the refresher sits in `smartctl`'s timeout. """ root = tmp_path / "root" write_fake_disk(root, "nvme0n1") write_fake_disk(root, "sdb") monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector( str(root), smartctl=make_fake_smartctl(tmp_path, {"nvme0n1": read_fixture(NVME_CAPTURE), "sdb": "Power On Hours: 1\n"}, slow=("sdb",))) collector.smart_timeout_seconds = 1.0 # the slow device costs ~1 s per pass first = collector.snapshot() assert any("timed out" in line for line in first["errors"]) temp = first["disk_io"]["devices"][0]["smart"]["temp_c"] collector.smart_cache_seconds = 0.0 started = time.perf_counter() served = collector.snapshot() elapsed = time.perf_counter() - started assert served["disk_io"]["devices"][0]["smart"]["temp_c"] == temp assert elapsed < 0.5, f"a wedged disk cost the snapshot {elapsed:.2f}s" # the refresher is single-flight: a second poll while it runs gets the cache too started = time.perf_counter() collector.snapshot() assert time.perf_counter() - started < 0.5 wait_for_calls(tmp_path, 4) # let the background pass finish before teardown def test_no_smart_flag_stays_silent_and_null(tmp_path, monkeypatch) -> None: root = tmp_path / "root" write_fake_disk(root, "nvme0n1") monkeypatch.setenv("FAKE_SMARTCTL_DIR", str(tmp_path / "smart-out")) collector = host_agent.Collector(str(root), smartctl=make_fake_smartctl(tmp_path, {}), smart=False) snap = collector.snapshot() assert snap["errors"] == [] assert snap["disk_io"]["devices"][0]["smart"] is None assert fake_calls(tmp_path) == [] def test_an_unexpected_smart_crash_degrades_the_addon_not_the_snapshot(tmp_path, monkeypatch) -> None: collector = host_agent.Collector("/", smart=False) monkeypatch.setattr(collector, "smart_enabled", True) monkeypatch.setattr(collector, "smart", lambda names: (_ for _ in ()).throw(RuntimeError("boom"))) snap = collector.snapshot() assert any(e.startswith("smart:") and "boom" in e for e in snap["errors"]) assert snap["disk_io"]["devices"], "the disk rows themselves come from /sys, not smartctl" def test_smartctl_captures_carry_no_unscrubbed_serials() -> None: """The capture script scrubs; this fails the day it stops doing its job. The NVMe namespace EUI-64 is on the list because it embeds the serial.""" for path in sorted(FIXTURES.glob("smartctl_*.txt")): for line in path.read_text().splitlines(): lowered = line.lower() if "serial number" in lowered or "wwn" in lowered or "eui" in lowered: assert "SCRUBBED" in line, f"{path.name}: unscrubbed identifier: {line!r}" # ---- the real collector --------------------------------------------------- # @linux_only def test_snapshot_against_the_live_system_is_complete() -> None: """Real /proc and /sys, but only of the box running pytest — no network, so this is not `live` (see tests/test_live_fleet.py): it belongs in the default run.""" collector = host_agent.Collector("/") snap = collector.snapshot() for key in ("schema", "ts", "host", "cpu", "memory", "filesystems", "disk_io", "network", "gpus", "temperatures", "processes", "errors"): assert key in snap, f"{key} missing from a real snapshot" assert snap["host"]["hostname"] assert snap["cpu"]["modes"]["total"] > 0 assert len(snap["cpu"]["cores"]) >= 1 assert snap["memory"]["total_bytes"] > 0 assert snap["filesystems"], "the box it runs on has at least one mounted filesystem" # SMART notes are allowed here: the box running pytest may have no smartctl or no # device read access, and both are documented degradations, never snapshot failures. hard_errors = [e for e in snap["errors"] if not e.startswith("smart")] assert hard_errors == [], f"degraded sections must be empty on a healthy box: {hard_errors}" def test_captured_fixture_has_the_shape_the_dashboard_depends_on(agent_snapshot) -> None: """Ground truth from driveripper: if the agent's schema drifts, this fails first.""" assert agent_snapshot["schema"] == host_agent.SCHEMA_VERSION host = agent_snapshot["host"] assert host["hostname"] == "driveripper" assert (host["cpu_cores_physical"], host["cpu_cores_logical"]) == (24, 48) assert host["uptime_seconds"] > 0 gpu = agent_snapshot["gpus"][0] assert gpu["model"] == "Radeon AI PRO R9700" assert "model_full" in gpu and "Navi 48" in gpu["model_full"] assert gpu["vram_total_bytes"] > 0 # the dashboard renders n/a for anything a driver omits — it must never invent a 0 assert gpu["busy_pct"] is None or 0.0 <= gpu["busy_pct"] <= 100.0 @linux_only def test_uptime_and_boot_ts_are_reported() -> None: """Both are needed: uptime for the table, boot_ts to detect a counter-reset reboot.""" info = host_agent.Collector("/").host_info() assert info["uptime_seconds"] and info["uptime_seconds"] > 0 assert info["boot_ts"] and info["boot_ts"] <= time.time() assert abs((time.time() - info["boot_ts"]) - info["uptime_seconds"]) < 60 assert info["cpu_cores_logical"] and info["cpu_cores_physical"] @linux_only def test_snapshot_is_repeatable_and_does_not_deadlock() -> None: """snapshot() reads processes() internally; a non-reentrant lock used to hang it.""" collector = host_agent.Collector("/") first = collector.snapshot() second = collector.snapshot(top=0) assert first["processes"]["total"] > 0 assert second["processes"]["top_cpu"] == [], "top=0 must skip the expensive scan" @linux_only def test_disk_io_comes_from_sysfs_not_proc_diskstats() -> None: """/proc/diskstats field order has moved between kernels; /sys/block//stat has not.""" devices = host_agent.Collector("/").disk_io()["devices"] assert devices for device in devices: assert device["reads_total"] is None or device["reads_total"] >= 0 assert device["size_bytes"] is None or device["size_bytes"] > 0 # ---- HTTP surface --------------------------------------------------------- # @pytest.fixture def agent_server(): """Run the real handler on an ephemeral loopback port; yield (get, port).""" servers: list[ThreadingHTTPServer] = [] def start(token: str | None = None) -> tuple[str, int]: host_agent.Handler.collector = host_agent.Collector("/") host_agent.Handler.token = token httpd = ThreadingHTTPServer(("127.0.0.1", 0), host_agent.Handler) servers.append(httpd) threading.Thread(target=httpd.serve_forever, daemon=True).start() return f"http://127.0.0.1:{httpd.server_address[1]}", httpd.server_address[1] yield start for httpd in servers: httpd.shutdown() httpd.server_close() def _get(url: str, headers: dict[str, str] | None = None) -> tuple[int, dict]: request = urllib.request.Request(url, headers=headers or {}) try: with urllib.request.urlopen(request, timeout=5) as response: return response.status, json.loads(response.read()) except urllib.error.HTTPError as error: body = error.read() try: return error.code, json.loads(body) except json.JSONDecodeError: # pragma: no cover - handler always sends JSON return error.code, {"raw": body.decode(errors="replace")} def test_health_stays_up_while_smart_is_degraded(agent_server, tmp_path) -> None: """Phase-09 criterion: SMART failures are add-on notes, never an agent outage.""" root = tmp_path / "root" write_fake_disk(root, "nvme0n1") collector = host_agent.Collector(str(root), smartctl="/nonexistent/smartctl") base, _port = agent_server() status, health = _get(base + "/health") assert status == 200 and health["ok"] is True handler = host_agent.Handler handler.collector = collector status, snap = _get(base + "/snapshot") assert status == 200 assert snap["disk_io"]["devices"][0]["smart"] is None assert any(e.startswith("smart:") for e in snap["errors"]) def test_health_info_schema_snapshot_all_answer(agent_server) -> None: base, _port = agent_server() for path in ("/health", "/info", "/schema", "/snapshot"): status, body = _get(base + path) assert status == 200, path assert body["ok"] is True, path def test_snapshot_query_param_controls_the_process_scan(agent_server) -> None: base, _port = agent_server() _status, body = _get(base + "/snapshot?top=0") assert body["processes"]["top_cpu"] == [] _status, full = _get(base + "/snapshot?top=3") assert len(full["processes"]["top_cpu"]) <= 3 def test_unknown_path_is_404_not_a_snapshot(agent_server) -> None: base, _port = agent_server() status, body = _get(base + "/etc/shadow") assert status == 404 assert body["ok"] is False def test_token_is_required_when_configured_but_health_stays_open(agent_server) -> None: base, _port = agent_server(token="s3cret") assert _get(base + "/snapshot")[0] == 401 assert _get(base + "/snapshot", {"Authorization": "Bearer wrong"})[0] == 401 assert _get(base + "/snapshot", {"Authorization": "Bearer s3cret"})[0] == 200 # /health stays open so a monitoring probe needs no secret assert _get(base + "/health")[0] == 200 def test_agent_has_no_write_path(agent_server) -> None: """No do_POST/do_PUT exists: read-only is structural, not a policy toggle.""" base, _port = agent_server() request = urllib.request.Request(base + "/snapshot", data=b"{}", method="POST") with pytest.raises(urllib.error.HTTPError) as excinfo: urllib.request.urlopen(request, timeout=5) assert excinfo.value.code in (405, 501) def test_refuses_to_bind_off_loopback_without_a_token() -> None: """0.0.0.0 with no token would hand host inventory to the whole LAN.""" assert host_agent.main(["--host", "0.0.0.0", "--port", "0"]) == 2 def test_once_mode_prints_json_and_exits(capsys) -> None: assert host_agent.main(["--once"]) == 0 printed = capsys.readouterr().out assert json.loads(printed)["ok"] is True def test_agent_is_dependency_free() -> None: """The agent must run on a stock distro python: stdlib imports, and few of them. Two checks, because they fail for different reasons: anything outside the stdlib means a `pip install` on every monitored host (deployment pain), while a growing footprint means the agent can break on an older interpreter than the dashboard's own Python. """ text = AGENT_PATH.read_text() imports = re.findall(r"^(?:import|from)\s+([a-zA-Z0-9_.]+)", text, re.M) # indented (function-local) imports count too — subprocess is loaded lazily on purpose imports += re.findall(r"^\s+(?:import|from)\s+([a-zA-Z0-9_.]+)", text, re.M) top = {name.split(".")[0] for name in imports} third_party = top - set(sys.stdlib_module_names) - {"host_agent"} assert not third_party, f"agent grew a non-stdlib import: {third_party}" assert top <= {"__future__", "argparse", "collections", "dataclasses", "http", "json", "os", "pathlib", "re", "socket", "subprocess", "threading", "time", "typing"}, f"agent import footprint grew: {sorted(top)}" # ---- how the token is supplied ------------------------------------------- # def test_token_file_wins_over_the_command_line(tmp_path) -> None: path = tmp_path / "token" path.write_text("from-file\n", encoding="utf-8") token, error = host_agent.resolve_token("from-flag", str(path)) assert error is None assert token == "from-file", "argv is world-readable in ps; the file has to win" def test_a_missing_token_file_is_fatal_and_says_which_one(tmp_path) -> None: missing = tmp_path / "nope" token, error = host_agent.resolve_token("", str(missing)) assert token is None assert error and str(missing) in error, "a silent fallback to no-token would expose the host" def test_an_empty_token_file_is_fatal(tmp_path) -> None: path = tmp_path / "token" path.write_text(" \n", encoding="utf-8") token, error = host_agent.resolve_token("ignored", str(path)) assert token is None and error is not None def test_no_token_at_all_is_allowed_on_loopback(tmp_path) -> None: token, error = host_agent.resolve_token("", "") assert (token, error) == (None, None) def test_main_refuses_to_bind_off_loopback_without_a_token(tmp_path) -> None: code = host_agent.main(["--host", "0.0.0.0", "--token-file", str(tmp_path / "missing")]) assert code == 2, "the agent must not start exposed and unauthenticated"