dashboards/tests/test_agent.py

"""The host agent: stdlib-only, read-only by construction, and wrong on no box.

Four 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;
  * the TLS mode (phase 15) answers a real handshake only to a client certificate the bundle's
    root signed and the bundle's denylist does not name, with certificates minted by the
    dashboard's own `certs.py` — there is no second generator to disagree with.
"""
from __future__ import annotations

import http.client
import importlib.util
import json
import re
import socket
import ssl
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
from datetime import UTC, datetime, timedelta
from http.server import ThreadingHTTPServer
from pathlib import Path
from types import SimpleNamespace
from typing import Any, cast

import pytest
from cryptography import x509
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey
from cryptography.hazmat.primitives.hashes import SHA256
from cryptography.hazmat.primitives.serialization import (
    Encoding,
    NoEncryption,
    PrivateFormat,
    load_pem_private_key,
)
from cryptography.x509.oid import NameOID

from dashboards import certs

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=<alias> ./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"])


class _TimeProxy:
    """The agent's `time` module with one instrumented function: `monotonic`.

    Everything else (`time()`, `sleep()`) stays the stdlib one, so patching the clock cannot
    leak into the rest of the snapshot or into pytest.
    """

    def __init__(self, clock: FakeSmartPassClock) -> None:
        self._clock = clock

    def monotonic(self) -> float:
        return self._clock.now

    def __getattr__(self, name: str) -> Any:
        """Everything the agent's `time` module offers besides `monotonic`, unchanged."""
        return getattr(time, name)


class FakeSmartPassClock:
    """A clock plus a `smartctl` spawn the test drives, so a **deadline** is assertable.

    The budget arithmetic in `_smart_pass` is invisible to a fake binary: `subprocess.run`'s
    `timeout` never reaches the child, and burning 80 real seconds to watch a 16-disk pass
    truncate is not a measurement anybody keeps. So this drives the two things the pass
    actually reads — `time.monotonic()` and `subprocess.run()` — and records the timeout each
    device was given, which is `min(deadline - now, 5 s)`: the pass's own view of how much
    budget was left when it asked. Both patches go on `host_agent`'s own module globals
    through proxies, never onto the stdlib modules themselves.
    """

    def __init__(self, *, seconds_per_device: float, start: float = 10_000.0) -> None:
        self.seconds_per_device = seconds_per_device
        self.now = start
        self.calls: list[tuple[str, float]] = []
        self.output = read_fixture(NVME_CAPTURE)

    def run(self, argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
        device = str(argv[-1]).rsplit("/", 1)[-1]
        timeout = float(kwargs["timeout"])
        self.calls.append((device, timeout))
        self.now += self.seconds_per_device
        return subprocess.CompletedProcess(argv, 0, self.output, "")

    def install(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr(host_agent, "time", _TimeProxy(self))
        monkeypatch.setattr(host_agent, "subprocess", SimpleNamespace(
            run=self.run,
            TimeoutExpired=subprocess.TimeoutExpired,
            CompletedProcess=subprocess.CompletedProcess))


# driveripper's real inventory: 8 NVMe + 8 spinning, which is the shape the budget scales for
SHELF_DISKS = [f"nvme{i}n1" for i in range(8)] + [f"sd{c}" for c in "abcdefgh"]


def write_disk_shelf(root: Path, names: list[str]) -> None:
    for name in names:
        write_fake_disk(root, name, model=f"SYSFS {name}")


def test_the_pass_budget_scales_with_the_disk_count(tmp_path, monkeypatch) -> None:
    """A 16-disk pass gets 5 s x 16, not the 20 s floor — because a truncated pass nulls the
    tail drives on every single poll (the bug that put this on the operator's screen on
    2026-10-19). The clock here burns 1.5 s per device, so the pass costs 24 s of budget:
    over the floor, far under the scale. With the flat budget this test truncates.
    """
    root = tmp_path / "root"
    write_disk_shelf(root, SHELF_DISKS)
    clock = FakeSmartPassClock(seconds_per_device=1.5)
    clock.install(monkeypatch)
    collector = host_agent.Collector(str(root))
    snap = collector.snapshot()

    asked = [name for name, _ in clock.calls]
    assert asked == snap["disk_io"]["device_names"], "every disk got its smartctl run"
    assert len(asked) == 16
    assert clock.seconds_per_device * len(asked) > host_agent.SMARTCTL_BUDGET_SECONDS, (
        "this pass costs more than the 20 s floor — that is what makes it a scaling test")
    assert [e for e in snap["errors"] if "budget" in e] == [], snap["errors"]
    assert all(device["smart"] is not None for device in snap["disk_io"]["devices"])
    # the *late* devices still hold the whole per-device bound, which is the point of scaling
    assert clock.calls[-1][1] == host_agent.SMARTCTL_TIMEOUT_SECONDS
    assert clock.calls[0][1] == host_agent.SMARTCTL_TIMEOUT_SECONDS


def test_the_pass_budget_floor_is_twenty_seconds(tmp_path, monkeypatch) -> None:
    """Scaling is `max(20 s, 5 s x disks)`, so one or two disks still get the 20 s floor and
    not a 5 s or 10 s window. The second device's timeout is the visible half: with a 20 s
    floor it is `min(20 - 18, 5)` = 2 s, where a budget of 5 s x 2 would have skipped it and a
    30 s floor would still have handed out the full 5 s.
    """
    root = tmp_path / "root"
    write_disk_shelf(root, ["nvme0n1", "sda"])
    clock = FakeSmartPassClock(seconds_per_device=18.0)
    clock.install(monkeypatch)
    collector = host_agent.Collector(str(root))
    snap = collector.snapshot()

    assert [name for name, _ in clock.calls] == ["nvme0n1", "sda"]
    assert clock.calls[0][1] == host_agent.SMARTCTL_TIMEOUT_SECONDS
    assert clock.calls[1][1] == pytest.approx(
        host_agent.SMARTCTL_BUDGET_SECONDS - clock.seconds_per_device)
    assert [e for e in snap["errors"] if "budget" in e] == []
    assert all(device["smart"] is not None for device in snap["disk_io"]["devices"])


def test_the_pass_budget_error_line_names_the_scaled_figure(tmp_path, monkeypatch) -> None:
    """When a scaled pass does run out, the operator's one line says *which* budget ran out:
    80 s on a 16-disk shelf, never the 20 s floor. A test that asserted only "a budget error
    appears" would pass on the pre-scaling code; this one cannot.
    """
    root = tmp_path / "root"
    write_disk_shelf(root, SHELF_DISKS)
    clock = FakeSmartPassClock(seconds_per_device=6.0)
    clock.install(monkeypatch)
    collector = host_agent.Collector(str(root))
    snap = collector.snapshot()

    scaled = max(host_agent.SMARTCTL_BUDGET_SECONDS,
                 host_agent.SMARTCTL_TIMEOUT_SECONDS * len(SHELF_DISKS))
    assert scaled == 80.0
    # 6 s per device: devices 1-14 fit inside 80 s (78 s spent), the 15th finds the deadline
    # gone, and its sibling is skipped with it.
    assert len(clock.calls) == 14, [name for name, _ in clock.calls]
    assert clock.calls[-1][1] == pytest.approx(scaled - 6.0 * 13)
    budget_lines = [e for e in snap["errors"] if "budget" in e]
    assert len(budget_lines) == 1, snap["errors"]
    assert budget_lines[0] == (f"smart: pass budget of {scaled:.0f}s exhausted, "
                               f"2 disk(s) skipped")
    assert f"{host_agent.SMARTCTL_BUDGET_SECONDS:.0f}s" not in budget_lines[0]
    by_name = {device["name"]: device for device in snap["disk_io"]["devices"]}
    assert [by_name[name]["smart"] is None for name in SHELF_DISKS[:14]] == [False] * 14
    assert [by_name[name]["smart"] is None for name in SHELF_DISKS[14:]] == [True, True]


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/<dev>/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
        # Cleared on purpose: `Handler` is process-wide state, so a token-mode server started
        # after a TLS test in this file must not inherit that test's bundle.
        host_agent.Handler.bundle = None
        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(monkeypatch) -> None:
    """0.0.0.0 with no token would hand host inventory to the whole LAN."""
    monkeypatch.delenv("AGENT_TOKEN", raising=False)
    monkeypatch.delenv("AGENT_CERT_BUNDLE", raising=False)
    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", "contextlib", "dataclasses", "http",
                   "json", "os", "pathlib", "re", "shutil", "socket", "ssl", "subprocess",
                   "tempfile", "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, monkeypatch) -> None:
    monkeypatch.delenv("AGENT_TOKEN", raising=False)
    monkeypatch.delenv("AGENT_CERT_BUNDLE", raising=False)
    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"


# ---- TLS mode: one installed bundle, and its client certificate is the auth --- #
#
# Everything below runs a real `ssl` handshake against the real handler, because the claims
# here are handshake claims: "optional at the handshake, enforced by the handler" is exactly
# the kind of thing that reads fine and then makes `/health` unprobeable. The certificates come
# from `dashboards.certs` over a dict store, so the agent is tested against the artifact the
# dashboard actually signs rather than against a second generator that could drift from it.


#: The agent module is loaded by `importlib` (it is one file with no importable package), so
#: pyright sees `host_agent` as an untyped module and rejects `host_agent.TlsBundle` in a type
#: position. This alias is the honest spelling: the object is that class, and the tests below
#: assert `isinstance` where the claim actually matters.
Bundle = Any


class DictMeta:
    """The three-method `certs.MetaStore` slice, backed by a dict (the `test_certs.py` idiom)."""

    def __init__(self) -> None:
        self.rows: dict[str, str] = {}

    def get_meta(self, key: str) -> str | None:
        return self.rows.get(key)

    def set_meta(self, key: str, value: str) -> None:
        self.rows[key] = value

    def delete_meta(self, key: str) -> int:
        return 1 if self.rows.pop(key, None) is not None else 0


def sign_client_leaf(state: certs.CertsState, cn: str, where: Path) -> SimpleNamespace:
    """A second client identity under the dashboard's own root, so it verifies and can be revoked.

    `certs.py` mints exactly one client certificate on purpose (the dashboard has one identity),
    so the revoked-peer and wrong-authority cases need a leaf signed by hand. The serial comes
    back because the denylist names a serial, not a name.
    """
    root_key = cast(EllipticCurvePrivateKey,
                    load_pem_private_key(state.root_key.encode("utf-8"), password=None))
    root_cert = x509.load_pem_x509_certificate(state.root_cert.encode("utf-8"))
    key = ec.generate_private_key(ec.SECP256R1())
    moment = datetime.now(UTC)
    cert = (
        x509.CertificateBuilder()
        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]))
        .issuer_name(root_cert.subject)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(moment - timedelta(days=1))
        .not_valid_after(moment + timedelta(days=365))
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
        .sign(root_key, SHA256())
    )
    cert_path = where / f"{cn}-cert.pem"
    key_path = where / f"{cn}-key.pem"
    cert_path.write_bytes(cert.public_bytes(Encoding.PEM))
    key_path.write_bytes(key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()))
    return SimpleNamespace(cert=cert_path, key=key_path, serial=cert.serial_number)


@pytest.fixture(scope="module")
def tls_fleet(tmp_path_factory: pytest.TempPathFactory) -> SimpleNamespace:
    """A dashboard CA, its client pair, one issued agent bundle, and one foreign authority."""
    base = tmp_path_factory.mktemp("agent-tls")
    store = DictMeta()
    state = certs.ensure_certs(store)
    state = certs.issue_agent_cert(store, state, "localhost", ["127.0.0.1", "localhost"])
    entry = state.agent("localhost")
    assert entry is not None
    for name in ("dashboard", "foreign"):
        (base / name).mkdir()
    dashboard = certs.write_material(state, base / "dashboard")
    foreign = certs.write_material(certs.ensure_certs(DictMeta()), base / "foreign")
    return SimpleNamespace(
        root_cert=state.root_cert,
        bundle=entry.bundle,
        bundle_rev=entry.bundle_rev,
        client_cert=str(dashboard.client_cert),
        client_key=str(dashboard.client_key),
        foreign_cert=str(foreign.client_cert),
        foreign_key=str(foreign.client_key),
        stolen=sign_client_leaf(state, "stolen", base),
    )


def write_bundle(tmp_path: Path, text: str) -> Path:
    """An installed bundle: one PEM file, exactly as `install-agent.sh` will place it."""
    path = tmp_path / "dashboards-agent.certs"
    path.write_text(text, encoding="ascii")
    return path


def load_bundle(tmp_path: Path, text: str) -> Bundle:
    path = write_bundle(tmp_path, text)
    bundle, error = host_agent.TlsBundle.load(str(path))
    assert error is None and bundle is not None
    return bundle


def client_context(ca: str, cert: str | None = None, key: str | None = None) -> ssl.SSLContext:
    """A client that trusts `ca`, and presents a client certificate only when given one."""
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    context.load_verify_locations(cadata=ca)
    if cert and key:
        context.load_cert_chain(cert, key)
    return context


@pytest.fixture
def tls_agent_server():
    """The real handler on a TLS-wrapped loopback port; yields the base URL."""
    servers: list[ThreadingHTTPServer] = []

    def start(bundle: Bundle, token: str | None = None) -> str:
        host_agent.Handler.collector = host_agent.Collector("/", smart=False)
        host_agent.Handler.token = token
        host_agent.Handler.bundle = bundle
        httpd = ThreadingHTTPServer(("127.0.0.1", 0), host_agent.Handler)
        httpd.socket = bundle.context.wrap_socket(httpd.socket, server_side=True)
        servers.append(httpd)
        threading.Thread(target=httpd.serve_forever, daemon=True).start()
        return f"https://127.0.0.1:{httpd.server_address[1]}"

    yield start
    for httpd in servers:
        httpd.shutdown()
        httpd.server_close()
    host_agent.Handler.bundle = None


def _json_or_raw(body: bytes) -> dict:
    """JSON when the handler answered with JSON; the raw text for the status page."""
    try:
        return json.loads(body)
    except json.JSONDecodeError:
        return {"raw": body.decode(errors="replace")}


def _https_get(url: str, context: ssl.SSLContext) -> tuple[int, dict]:
    request = urllib.request.Request(url)
    try:
        with urllib.request.urlopen(request, timeout=15, context=context) as response:
            return response.status, _json_or_raw(response.read())
    except urllib.error.HTTPError as error:
        return error.code, _json_or_raw(error.read())


def test_the_agent_and_the_dashboard_agree_on_the_bundle_lines() -> None:
    """The agent parses a PEM by text with stdlib only, so the two literals are the contract."""
    assert host_agent.BUNDLE_REV_LINE == certs.BUNDLE_REV_LINE
    assert host_agent.REVOKED_LINE == certs.REVOKED_LINE


def test_the_bundle_lines_are_read_as_written() -> None:
    text = "# BUNDLE-REV 12\n# REVOKED 5\n# REVOKED 77\n"
    assert host_agent.bundle_revision(text) == 12
    assert host_agent.bundle_revoked(text) == frozenset({"5", "77"})
    assert host_agent.bundle_revision("nothing at all\n") is None
    assert host_agent.bundle_revision("# BUNDLE-REV not-a-number\n") is None
    assert host_agent.bundle_revoked("# REVOKED\n") == frozenset()


@pytest.mark.parametrize("revoked, peer_hex, expected", [
    # the shape a real `x509.random_serial_number()` serial arrives in, both spellings
    ("57989923439447977599989997426575336710275559875",
     "A285BAC84EB9C5FC282EF7CDB8D962F38BF71C3", True),
    ("12648430", "C0FFEE", True),
    ("14", "0E", True),
    ("0E", "0E", True),
    ("15", "0E", False),
    ("0", "0E", False),
])
def test_the_denylist_matches_a_serial_whichever_way_it_is_written(revoked: str,
                                                                   peer_hex: str,
                                                                   expected: bool) -> None:
    """`getpeercert()` says hex, `certs.py` writes decimal, and the agent must meet in the middle.

    An agent comparing one spelling to the other accepts a revoked client forever: the operator
    clicked revoke, `/certs` says revoked, and the fleet keeps trusting the stolen identity.
    """
    cert = {"serialNumber": peer_hex} if peer_hex else {}
    assert host_agent.serial_is_revoked(cert, frozenset({revoked})) is expected


def test_a_certificate_with_no_serial_is_never_treated_as_revoked() -> None:
    assert host_agent.serial_is_revoked({}, frozenset({"1", "0"})) is False


def test_resolve_bundle_reports_the_revision_and_the_denylist_of_a_real_bundle(
        tls_fleet: SimpleNamespace, tmp_path: Path) -> None:
    text = tls_fleet.bundle + f"{host_agent.REVOKED_LINE}{tls_fleet.stolen.serial}\n"
    context, revision, revoked, error = host_agent.resolve_bundle(str(write_bundle(tmp_path, text)))
    assert error is None
    assert context is not None and isinstance(context, ssl.SSLContext)
    assert revision == tls_fleet.bundle_rev
    assert revoked == frozenset({str(tls_fleet.stolen.serial)})


@pytest.mark.parametrize("shape", ["missing", "empty", "no-key", "no-revision"])
def test_a_bundle_that_cannot_be_loaded_is_an_error_naming_the_file(
        tls_fleet: SimpleNamespace, tmp_path: Path, shape: str) -> None:
    """Every failure is a message the operator can act on, and none of them is "no TLS today"."""
    if shape == "missing":
        path = tmp_path / "absent.certs"
    elif shape == "empty":
        path = write_bundle(tmp_path, "   \n")
    elif shape == "no-key":
        body = [block for block in re.findall(
            r"-----BEGIN [A-Z0-9 ]+-----.*?-----END [A-Z0-9 ]+-----\n", tls_fleet.bundle, re.S)
            if block.startswith("-----BEGIN CERTIFICATE")]
        path = write_bundle(tmp_path, "# BUNDLE-REV 1\n" + "".join(body))
    else:
        path = write_bundle(tmp_path, tls_fleet.bundle.replace("# BUNDLE-REV 1\n", ""))
    context, revision, revoked, error = host_agent.resolve_bundle(str(path))
    assert context is None and revision == 0 and revoked == frozenset()
    assert error is not None and ("absent" in error if shape == "missing" else str(path) in error)
    if shape == "no-revision":
        assert "BUNDLE-REV" in error


def test_the_unpacked_pems_are_gone_once_the_context_has_them(
        tls_fleet: SimpleNamespace, tmp_path: Path, monkeypatch) -> None:
    """The unit's RuntimeDirectory is used when it exists, and holds nothing afterwards.

    `ssl` loads a chain from paths, so the key is written; leaving it there means the private key
    outlives the process on a disk an operator did not choose, and `kill -9` has no cleanup path.
    """
    runtime = tmp_path / "run"
    runtime.mkdir()
    monkeypatch.setattr(host_agent, "BUNDLE_RUNTIME_DIR", runtime)
    context, _revision, _revoked, error = host_agent.resolve_bundle(str(write_bundle(tmp_path,
                                                                                       tls_fleet.bundle)))
    assert error is None and context is not None
    assert list(runtime.iterdir()) == [], "the private key must not stay on disk once loaded"


def test_a_missing_runtime_directory_falls_back_to_a_temporary_one(
        tls_fleet: SimpleNamespace, tmp_path: Path, monkeypatch) -> None:
    monkeypatch.setattr(host_agent, "BUNDLE_RUNTIME_DIR", tmp_path / "not-mounted")
    context, revision, _revoked, error = host_agent.resolve_bundle(str(write_bundle(tmp_path,
                                                                                    tls_fleet.bundle)))
    assert error is None and context is not None
    assert revision == tls_fleet.bundle_rev


def test_a_client_certificate_signed_by_the_root_reads_the_snapshot(
        tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
    """The whole point of the phase, plus the two fields task 03 reads.

    `bundle_rev` is how `/certs` tells "this host runs an old bundle" from "this host is current",
    so it is the number the bundle carried, not a value the agent made up.
    """
    base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle))
    status, body = _https_get(base + "/snapshot",
                              client_context(tls_fleet.root_cert, tls_fleet.client_cert,
                                             tls_fleet.client_key))
    assert status == 200
    assert body["ok"] is True and body["host"]["hostname"]
    assert body["tls"] is True
    assert body["bundle_rev"] == tls_fleet.bundle_rev
    for path in ("/info", "/schema", "/"):
        status, _body = _https_get(base + path, client_context(tls_fleet.root_cert,
                                                               tls_fleet.client_cert,
                                                               tls_fleet.client_key))
        assert status == 200, path


def test_health_is_probeable_with_the_root_ca_alone_while_every_other_path_is_not(
        tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
    """The `CERT_OPTIONAL` assumption: a probe with no client certificate must still work.

    `install-agent.sh --status` and the systemd probe hold the CA and were never handed a client
    certificate; `CERT_REQUIRED` would make every TLS install look dead. Everything that carries
    metrics is 403 — including the status page, which repeats hostname, kernel and CPU model.
    """
    base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle))
    probe = client_context(tls_fleet.root_cert)
    status, body = _https_get(base + "/health", probe)
    assert status == 200 and body["ok"] is True
    for path in ("/snapshot", "/info", "/schema", "/", "/nope"):
        refused, detail = _https_get(base + path, probe)
        assert refused == 403, path
        assert detail["detail"] == "client certificate required", path


def test_a_client_signed_by_another_authority_is_refused(
        tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
    """A certificate that does not chain to the bundle's root never reaches a handler.

    The server verifies at the handshake, so the answer is an aborted connection rather than an
    HTTP code — the assertion is only ever `not 200` because which alert arrives is the peer
    OpenSSL's choice, not ours.
    """
    base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle))
    stranger = client_context(tls_fleet.root_cert, tls_fleet.foreign_cert, tls_fleet.foreign_key)
    with pytest.raises((OSError, http.client.HTTPException)) as caught:
        _https_get(base + "/snapshot", stranger)
    assert not isinstance(caught.value, urllib.error.HTTPError)


def test_a_plaintext_request_to_a_tls_agent_gets_nothing(
        tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
    """A port probe cannot read the answer, and does not get a helpful error either."""
    base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle))
    port = int(base.rsplit(":", 1)[-1])
    with socket.create_connection(("127.0.0.1", port), timeout=15) as raw:
        raw.sendall(b"GET /snapshot HTTP/1.1\r\nHost: x\r\n\r\n")
        raw.settimeout(5)
        try:
            payload = raw.recv(200)
        except OSError:
            payload = b""
    assert b"hostname" not in payload


def test_a_revoked_client_certificate_is_refused_and_the_bundle_is_re_read(
        tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
    """The two halves: the denylist bites, and re-deploying the bundle takes effect no later
    than the next request — no restart, which is the only reason the lines live in the file."""
    path = write_bundle(tmp_path, tls_fleet.bundle
                        + f"{host_agent.REVOKED_LINE}{tls_fleet.stolen.serial}\n")
    bundle, error = host_agent.TlsBundle.load(str(path))
    assert error is None and bundle is not None
    base = tls_agent_server(bundle)
    stolen = client_context(tls_fleet.root_cert, str(tls_fleet.stolen.cert), str(tls_fleet.stolen.key))

    status, body = _https_get(base + "/snapshot", stolen)
    assert status == 403
    assert body["detail"] == "client certificate revoked"
    # A withdrawn identity is refused even on the open path: the exemption is for a probe that
    # holds the CA and no certificate, not for a certificate the operator has just taken away.
    assert _https_get(base + "/health", stolen)[0] == 403
    # The dashboard's own identity is untouched — the list names a serial, not a host.
    assert _https_get(base + "/snapshot", client_context(tls_fleet.root_cert, tls_fleet.client_cert,
                                                         tls_fleet.client_key))[0] == 200

    path.write_text(tls_fleet.bundle, encoding="ascii")  # a redeploy: the line is gone
    status, body = _https_get(base + "/snapshot", stolen)
    assert status == 200, "the denylist is re-read when the file changes, without a restart"


def test_a_renamed_denylist_entry_is_re_read_on_the_next_request(
        tls_fleet: SimpleNamespace, tmp_path: Path) -> None:
    """An unreadable bundle keeps the list this agent already had rather than forget it."""
    path = write_bundle(tmp_path, tls_fleet.bundle
                        + f"{host_agent.REVOKED_LINE}{tls_fleet.stolen.serial}\n")
    bundle, error = host_agent.TlsBundle.load(str(path))
    assert error is None and bundle is not None
    assert bundle.revoked_serials() == frozenset({str(tls_fleet.stolen.serial)})
    path.unlink()
    assert bundle.revoked_serials() == frozenset({str(tls_fleet.stolen.serial)})


def test_the_token_is_not_read_on_the_tls_path(
        tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
    """Bundle plus token: the handshake is the auth, so a stolen token cannot outlive a revoke.

    Checking the bearer on the TLS path too would not be defence in depth — it would be a second
    credential that keeps working after the operator revoked the certificate, which is the one
    thing the denylist exists to prevent.
    """
    base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle), token="s3cret")
    status, _body = _https_get(base + "/snapshot",
                               client_context(tls_fleet.root_cert, tls_fleet.client_cert,
                                              tls_fleet.client_key))
    assert status == 200


def test_token_mode_reports_no_tls_fields(agent_server) -> None:
    """Token mode is byte-identical: no `tls`, no `bundle_rev`, no new field that an agent
    installed before phase 15 does not have."""
    assert host_agent.Handler.bundle is None, "the token-mode fixture must not inherit a bundle"
    base, _port = agent_server(token="s3cret")
    _status, body = _get(base + "/snapshot", {"Authorization": "Bearer s3cret"})
    assert body["ok"] is True
    assert "tls" not in body and "bundle_rev" not in body


def test_serve_wraps_the_socket_and_says_tls_and_the_revision(
        tls_fleet: SimpleNamespace, tmp_path: Path, capsys, monkeypatch) -> None:
    """`serve()` is the function the unit runs, so its start line is what an operator reads.

    The server is recorded rather than guessed at a port: the test needs the ephemeral port the
    OS chose *and* a way to stop `serve_forever`, which is the same object.
    """
    created: list[ThreadingHTTPServer] = []

    class Recording(ThreadingHTTPServer):
        def __init__(self, *args: object, **kwargs: object) -> None:
            super().__init__(*args, **kwargs)  # type: ignore[arg-type]
            created.append(self)

    monkeypatch.setattr(host_agent, "ThreadingHTTPServer", Recording)
    bundle = load_bundle(tmp_path, tls_fleet.bundle)
    thread = threading.Thread(target=host_agent.serve,
                              args=("127.0.0.1", 0, host_agent.Collector("/", smart=False),
                                    None, bundle),
                              daemon=True)
    thread.start()
    deadline = time.monotonic() + 15
    while not created and time.monotonic() < deadline:
        time.sleep(0.01)
    assert created, "serve() never built its server"
    httpd = created[0]
    port = httpd.server_address[1]
    status, _body = _https_get(f"https://127.0.0.1:{port}/health",
                               client_context(tls_fleet.root_cert))
    assert status == 200
    httpd.shutdown()
    thread.join(timeout=15)
    printed = capsys.readouterr().out
    assert "tls (client cert required, rev " in printed
    # the scheme is the whole claim here (the port in the line is the requested one, which is 0
    # when the OS picks, exactly as in token mode)
    assert "[agent] listening on https://127.0.0.1" in printed
    assert "token" not in printed, "a TLS agent does not read a token"


def test_a_bundle_allows_the_off_loopback_bind_without_a_token(
        tls_fleet: SimpleNamespace, tmp_path: Path, monkeypatch) -> None:
    """The client certificate *is* the protection, so the exit-2 refusal does not apply."""
    served: list[tuple] = []
    monkeypatch.setattr(host_agent, "serve",
                        lambda *args: served.append(args))
    path = write_bundle(tmp_path, tls_fleet.bundle)
    assert host_agent.main(["--host", "0.0.0.0", "--port", "0", "--cert-bundle", str(path)]) == 0
    assert served, "a bundle-configured agent must start"
    host_arg, _port, _collector, token_arg, bundle_arg = served[0]
    assert (host_arg, token_arg) == ("0.0.0.0", None)
    assert isinstance(bundle_arg, host_agent.TlsBundle)


@pytest.mark.parametrize("extra", [[], ["--cert-bundle", ""]])
def test_off_loopback_without_a_token_or_a_bundle_is_refused_and_names_both(
        tmp_path: Path, capsys, monkeypatch, extra: list[str]) -> None:
    monkeypatch.delenv("AGENT_TOKEN", raising=False)
    monkeypatch.delenv("AGENT_CERT_BUNDLE", raising=False)
    args = ["--host", "0.0.0.0", "--port", "0", *extra]
    assert host_agent.main(args) == 2
    printed = capsys.readouterr().out
    assert "--token" in printed and "--cert-bundle" in printed


def test_an_unreadable_bundle_is_fatal_at_startup(tmp_path: Path, capsys) -> None:
    """Exit 2 naming the file: a degraded-to-plain-HTTP agent is the answer that is worse."""
    missing = tmp_path / "absent.certs"
    assert host_agent.main(["--host", "127.0.0.1", "--port", "0",
                            "--cert-bundle", str(missing)]) == 2
    assert str(missing) in capsys.readouterr().out


def schema_documents(field: str) -> bool:
    """Is this field in the SCHEMA block? `GET /schema` answers with that text, self-documenting
    an install for whoever has to debug it, so a field the payload carries belongs there."""
    doc = host_agent.__doc__ or ""
    schema = doc.split("SCHEMA (GET /snapshot)", 1)[-1].split("Deployment:", 1)[0]
    return re.search(rf"^\s+{field}\s", schema, re.M) is not None


def test_the_agent_version_and_schema_document_the_new_fields() -> None:
    """1.1.0 because the payload grew, and the docstring is what `GET /schema` answers."""
    assert host_agent.AGENT_VERSION == "1.1.0"
    assert schema_documents("tls") and schema_documents("bundle_rev")
    assert "--cert-bundle" in (host_agent.__doc__ or "")