Files
dashboards/tests/test_agent.py
T
ducoterra 44d61d3ab9 phase: 28_gateway_agent_proxy
**Phase 28 final pass — complete, 100 % green.** All five tasks were already `complete/`; this pass verified the criteria, re-ran the live half, and did the docs pass.

- Gate `bash .agents/validate.sh` **rc=0**: `1650 passed, 15 skipped`, coverage **98.97 %** (floor 94), `ruff` clean, `pyright` **0 errors**, `node --check` clean, hygiene clean (`/tmp/validate-done.log`).
- Live: `pytest -m live --live` **11 passed**; `./scripts/dev-server.sh status` nine sources `ok`; mTLS re-read of `gateway` → agent **1.2.0**, all three sections `available: true`, 6/6 frontends routed, 11 nginx servers + `$targetBackend`, **19** listeners, no core errors (`data/logs/28-finalpass-live.out`); installed agent file byte-identical to this tree (`b0e9ebf1b718…`).
- Criteria: 1 ✅ pytest/coverage · 2 ✅ three sections over a captured tree, `false`/`null` + one `errors` line each elsewhere with `cpu`/`mem`/`disk_io`/`system` intact · 3 ✅ `AGENT_VERSION` 1.2.0 + SCHEMA names all three · 4 ✅ six fixtures verbatim, capture refuses a bad read, 116 socket address columns hold no globally routable endpoint · 5 ✅ gateway runs 1.2.0 and answers over mTLS (transcript `data/logs/28-task05-rollout.md`, `28-finalpass-live.out`) · 6 ✅ validate · 7 ✅ additive: HEAD's 1.1.0 and this 1.2.0 `collect()`ed over one `--root` are key-for-key equal plus the three new sections, `--no-proxy` equal exactly (`data/logs/verify-28-final.py` → 41 PASS, `28-final-verify.out`).
- Docs fixed (README/AGENTS/Containerfile, all re-green): README's opening said "Four dashboards" above a five-row table and said the new sections are "stored in the snapshot" (they are in the **agent's** document — no `/api` payload, no `sample` row, nothing charts them until phase 29); `Containerfile`'s header still said "three dashboards"; AGENTS's coverage figure (98.94 → **98.97 %**), invariant 6 gained the captured-socket-table identifier rule, the agent bullet gained "phase 28 stops at the wire", and the fleet record names the stash-and-compare control plus the volatile listener count. Full record: `data/logs/NOTES-28-final.md`.
- Deviations: none functional; no git index/history touched, phase files untouched (`PLAN.md` and `complete/` unchanged). `uv pip list --outdated` still lists 4 packages (fastapi 0.153→0.161 etc.) — advisory only, dependencies untouched by this phase, left for a deliberate pass.
- Next pending phase: `.agents/phases/todo/29_gateway_dashboard_page/` (the page that consumes these three sections).
2026-10-22 20:24:01 -04:00

3038 lines
151 KiB
Python

"""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 ipaddress
import json
import os
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, collectors
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")
#: The add-on sections report their own absence in `errors` (`smart:`, `haproxy:`), because
#: "this box has no smartmontools / no haproxy" is a fact the page that owns the add-on has to
#: be able to print. Tests that assert a snapshot is otherwise clean filter the same prefixes
#: the dashboard does — `collectors.ADDON_ERROR_PREFIXES`, pinned equal to this list by
#: `test_the_addon_error_prefixes_are_the_same_list_the_dashboard_filters`, so neither side can
#: quietly grow a fourth opinion.
ADDON_ERROR_PREFIXES = ("smart", "haproxy", "nginx", "sockets")
def test_the_addon_error_prefixes_are_the_same_list_the_dashboard_filters() -> None:
"""One list, two readers. The agent writes `smart:` / `haproxy:` / `nginx:` lines, this suite
filters them to decide a snapshot is otherwise clean, and `collectors.py` filters the same
prefixes to decide whether the **system** payload is `degraded` — which is what an overview
alert is built from. A new add-on section that forgets the prefix on the dashboard side puts
a WARN on every host that never installed it (invariant 3), and this is the test that makes
that a red build instead of a fleet-wide false alarm.
"""
assert collectors.ADDON_ERROR_PREFIXES == ADDON_ERROR_PREFIXES
def core_errors(snapshot: dict[str, Any]) -> list[str]:
"""`errors` minus the add-ons' own "not here" lines: the failures that matter to a test."""
return [line for line in snapshot["errors"]
if not str(line).lstrip().lower().startswith(ADDON_ERROR_PREFIXES)]
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 core_errors(snap) == [], 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 core_errors(snap) == []
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}"
# ---- haproxy config: the additive `haproxy` section (phase 28) --------------- #
#
# The text below is this fleet's gateway config as read (read-only) from
# `gateway:/etc/haproxy/haproxy.cfg` on 2026-10-21: `global` and `defaults`, then six
# `frontend` blocks of exactly `bind :::<port> v4v6` + `default_backend <name>`, then six
# `backend` blocks of exactly `server <name> <domain>:<port> check inter 5s`. The structure, the
# indentation and the option spelling are verbatim; the service names and domains are synthetic,
# which is the `smartctl_reference_*` precedent — the shape is what the parser has to match, and
# a tracked file holding this fleet's whole ingress map is a disclosure the parser does not need.
# `tests/fixtures/gateway_haproxy.cfg` is the **verbatim** capture of that file (read-only over
# the `gateway-ops` key, 2026-10-21, no credential in it because haproxy configs hold none), and
# `test_the_recorded_gateway_haproxy_config_parses_too` pins the real service names against it.
# `scripts/capture-gateway-config.sh` (task 04) re-records the same bytes.
GATEWAY_HAPROXY_SHAPE = """global
log /dev/log local0
maxconn 4096
defaults
mode tcp
timeout connect 5s
timeout client 30s
timeout server 30s
frontend haproxy-22
bind :::22 v4v6
default_backend svc-2222
frontend haproxy-2000
bind :::2000 v4v6
default_backend svc-2000
frontend haproxy-5201
bind :::5201 v4v6
default_backend svc-5201
frontend haproxy-8080
bind :::8080 v4v6
default_backend svc-8080
frontend haproxy-25565
bind :::25565 v4v6
default_backend svc-25565
frontend haproxy-25566
bind :::25566 v4v6
default_backend svc-25566
backend svc-2222
server svc2222 svc2222.example.com:2222 check inter 5s
backend svc-2000
server svc2000 svc2000.example.com:2000 check inter 5s
backend svc-5201
server svc5201 svc5201.example.com:5201 check inter 5s
backend svc-8080
server svc8080 svc8080.example.com:8080 check inter 5s
backend svc-25565
server svc25565 svc25565.example.com:25565 check inter 5s
backend svc-25566
server svc25566 svc25566.example.com:25566 check inter 5s
"""
def haproxy_section(tmp_path: Path, text: str | None = None, **kwargs: Any) -> dict[str, Any]:
"""One collector's `haproxy` section over a tree that holds (or does not) a config."""
if text is not None:
write_file(tmp_path / "root", "etc/haproxy/haproxy.cfg", text)
errors: list[str] = []
section = host_agent.Collector(str(tmp_path / "root"), **kwargs).haproxy(errors)
section["_errors"] = errors # the pass writes into the caller's list; tests read both
return section
def test_the_captured_haproxy_shape_parses_into_frontends_and_backends() -> None:
"""The whole point of the section: six listening pipes and six targets, as the config says."""
parsed = host_agent.parse_haproxy_config(GATEWAY_HAPROXY_SHAPE)
assert parsed["recognised"] is True
assert [frontend["name"] for frontend in parsed["frontends"]] == [
"haproxy-22", "haproxy-2000", "haproxy-5201", "haproxy-8080",
"haproxy-25565", "haproxy-25566"]
# the split the dashboard's whole page depends on: `:::22` is the address `::` on port 22,
# not the address `:::` and not port `0` (an IPv6 bind is the common case on this host)
assert [frontend["binds"] for frontend in parsed["frontends"]] == [
[{"addr": "::", "port": port}] for port in (22, 2000, 5201, 8080, 25565, 25566)]
assert [frontend["default_backend"] for frontend in parsed["frontends"]] == [
"svc-2222", "svc-2000", "svc-5201", "svc-8080", "svc-25565", "svc-25566"]
assert [backend["servers"] for backend in parsed["backends"]] == [
[{"name": f"svc{port}", "target": f"svc{port}.example.com", "port": port}]
for port in (2222, 2000, 5201, 8080, 25565, 25566)]
# routing is the names themselves, and the config joins: every frontend's default_backend is
# one of the backends listed beside it (phase 29 draws its pipes from exactly this)
names = {backend["name"] for backend in parsed["backends"]}
assert {frontend["default_backend"] for frontend in parsed["frontends"]} <= names
# `global`/`defaults` are a section each, not a frontend and not a backend
assert len(parsed["backends"]) == 6
def test_the_recorded_gateway_haproxy_config_parses_too() -> None:
"""The real file, pinned: the six pipes the gateway actually routes on, and the six targets
they land in. A re-capture that drifts — a frontend removed, a bind rewritten, a `server`
line re-flowed — fails here before it fails phase 29's page."""
parsed = host_agent.parse_haproxy_config((FIXTURES / "gateway_haproxy.cfg").read_text())
assert parsed["recognised"] is True
assert [frontend["name"] for frontend in parsed["frontends"]] == [
"haproxy-22", "haproxy-2000", "haproxy-5201", "haproxy-8080",
"haproxy-25565", "haproxy-25566"]
assert [frontend["binds"] for frontend in parsed["frontends"]] == [
[{"addr": "::", "port": port}] for port in (22, 2000, 5201, 8080, 25565, 25566)]
assert [frontend["default_backend"] for frontend in parsed["frontends"]] == [
"gitea-2222", "sec-gitea-2000", "iperf3-5201", "unifi-external-8080",
"testcraft-25565", "nimcraft-25566"]
assert [(backend["name"], backend["servers"]) for backend in parsed["backends"]][:2] == [
("gitea-2222", [{"name": "gitea", "target": "reeseapps-gitea.reeselink.com",
"port": 2222}]),
("sec-gitea-2000", [{"name": "sec-gitea", "target": "stackexpected-gitea.reeselink.com",
"port": 2222}]),
]
# every frontend's target exists beside it — the join phase 29 draws its pipes from
names = {backend["name"] for backend in parsed["backends"]}
assert {frontend["default_backend"] for frontend in parsed["frontends"]} <= names
for backend in parsed["backends"]:
for server in backend["servers"]:
assert server["name"] and server["target"] and server["port"] is not None
def test_a_bind_splits_at_the_last_colon_and_one_without_a_port_is_skipped() -> None:
parsed = host_agent.parse_haproxy_config(
"frontend a\n"
" bind :::8443 v4v6\n"
" bind *:80 ssl crt /etc/haproxy/certs/example.com.pem\n"
" bind :8080\n"
" bind [::1]:9090\n"
" bind unix@/run/haproxy/admin.sock\n"
" bind 10.1.0.1:8000-8100\n"
" bind 127.0.0.1\n"
)
assert parsed["frontends"][0]["binds"] == [
{"addr": "::", "port": 8443},
{"addr": "*", "port": 80},
{"addr": None, "port": 8080}, # haproxy's "every address", spelled by absence
{"addr": "[::1]", "port": 9090},
]
assert parsed["recognised"] is True
def test_a_server_that_names_no_port_keeps_null_and_unknown_lines_are_ignored() -> None:
"""Rule 2 on the one optional field, and the reason a real config cannot amber the page."""
parsed = host_agent.parse_haproxy_config(
"backend bare\n"
" balance roundrobin\n"
" default-server check inter 5s\n"
" server withport 10.1.0.9:8080 check\n"
" server noport 10.1.0.10\n"
" server servicename host.example.com:http check\n"
" server broken\n"
" server-template tpl 4 pool.example.com:8080\n"
" option httpchk GET /healthz\n"
" http-request deny deny_status 403\n"
)
assert parsed["backends"][0]["servers"] == [
{"name": "withport", "target": "10.1.0.9", "port": 8080},
{"name": "noport", "target": "10.1.0.10", "port": None},
{"name": "servicename", "target": "host.example.com:http", "port": None},
]
# a `backend` body never grows frontend fields, however many directives it holds
assert "binds" not in parsed["backends"][0]
def test_a_listen_section_is_reported_as_the_pair_haproxy_says_it_is() -> None:
"""haproxy's shorthand *is* a frontend and a backend of the same name; modelling it keeps a
config that uses it from reading as "this gateway routes nothing"."""
parsed = host_agent.parse_haproxy_config(
"listen stats\n"
" bind :::8404 v4v6\n"
" stats uri /stats\n"
" server own 127.0.0.1:8405\n"
)
assert parsed["frontends"] == [{"name": "stats", "binds": [{"addr": "::", "port": 8404}],
"default_backend": "stats", "backends": ["stats"]}]
assert parsed["backends"] == [{"name": "stats",
"servers": [{"name": "own", "target": "127.0.0.1",
"port": 8405}]}]
def test_a_backend_computed_per_request_contributes_no_name() -> None:
"""`use_backend %[req.hdr(host),map_dom(...)]` has no name in it, and putting the expression
where the dashboard expects a backend would be a wrong answer, not a partial one."""
parsed = host_agent.parse_haproxy_config(
"frontend mapped\n"
" bind :::443 v4v6\n"
" use_backend %[req.hdr(host),lower,map_dom(/etc/haproxy/maps/domains.map)]\n"
" use_backend canary if is_canary\n"
" use_backend canary if is_other_canary\n"
)
assert parsed["frontends"][0]["default_backend"] is None
assert parsed["frontends"][0]["backends"] == ["canary"]
def test_a_haproxy_config_path_is_an_argument(tmp_path: Path) -> None:
"""The path is a flag because a package or a config management tree may keep it elsewhere;
it resolves under `--root`, so the fake-tree tests read the fake tree."""
write_file(tmp_path / "root", "opt/haproxy/etc/haproxy.cfg",
"frontend f\n bind :::1234 v4v6\n")
errors: list[str] = []
section = host_agent.Collector(
str(tmp_path / "root"), haproxy_config="/opt/haproxy/etc/haproxy.cfg").haproxy(errors)
assert section["available"] is True
assert section["config"] == "/opt/haproxy/etc/haproxy.cfg"
assert section["frontends"][0]["binds"] == [{"addr": "::", "port": 1234}]
assert errors == []
def test_a_relative_haproxy_config_path_is_taken_as_written(tmp_path: Path, monkeypatch) -> None:
"""`--smartctl` and haproxy's own `-f` both resolve a relative path against the working
directory, so this one does too — only the absolute default is read under `--root`."""
(tmp_path / "haproxy.cfg").write_text("frontend f\n bind :::7 v4v6\n")
monkeypatch.chdir(tmp_path)
errors: list[str] = []
section = host_agent.Collector("/", haproxy_config="haproxy.cfg").haproxy(errors)
assert section["available"] is True
assert section["frontends"][0]["binds"] == [{"addr": "::", "port": 7}]
assert errors == []
def test_a_missing_haproxy_config_is_available_false_with_one_errors_line(tmp_path: Path) -> None:
"""A host without haproxy is unconfigured, not broken: one line for the pass, and the other
sections still there."""
section = haproxy_section(tmp_path)
assert section["available"] is False
assert section["config"] == host_agent.HAPROXY_CONFIG_PATH
assert section["frontends"] == [] and section["backends"] == []
haproxy_errors = [line for line in section["_errors"] if line.startswith("haproxy:")]
assert len(haproxy_errors) == 1, "one missing file is one fact, not one per section"
assert "not found" in haproxy_errors[0]
# cross-module contract, the `smartctl: not found` pin's shape: the phase-29 gateway source
# reads this word to answer "is haproxy installed?", and `collectors` keeps the line out of
# the system page's `degraded`. Reword it here and that test fails instead of a page
# losing its explanation — and the line must name the path it looked at.
assert host_agent.HAPROXY_CONFIG_PATH in haproxy_errors[0]
assert "--no-proxy" in haproxy_errors[0], "the line has to say how to stop asking"
def test_the_haproxy_pass_says_nothing_when_the_config_reads_cleanly(tmp_path: Path) -> None:
section = haproxy_section(tmp_path, GATEWAY_HAPROXY_SHAPE)
assert section["available"] is True
assert section["_errors"] == [], "a read that worked writes no line"
assert len(section["frontends"]) == 6
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0,
reason="root reads any mode, so a 000 file is unreadable only below it")
def test_an_unreadable_haproxy_config_is_available_null_with_one_errors_line(
tmp_path: Path) -> None:
"""The file is there and the agent user cannot read it — a different operator action from
"no haproxy here", so a different `available` and a different sentence."""
path = tmp_path / "root" / "etc" / "haproxy" / "haproxy.cfg"
path.parent.mkdir(parents=True)
path.write_text(GATEWAY_HAPROXY_SHAPE)
path.chmod(0o000)
try:
section = haproxy_section(tmp_path)
finally:
path.chmod(0o644)
assert section["available"] is None
assert len(section["_errors"]) == 1 and section["_errors"][0].startswith("haproxy:")
assert "could not be read" in section["_errors"][0]
def test_a_directory_where_the_config_should_be_is_available_null(tmp_path: Path) -> None:
"""Root-proof arm of the same fact: the path exists and is not a readable file."""
(tmp_path / "root" / "etc" / "haproxy" / "haproxy.cfg").mkdir(parents=True)
section = haproxy_section(tmp_path)
assert section["available"] is None
assert len(section["_errors"]) == 1 and "could not be read" in section["_errors"][0]
def test_an_oversized_haproxy_config_is_refused_rather_than_read(
tmp_path: Path, monkeypatch) -> None:
"""The bound exists so a `--haproxy-config` aimed at a log or a device node cannot cost the
snapshot an unbounded read."""
monkeypatch.setattr(host_agent, "HAPROXY_CONFIG_MAX_BYTES", 64)
section = haproxy_section(tmp_path, "frontend f\n" + " bind :::1 v4v6\n" * 40)
assert section["available"] is None
assert len(section["_errors"]) == 1 and "could not be read" in section["_errors"][0]
def test_a_config_that_is_not_haproxy_degrades_only_its_own_section(tmp_path: Path) -> None:
"""Garbage text is `available: null` plus one line — and the snapshot it lives in is intact,
which is the invariant-4 half of this add-on."""
root = tmp_path / "root"
write_file(root, "proc/stat",
"cpu 100 0 50 800 0 0 0 0 0 0\ncpu0 100 0 50 800 0 0 0 0 0 0\n")
write_file(root, "proc/meminfo", "MemTotal: 8000000 kB\nMemFree: 1000000 kB\n"
"MemAvailable: 4000000 kB\n")
write_file(root, "etc/haproxy/haproxy.cfg", "\x00\x01garbage\nnot a haproxy config at all\n")
snap = host_agent.Collector(str(root)).snapshot()
assert snap["haproxy"]["available"] is None
assert snap["haproxy"]["frontends"] == []
haproxy_errors = [line for line in snap["errors"] if line.startswith("haproxy:")]
assert len(haproxy_errors) == 1, snap["errors"]
assert "no haproxy section" in haproxy_errors[0]
# the sections that are not this add-on still answered
assert snap["cpu"]["modes"]["total"] > 0
assert snap["memory"]["total_bytes"] == 8000000 * 1024.0
assert core_errors(snap) == [], snap["errors"]
def test_no_proxy_removes_the_haproxy_section_and_says_nothing(tmp_path: Path) -> None:
"""The `--no-smart` precedent: opting out is silent, and "not collected" stays
distinguishable from "collected and empty" because the key is absent."""
root = tmp_path / "root"
write_file(root, "etc/haproxy/haproxy.cfg", GATEWAY_HAPROXY_SHAPE)
snap = host_agent.Collector(str(root), proxy=False).snapshot()
assert "haproxy" not in snap, "--no-proxy removes the key, not just its contents"
assert core_errors(snap) == []
def test_the_no_proxy_flag_reaches_the_collector_from_the_command_line(capsys) -> None:
"""A flag nobody wired is a flag that does nothing: `--once` proves the CLI path, and one
flag covers the whole add-on family (`haproxy`, `nginx` and `sockets` leave together)."""
assert host_agent.main(["--once", "--no-proxy"]) == 0
snap = json.loads(capsys.readouterr().out)
assert snap["ok"] is True
assert "haproxy" not in snap and "nginx" not in snap and "sockets" not in snap
assert host_agent.main(["--once"]) == 0
out = json.loads(capsys.readouterr().out)
assert "haproxy" in out and "nginx" in out and "sockets" in out
def test_the_haproxy_section_is_documented_in_the_schema() -> None:
"""`GET /schema` is how an install documents itself, so the section an operator is looking at
has to be named there with every key it carries — the `tls`/`bundle_rev` precedent."""
assert schema_documents("haproxy")
doc = host_agent.__doc__ or ""
block = doc.split(" haproxy ", 1)[-1].split(" errors ", 1)[0]
assert block, "the haproxy section has no SCHEMA block to document it"
for key in ("available", "config", "frontends", "binds", "addr", "port", "default_backend",
"backends", "servers", "target"):
assert key in block, f"haproxy.{key} missing from the SCHEMA block"
assert "--no-proxy" in block and "--haproxy-config" in block, \
"the two flags that decide whether this section exists belong in its documentation"
# ---- nginx config: the additive `nginx` section (phase 28) ------------------ #
#
# The three texts below are this fleet's gateway nginx tree as read (read-only) from
# `gateway:/etc/nginx/nginx.conf`, `/etc/nginx/conf.d/http.conf` and
# `/etc/nginx/stream.d/stream.conf` on 2026-10-21: a main config whose top level is `events` +
# `stream { include …/stream.d/*.conf; }` + `http { … include …/conf.d/*.conf; }` with one
# catch-all `server` on port 80; one `conf.d` file of `listen 4443 ssl proxy_protocol` /
# `listen 8443 ssl` vhosts each with a `server_name <domain>` and a `location / { proxy_pass
# http://<domain>:<port>; }`; one `stream.d` file with a
# `map $ssl_preread_server_name $targetBackend { … }` block and a `server` that ends in
# `proxy_pass $targetBackend; ssl_preread on;`. The structure, the indentation and the flag
# spelling are verbatim; the domains and targets are the `smartctl_reference_*` precedent — the
# shape is what the parser must match, and this fleet's ingress map is not a tracked file's
# business. `tests/fixtures/gateway_nginx_*.conf` (task 04) is the verbatim capture, and the
# re-record script is `scripts/capture-gateway-config.sh`.
GATEWAY_NGINX_MAIN = """user nginx;
worker_processes auto;
error_log /var/log/nginx/error.log notice;
pid /run/nginx.pid;
events {
worker_connections 1024;
}
stream {
include /etc/nginx/stream.d/*.conf;
}
http {
real_ip_header proxy_protocol;
include /etc/nginx/mime.types;
default_type application/octet-stream;
map $http_upgrade $connection_upgrade {
default upgrade;
'' close;
}
server {
listen 80;
listen [::]:80;
server_name _;
return 301 https://$host$request_uri;
}
include /etc/nginx/conf.d/*.conf;
}
"""
GATEWAY_NGINX_HTTP = """server {
listen 4443 ssl proxy_protocol;
listen [::]:4443 ssl proxy_protocol;
server_name nextcloud.example.com;
client_max_body_size 0;
ssl_certificate /etc/letsencrypt/live/nextcloud.example.com/fullchain.pem;
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
location / {
proxy_pass http://nextcloud.internal.test:11000;
proxy_set_header Host $host;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
server {
listen 8443 ssl;
listen [::]:8443 ssl;
server_name alicia.example.com;
location / {
proxy_pass http://127.0.0.1:8081;
proxy_set_header Host $host;
}
}
"""
GATEWAY_NGINX_STREAM = """map $ssl_preread_server_name $targetBackend {
default 127.0.0.1:4443;
brain.example.com brain-of-reese.internal.test:443;
}
server {
listen 443;
listen [::]:443;
proxy_connect_timeout 1800s;
proxy_timeout 1800s;
proxy_protocol on;
proxy_pass $targetBackend;
ssl_preread on;
resolver 10.1.0.1;
}
"""
def nginx_tree(root: Path, *, main: str | None = GATEWAY_NGINX_MAIN,
http: str | None = GATEWAY_NGINX_HTTP, stream: str | None = GATEWAY_NGINX_STREAM,
extra: dict[str, str] | None = None) -> Path:
"""Write the captured gateway tree (or any part of it) under a fake `--root`."""
for relative, text in (("etc/nginx/nginx.conf", main),
("etc/nginx/conf.d/http.conf", http),
("etc/nginx/stream.d/stream.conf", stream)):
if text is not None:
write_file(root, relative, text)
for relative, text in (extra or {}).items():
write_file(root, relative, text)
return root
def nginx_section(tmp_path: Path, *, tree: bool = True, **kwargs: Any) -> dict[str, Any]:
"""One collector's `nginx` section over a tree that holds (or does not) the configs."""
root = tmp_path / "root"
if tree:
nginx_tree(root)
errors: list[str] = []
section = host_agent.Collector(str(root), **kwargs).nginx(errors)
section["_errors"] = errors # the pass writes into the caller's list; tests read both
return section
def by_variable(maps: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""The `maps` list keyed by variable name, because that is the key the dashboard joins on."""
return {entry["variable"]: entry for entry in maps}
def test_the_captured_nginx_shape_parses_into_servers_maps_and_targets() -> None:
"""The whole point of the section: every pipe nginx listens on and where each one lands."""
parsed = host_agent.parse_nginx_config(GATEWAY_NGINX_MAIN, {
"/etc/nginx/conf.d/http.conf": GATEWAY_NGINX_HTTP,
"/etc/nginx/stream.d/stream.conf": GATEWAY_NGINX_STREAM,
})
assert parsed["recognised"] is True
assert parsed["error"] is None and parsed["unparsed"] == []
# document order: the main config's own server, then stream.d (included first), then conf.d
assert [server["server_names"] for server in parsed["servers"]] == [
["_"], [], ["nextcloud.example.com"], ["alicia.example.com"]]
# `listen` keeps the address the config wrote: a bare `80` has no address (nginx's own "every
# address"), `[::]:4443` is the address `::` and not the string `[::]`, and the `ssl
# proxy_protocol` flags are not this section's business
assert [server["listen"] for server in parsed["servers"]] == [
[{"addr": None, "port": 80}, {"addr": "::", "port": 80}],
[{"addr": None, "port": 443}, {"addr": "::", "port": 443}],
[{"addr": None, "port": 4443}, {"addr": "::", "port": 4443}],
[{"addr": None, "port": 8443}, {"addr": "::", "port": 8443}],
]
assert [server["proxy_passes"] for server in parsed["servers"]] == [
[],
# the stream form: no `location` exists, and the target is a variable, not an address
[{"location": None, "scheme": None, "target": None, "variable": "targetBackend"}],
[{"location": "/", "scheme": "http", "variable": None,
"target": {"host": "nextcloud.internal.test", "port": 11000}}],
[{"location": "/", "scheme": "http", "variable": None,
"target": {"host": "127.0.0.1", "port": 8081}}],
]
maps = by_variable(parsed["maps"])
assert set(maps) == {"connection_upgrade", "targetBackend"}
# the variable name is spelled one way on both sides of the join, and the `$` is nginx's
# syntax rather than part of the name — so it is dropped here once, not in the page
assert maps["targetBackend"]["entries"] == [
{"key": "default", "target": {"host": "127.0.0.1", "port": 4443}},
{"key": "brain.example.com", "target": {"host": "brain-of-reese.internal.test",
"port": 443}},
]
# …and the `proxy_pass $targetBackend` above still carries `target: null`: the map decides
# per connection, so merging it here would be a number the wire never published
assert parsed["servers"][1]["proxy_passes"][0]["target"] is None
# `$connection_upgrade`'s values are header words, not hosts: keys reported, targets null
assert maps["connection_upgrade"]["entries"] == [
{"key": "default", "target": None}, {"key": "", "target": None}]
# the include lines, as written, in document order — including the one file this tree has no
# opinion about because it is not a config (`mime.types`)
assert parsed["includes"] == ["/etc/nginx/stream.d/*.conf", "/etc/nginx/mime.types",
"/etc/nginx/conf.d/*.conf"]
assert parsed["files"] == ["/etc/nginx/stream.d/stream.conf", "/etc/nginx/conf.d/http.conf"]
def test_the_section_names_the_files_the_numbers_came_from(tmp_path: Path) -> None:
"""`files` is the answer to "which files is this page drawing?", main config first, and it
speaks the config's paths rather than the ones under `--root`."""
section = nginx_section(tmp_path)
assert section["available"] is True
assert section["files"] == ["/etc/nginx/nginx.conf",
"/etc/nginx/stream.d/stream.conf",
"/etc/nginx/conf.d/http.conf"]
assert section["_errors"] == [], "a read that worked writes no line"
assert len(section["servers"]) == 4
assert len(section["maps"]) == 2
def test_a_listen_token_splits_the_three_forms_nginx_writes() -> None:
parsed = host_agent.parse_nginx_config(
"http { server {\n"
" listen 80;\n"
" listen [::]:4443 ssl;\n"
" listen 10.1.0.5:9000;\n"
" listen 8080 default_server;\n"
" listen unix:/run/nginx.sock;\n"
" listen [::1;\n" # a bracket that never closes is not an address
" listen [::1];\n" # an address with no port is not one either
" listen;\n"
" } }"
)
assert parsed["servers"][0]["listen"] == [
{"addr": None, "port": 80},
{"addr": "::", "port": 4443},
{"addr": "10.1.0.5", "port": 9000},
{"addr": None, "port": 8080},
]
def test_a_proxy_pass_reports_the_address_the_variable_or_neither() -> None:
"""Three spellings of the same directive, and the agent never resolves one of them."""
parsed = host_agent.parse_nginx_config(
"http { server {\n"
" location /a/ { proxy_pass http://h1.test:8080/api; }\n"
" location /b/ { proxy_pass http://pool_name; }\n"
" location /c/ { proxy_pass $backend; }\n"
" location /d/ { proxy_pass unix:/run/app.sock; }\n"
" location /e/ { proxy_pass http://host:http; }\n"
" } }"
)
assert parsed["servers"][0]["proxy_passes"] == [
# the trailing path is dropped the way haproxy's `check inter 5s` is: it is not a place
{"location": "/a/", "scheme": "http", "target": {"host": "h1.test", "port": 8080},
"variable": None},
# an `upstream` name is a host with no port — the name is the fact, the port is not
{"location": "/b/", "scheme": "http", "target": {"host": "pool_name", "port": None},
"variable": None},
{"location": "/c/", "scheme": None, "target": None, "variable": "backend"},
# a unix socket is neither a host nor a variable: the line is reported with nulls
{"location": "/d/", "scheme": None, "target": None, "variable": None},
# a colon that is not a port either: the scheme is a fact, the host would be a guess
{"location": "/e/", "scheme": "http", "target": None, "variable": None},
]
@pytest.mark.parametrize(("text", "why"), [
("http { server { listen 80;", "an opening brace is never closed"),
("http { server { listen 80", "the file ends inside the directive"),
("} http { }", "no opening one before it"),
("http { server { listen 80 } }", "is still open"),
("http { { listen 80; } }", "with no block name"),
('http { add_header X "unclosed; }', "quoted string is never closed"),
])
def test_a_config_nginx_would_refuse_to_start_on_fails_the_scanner(text: str,
why: str) -> None:
"""Six malformed files, one verdict each, and no exception on a snapshot path.
The sentences matter because they are what the section says: this is the arm that answers
"the file is there and I could not read it as nginx", and a scanner that raised instead of
reporting would fail the whole `/snapshot` over one half-written vhost.
"""
parsed = host_agent.parse_nginx_config(text)
assert parsed["error"] is not None and why in parsed["error"], parsed
assert parsed["recognised"] is False and parsed["servers"] == []
assert host_agent.nginx_include_paths(text) == [], "no includes out of a file that is not a tree"
def test_an_upstream_block_keeps_its_server_lines_as_targets() -> None:
"""`server` means two different things in nginx, and the scanner tells them apart."""
parsed = host_agent.parse_nginx_config(
"upstream pool_name {\n"
" server 10.1.0.9:8080 max_fails=3 fail_timeout=30s;\n"
" server backend.test;\n"
" server [::1]:8080;\n"
" keepalive 32;\n"
"}\n"
"http { server { listen 80; } }"
)
assert parsed["upstreams"] == [{"name": "pool_name", "servers": [
{"host": "10.1.0.9", "port": 8080},
{"host": "backend.test", "port": None},
{"host": "::1", "port": 8080},
]}]
# an upstream's `server` lines are not server blocks, however many of them there are
assert len(parsed["servers"]) == 1
def test_a_quoted_value_holding_a_semicolon_is_one_word() -> None:
"""The real config writes `add_header … "max-age=31536000; includeSubDomains" always;`. A
`;` inside quotes is data, and cutting on it would halve the file's block structure — same
for the `{`, and an escaped quote inside the value stays inside it. A bare `;` with nothing
before it is whitespace, not an empty directive."""
parsed = host_agent.parse_nginx_config(
'http { server { listen 80; ; add_header X "a;b { \\"c\\"" always; return 404; } }'
)
assert parsed["servers"][0]["listen"] == [{"addr": None, "port": 80}]
assert parsed["error"] is None
# the stray `;` did not open an empty directive and the escaped quote did not end the value
assert parsed["recognised"] is True and len(parsed["servers"]) == 1
def test_a_commented_include_line_is_not_an_include() -> None:
"""An operator commented the old vhost file out; following it would report a pipe that is
not there, and reading it would cost a file the config took out of service."""
parsed = host_agent.parse_nginx_config(
"http {\n # include /etc/nginx/conf.d/*.conf;\n server { listen 80; }\n}",
{"/etc/nginx/conf.d/http.conf": 'server { listen 9999; server_name ghost.test; }'},
)
assert parsed["includes"] == [] and parsed["files"] == []
assert [server["listen"] for server in parsed["servers"]] == [[{"addr": None, "port": 80}]]
def test_includes_are_followed_one_level_and_no_deeper(tmp_path: Path) -> None:
"""The captured tree is one level deep. Going deeper would mean reading files the pass was
never bounded for, so an include inside an included file contributes nothing — and no
error, because the config is not wrong, the pass is deliberately shallow."""
root = tmp_path / "root"
nginx_tree(root, stream=None, http=None, extra={
# deliberately outside conf.d, so the main config's own glob cannot be what read it
"etc/nginx/hidden/deep.conf": 'server { listen 7777; server_name deep.test; }\n',
})
write_file(root, "etc/nginx/conf.d/http.conf",
GATEWAY_NGINX_HTTP + "\ninclude /etc/nginx/hidden/deep.conf;\n")
errors: list[str] = []
section = host_agent.Collector(str(root)).nginx(errors)
assert section["available"] is True
assert section["files"] == ["/etc/nginx/nginx.conf", "/etc/nginx/conf.d/http.conf"]
assert 7777 not in [entry["port"] for server in section["servers"] for entry
in server["listen"]], "the second level is not this pass's job"
assert errors == []
def test_a_literal_include_is_read_when_this_host_has_it(tmp_path: Path) -> None:
"""The non-glob arm of the include line. `mime.types` is the captured case: it exists on a
real host, holds a `types { … }` block this section does not model, and costs one file."""
root = tmp_path / "root"
nginx_tree(root, http=None, stream=None, extra={
"etc/nginx/mime.types": "types {\n text/html html htm;\n}\n",
})
errors: list[str] = []
section = host_agent.Collector(str(root)).nginx(errors)
assert section["available"] is True
assert section["files"] == ["/etc/nginx/nginx.conf", "/etc/nginx/mime.types"]
assert section["servers"][0]["listen"] == [{"addr": None, "port": 80},
{"addr": "::", "port": 80}]
assert errors == []
def test_an_include_naming_a_file_this_host_does_not_have_says_nothing(tmp_path: Path) -> None:
"""`include /etc/nginx/mime.types;` is in every main config and in no fake tree. Absence is
not a fault, and a line per absent file would bury the arms that are."""
section = nginx_section(tmp_path)
assert "/etc/nginx/mime.types" not in section["files"]
assert section["_errors"] == [], "absence is the `available: false` arm one level down"
def test_the_include_file_bound_drops_the_excess_with_one_line_naming_the_count(
tmp_path: Path) -> None:
"""The SMART pass-budget shape: cover what you can, say what you could not, keep the section
`available: true` because the config was read."""
root = tmp_path / "root"
nginx_tree(root, http=None, stream=None)
for index in range(10):
write_file(root, f"etc/nginx/conf.d/vhost{index}.conf",
f"server {{ listen {8100 + index}; server_name h{index}.test; }}\n")
errors: list[str] = []
section = host_agent.Collector(str(root)).nginx(errors)
assert section["available"] is True
limit = host_agent.NGINX_INCLUDE_MAX_FILES
assert section["files"] == ["/etc/nginx/nginx.conf"] + [
f"/etc/nginx/conf.d/vhost{index}.conf" for index in range(limit)]
assert len(section["servers"]) == limit + 1 # main's own server plus the 8 that were read
nginx_errors = [line for line in errors if line.startswith("nginx:")]
assert len(nginx_errors) == 1, "one bound is one fact, not one per dropped file"
assert f"{10 - limit} included file(s)" in nginx_errors[0], nginx_errors
assert f"{limit}-file" in nginx_errors[0], "the line names the bound it hit"
def test_the_include_byte_bound_drops_the_file_that_would_cross_it(tmp_path: Path,
monkeypatch) -> None:
"""The other half of the bound: one huge include must not turn a snapshot into a file copy."""
monkeypatch.setattr(host_agent, "NGINX_INCLUDE_MAX_BYTES", 120)
root = tmp_path / "root"
nginx_tree(root, stream=None)
# a padding comment, so the file is still valid nginx: it has to be dropped for its size and
# not because it is the half-written vhost the next test is about
write_file(root, "etc/nginx/conf.d/http.conf", f"# {'x' * 100}\n")
write_file(root, "etc/nginx/conf.d/second.conf", "server { listen 8123; }\n")
errors: list[str] = []
section = host_agent.Collector(str(root)).nginx(errors)
assert section["available"] is True
assert section["files"] == ["/etc/nginx/nginx.conf", "/etc/nginx/conf.d/http.conf"]
nginx_errors = [line for line in errors if line.startswith("nginx:")]
assert len(nginx_errors) == 1 and "1 included file(s)" in nginx_errors[0]
assert "120-byte" in nginx_errors[0], "the line names the bound it hit, in the unit it is set in"
def test_an_oversized_main_config_is_refused_rather_than_read(tmp_path: Path,
monkeypatch) -> None:
"""The bound exists so a `--nginx-config` aimed at a log or a device node cannot cost the
snapshot an unbounded read — the haproxy rule, on the other half of the add-on."""
monkeypatch.setattr(host_agent, "NGINX_CONFIG_MAX_BYTES", 64)
section = nginx_section(tmp_path)
assert section["available"] is None
assert len(section["_errors"]) == 1 and "could not be read" in section["_errors"][0]
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0,
reason="root reads any mode, so a 000 file is unreadable only below it")
def test_an_unreadable_main_config_is_available_null_with_one_errors_line(
tmp_path: Path) -> None:
"""The file is there and the agent user cannot read it — a different operator action from
"no nginx here", so a different `available` and a different sentence."""
root = tmp_path / "root"
nginx_tree(root)
path = root / "etc" / "nginx" / "nginx.conf"
path.chmod(0o000)
try:
section = nginx_section(tmp_path, tree=False) # do not rewrite the tree over the mode
finally:
path.chmod(0o644)
assert section["available"] is None
assert len(section["_errors"]) == 1 and section["_errors"][0].startswith("nginx:")
assert "could not be read" in section["_errors"][0]
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0,
reason="root reads any mode, so a 000 file is unreadable only below it")
def test_an_unreadable_included_file_is_reported_once_and_the_rest_still_lands(
tmp_path: Path) -> None:
"""One denied vhost file must not cost the whole section — and must not pass in silence
either, because the page would draw four pipes where nginx has five."""
root = tmp_path / "root"
nginx_tree(root, main=None)
(root / "etc" / "nginx").mkdir(parents=True, exist_ok=True)
write_file(root, "etc/nginx/nginx.conf", "http { include /etc/nginx/conf.d/*.conf; }\n")
write_file(root, "etc/nginx/conf.d/second.conf", "server { listen 8124; }\n")
denied = root / "etc" / "nginx" / "conf.d" / "http.conf"
denied.chmod(0o000)
try:
errors: list[str] = []
section = host_agent.Collector(str(root)).nginx(errors)
finally:
denied.chmod(0o644)
assert section["available"] is True
assert [entry["port"] for server in section["servers"] for entry in server["listen"]] == [8124]
nginx_errors = [line for line in errors if line.startswith("nginx:")]
assert len(nginx_errors) == 1, "one denied file is one line, not one per include line"
assert "could not be read" in nginx_errors[0]
assert "/etc/nginx/conf.d/http.conf" in nginx_errors[0], "it names what it could not read"
def test_an_included_file_that_will_not_parse_is_dropped_and_named(tmp_path: Path) -> None:
"""A half-written vhost file is that file's problem: the section keeps every other server
and says which document it could not read, because the main config still answered."""
root = tmp_path / "root"
nginx_tree(root, stream=None, http="server { listen 4443;\n")
errors: list[str] = []
section = host_agent.Collector(str(root)).nginx(errors)
assert section["available"] is True
assert section["files"] == ["/etc/nginx/nginx.conf"]
nginx_errors = [line for line in errors if line.startswith("nginx:")]
assert len(nginx_errors) == 1
assert "could not be parsed" in nginx_errors[0]
assert "/etc/nginx/conf.d/http.conf" in nginx_errors[0]
# the main config's own server is still there: one bad include degrades, it does not blank
assert [entry["port"] for entry in section["servers"][0]["listen"]] == [80, 80]
def test_a_missing_nginx_config_is_available_false_with_one_errors_line(tmp_path: Path) -> None:
"""A host without nginx is unconfigured, not broken: one line for the pass, and the other
sections still there."""
section = nginx_section(tmp_path, tree=False)
assert section["available"] is False
assert section["config"] == host_agent.NGINX_CONFIG_PATH
assert section["files"] == [] and section["servers"] == []
nginx_errors = [line for line in section["_errors"] if line.startswith("nginx:")]
assert len(nginx_errors) == 1, "one missing file is one fact, not one per section"
assert "not found" in nginx_errors[0]
# cross-module contract, the `smartctl: not found` and `haproxy: … not found` pins' shape:
# the phase-29 gateway source reads this word to answer "is nginx installed?", and
# `collectors` keeps the line out of the system page's `degraded`.
assert host_agent.NGINX_CONFIG_PATH in nginx_errors[0]
assert "--nginx-config" in nginx_errors[0], "the line names the flag that moves the path"
assert "--no-proxy" in nginx_errors[0], "the line has to say how to stop asking"
def test_text_that_is_not_an_nginx_config_degrades_only_its_own_section(tmp_path: Path) -> None:
"""Valid text that names no nginx context is a different fact from an unreadable file, and
it is still one line and still only this section's problem."""
root = tmp_path / "root"
nginx_tree(root, http=None, stream=None)
write_file(root, "etc/nginx/nginx.conf", "some-other-daemon on;\nopts 3;\n")
errors: list[str] = []
section = host_agent.Collector(str(root)).nginx(errors)
assert section["available"] is None
assert section["servers"] == [] and section["files"] == []
nginx_errors = [line for line in errors if line.startswith("nginx:")]
assert len(nginx_errors) == 1, errors
assert "no nginx context" in nginx_errors[0]
def test_a_corrupt_nginx_config_leaves_the_rest_of_the_snapshot_intact(tmp_path: Path) -> None:
"""The invariant-4 half of this add-on: an unparseable main config is `available: null` plus
one line, and `/snapshot` still answers with its cpu and memory sections."""
root = tmp_path / "root"
write_file(root, "proc/stat",
"cpu 100 0 50 800 0 0 0 0 0 0\ncpu0 100 0 50 800 0 0 0 0 0 0\n")
write_file(root, "proc/meminfo", "MemTotal: 8000000 kB\nMemFree: 1000000 kB\n"
"MemAvailable: 4000000 kB\n")
write_file(root, "etc/nginx/nginx.conf", "http { server { listen 80;\n")
snap = host_agent.Collector(str(root)).snapshot()
assert snap["nginx"]["available"] is None
assert snap["nginx"]["servers"] == []
nginx_errors = [line for line in snap["errors"] if line.startswith("nginx:")]
assert len(nginx_errors) == 1, snap["errors"]
assert "could not be parsed" in nginx_errors[0]
assert snap["cpu"]["modes"]["total"] > 0
assert snap["memory"]["total_bytes"] == 8000000 * 1024.0
assert core_errors(snap) == [], snap["errors"]
def test_an_nginx_failure_inside_the_collector_still_yields_a_section(tmp_path: Path,
monkeypatch) -> None:
"""Whatever raises in here, the document keeps the key with `available: null` — a snapshot
that loses the key would read as a pre-1.2.0 agent, not as a failed add-on."""
def explode(*args: object) -> dict[str, Any]:
raise RuntimeError("boom")
monkeypatch.setattr(host_agent.Collector, "nginx", explode)
snap = host_agent.Collector(str(tmp_path / "root")).snapshot()
assert snap["nginx"] == {"available": None, "config": host_agent.NGINX_CONFIG_PATH,
"files": [], "servers": [], "upstreams": [], "maps": []}
assert any(line.startswith("nginx:") for line in snap["errors"])
assert core_errors(snap) == [], "the add-on's own failure line is not a system degradation"
def test_an_nginx_config_path_is_an_argument(tmp_path: Path) -> None:
"""Packaged and containerised nginx trees keep the main config in different places; the path
resolves under `--root`, so the fake-tree tests read the fake tree."""
write_file(tmp_path / "root", "opt/nginx/conf/nginx.conf",
'http { server { listen 7777; server_name arg.test; } }\n')
errors: list[str] = []
section = host_agent.Collector(
str(tmp_path / "root"), nginx_config="/opt/nginx/conf/nginx.conf").nginx(errors)
assert section["available"] is True
assert section["config"] == "/opt/nginx/conf/nginx.conf"
assert section["files"] == ["/opt/nginx/conf/nginx.conf"]
assert section["servers"][0]["listen"] == [{"addr": None, "port": 7777}]
assert errors == []
def test_a_relative_include_line_is_resolved_and_labelled_as_written(tmp_path: Path,
monkeypatch) -> None:
"""The relative arm of both halves: an include argument written without a leading `/` is
resolved against the working directory (where nginx's own prefix would resolve it) and stays
spelled that way in `files`, because that is how the config wrote it."""
(tmp_path / "nginx.conf").write_text("http { include conf.d/http.conf; }\n")
write_file(tmp_path, "conf.d/http.conf", "server { listen 8123; server_name rel.test; }\n")
monkeypatch.chdir(tmp_path)
errors: list[str] = []
section = host_agent.Collector("/", nginx_config="nginx.conf").nginx(errors)
assert section["available"] is True
assert section["files"] == ["nginx.conf", "conf.d/http.conf"]
assert section["servers"][0]["listen"] == [{"addr": None, "port": 8123}]
assert errors == []
def test_a_relative_nginx_config_path_is_taken_as_written(tmp_path: Path, monkeypatch) -> None:
"""nginx's own `-c` resolves a relative path against the working directory, so this one does
too — only the absolute default is read under `--root`."""
(tmp_path / "nginx.conf").write_text("http { server { listen 7; } }\n")
monkeypatch.chdir(tmp_path)
errors: list[str] = []
section = host_agent.Collector("/", nginx_config="nginx.conf").nginx(errors)
assert section["available"] is True
assert section["servers"][0]["listen"] == [{"addr": None, "port": 7}]
assert errors == []
def test_two_include_lines_naming_one_file_read_it_once(tmp_path: Path) -> None:
"""`include conf.d/http.conf;` and `include conf.d/*.conf;` name the same file, and reading
it twice would double every server block it holds."""
root = tmp_path / "root"
nginx_tree(root, stream=None, main="http {\n"
" include /etc/nginx/conf.d/http.conf;\n"
" include /etc/nginx/conf.d/*.conf;\n"
"}\n")
errors: list[str] = []
section = host_agent.Collector(str(root)).nginx(errors)
assert section["available"] is True
assert section["files"] == ["/etc/nginx/nginx.conf", "/etc/nginx/conf.d/http.conf"]
# two vhosts in http.conf, and not four
assert [entry["port"] for server in section["servers"] for entry in server["listen"]] == [
4443, 4443, 8443, 8443]
assert errors == []
def test_the_include_paths_of_a_config_that_will_not_parse_are_simply_none() -> None:
"""The collector asks this before it can report anything, so a broken file must answer
empty rather than raise — and the parse failure is then said once, by the parser."""
assert host_agent.nginx_include_paths("http { server { listen 80;\n") == []
assert host_agent.nginx_include_paths("http { include /a/*.conf; }") == ["/a/*.conf"]
def test_no_proxy_removes_the_nginx_section_and_says_nothing(tmp_path: Path) -> None:
"""The `--no-smart` precedent, on the second section: opting out is silent, and "not
collected" stays distinguishable from "collected and empty" because the key is absent."""
root = tmp_path / "root"
nginx_tree(root)
snap = host_agent.Collector(str(root), proxy=False).snapshot()
assert "nginx" not in snap, "--no-proxy removes the key, not just its contents"
assert core_errors(snap) == []
def test_the_nginx_section_reaches_the_command_line(tmp_path: Path, capsys) -> None:
"""A flag nobody wired is a flag that does nothing: `--once` proves the CLI path for both
the path argument and the opt-out."""
config = tmp_path / "nginx.conf"
config.write_text('http { server { listen 7778; server_name cli.test; } }\n')
assert host_agent.main(["--once", "--nginx-config", str(config)]) == 0
snap = json.loads(capsys.readouterr().out)
assert snap["ok"] is True
assert snap["nginx"]["available"] is True
assert snap["nginx"]["servers"][0]["listen"] == [{"addr": None, "port": 7778}]
assert host_agent.main(["--once", "--no-proxy"]) == 0
assert "nginx" not in json.loads(capsys.readouterr().out)
def test_the_nginx_section_is_documented_in_the_schema() -> None:
"""`GET /schema` is how an install documents itself, so the section an operator is looking at
has to be named there with every key it carries — the `haproxy` block's precedent."""
assert schema_documents("nginx")
doc = host_agent.__doc__ or ""
block = doc.split(" nginx ", 1)[-1].split(" errors ", 1)[0]
assert block, "the nginx section has no SCHEMA block to document it"
for key in ("available", "config", "files", "servers", "listen", "addr", "port",
"server_names", "proxy_passes", "location", "scheme", "target", "variable",
"upstreams", "maps", "entries", "key"):
assert key in block, f"nginx.{key} missing from the SCHEMA block"
assert "--no-proxy" in block and "--nginx-config" in block, \
"the two flags that decide whether this section exists belong in its documentation"
# ---- live sockets: the additive `sockets` section (phase 28) ---------------- #
#: A fragment of `/proc/net/tcp` as the deployed kernel writes it: the header verbatim, the
#: column order and spacing verbatim, the addresses documentation space (RFC 5737) and the
#: hexadecimals computed from them the way `/proc` computes them — one 32-bit word, printed
#: little-endian. `tests/fixtures/gateway_proc_net_tcp{,6}.txt` is the verbatim capture; this is
#: the pinned shape with the values a test can assert, the `smartctl_reference_*` precedent.
#: Rows: LISTEN 0.0.0.0:80, ESTABLISHED 10.1.0.5:443 <- 203.0.113.7:50000, that same pair in
#: TIME_WAIT, and LISTEN 127.0.0.1:8082.
REFERENCE_PROC_NET_TCP = """\
sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode
0: 00000000:0050 00000000:0000 0A 00000000:00000000 00:00000000 00000000 0 0 4314 1 0000000000000000 100 0 0 10 0
1: 0500010A:01BB 077100CB:C350 01 00000000:00000000 00:00000000 00000000 1000 0 5122 1 00000000d22dedba 20 4 28 10 -1
2: 0500010A:01BB 077100CB:C350 06 00000000:00000000 00:00000000 00000000 0 0 0 3 0000000000000000 20 0 0 10 0
3: 0100007F:1F92 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 4389 1 0000000000000000 100 0 0 10 0
"""
#: The v6 half, same shape: LISTEN `::`:443 (what haproxy writes as `bind :::443 v4v6`), an
#: ESTABLISHED pair between two documentation-space v6 addresses (RFC 3849), the same pair in
#: TIME_WAIT, and LISTEN `::ffff:127.0.0.1:8082` — the IPv4-mapped spelling a v6 socket uses,
#: which stays a v6 row instead of being folded into the v4 table.
REFERENCE_PROC_NET_TCP6 = """\
sl local_address remote_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode
0: 00000000000000000000000000000000:01BB 00000000000000000000000000000000:0000 0A 00000000:00000000 00:00000000 00000000 0 0 4315 1 0000000000000000 100 0 0 10 0
1: B80D0120000000000000000005000000:20FB B80D0120000001000000000009000000:CE41 01 00000000:00000000 00:00000000 00000000 1000 0 5130 1 00000000d22dedba 20 4 28 10 -1
2: B80D0120000000000000000005000000:20FB B80D0120000001000000000009000000:CE41 06 00000000:00000000 00:00000000 00000000 0 0 0 3 0000000000000000 20 0 0 10 0
3: 0000000000000000FFFF00000100007F:1F92 00000000000000000000000000000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 4390 1 0000000000000000 100 0 0 10 0
"""
def sockets_tree(root: Path, tcp: str = REFERENCE_PROC_NET_TCP,
tcp6: str | None = REFERENCE_PROC_NET_TCP6) -> Path:
"""A proc tree whose only interesting content is the two socket tables."""
write_file(root, "proc/stat", "cpu 9000 0 1000 80000\n")
write_file(root, "proc/meminfo", "MemTotal: 8000000 kB\nMemAvailable: 5000000 kB\n")
write_file(root, "proc/net/tcp", tcp)
if tcp6 is not None:
write_file(root, "proc/net/tcp6", tcp6)
return root
def test_a_captured_proc_net_tcp_fragment_parses_into_listens_and_established() -> None:
"""The v4 table's two facts, exactly, with `/proc`'s own hexadecimals."""
listens, established = host_agent.parse_proc_net_tcp(REFERENCE_PROC_NET_TCP, "v4")
assert listens == [
{"family": "v4", "addr": "0.0.0.0", "port": 80},
{"family": "v4", "addr": "127.0.0.1", "port": 8082},
]
assert established == [
{"family": "v4",
"local": {"addr": "10.1.0.5", "port": 443},
"remote": {"addr": "203.0.113.7", "port": 50000}},
]
@pytest.mark.parametrize("state", ["02", "03", "04", "05", "06", "07", "08", "09", "0B", "0C"])
def test_a_socket_in_transit_is_never_reported_as_a_pipe(state: str) -> None:
"""Every kernel state but LISTEN and ESTABLISHED is a socket that is not a pipe.
`06` is TIME_WAIT, which is what a proxy's finished client connections turn into by the
thousand: reporting them would draw two pipes for one connection and bury the live pairs
under rows that stopped being true before the page painted. The kernel's own list
(`net/tcp_states.h`) is what this walks, so a state added upstream is skipped too.
"""
line = (f" 0: 0500010A:01BB 077100CB:C350 {state} 00000000:00000000 00:00000000"
" 00:00000000 00000000 1000 0 5122 1 00000000d22dedba 20 4 28 10 -1")
assert host_agent.parse_proc_net_tcp(line, "v4") == ([], [])
def test_the_captured_fragment_actually_holds_the_row_it_omits() -> None:
"""The TIME_WAIT assertion above is only worth anything if the capture has one to omit."""
states = [line.split()[3] for line in REFERENCE_PROC_NET_TCP.splitlines()[1:]
if len(line.split()) > 3]
assert states == ["0A", "01", "06", "0A"], "the v4 fragment lost its socket in transit"
v6_states = [line.split()[3] for line in REFERENCE_PROC_NET_TCP6.splitlines()[1:]
if len(line.split()) > 3]
assert v6_states == ["0A", "01", "06", "0A"]
listens, established = host_agent.parse_proc_net_tcp(REFERENCE_PROC_NET_TCP, "v4")
assert len(established) == 1, "the TIME_WAIT row leaked into the established list"
assert len(listens) == 2
def test_a_captured_proc_net_tcp6_fragment_unrolls_its_little_endian_words() -> None:
"""v6 is four 32-bit words, each printed little-endian; the parser must unroll each group.
A parser that read the hex as written would print `b80d:120:…` — a plausible-looking address
that is not the one the kernel reported, which is worse than a gap beside a real socket.
"""
listens, established = host_agent.parse_proc_net_tcp(REFERENCE_PROC_NET_TCP6, "v6")
assert listens == [
{"family": "v6", "addr": "::", "port": 443},
{"family": "v6", "addr": "::ffff:127.0.0.1", "port": 8082},
]
assert established == [
{"family": "v6",
"local": {"addr": "2001:db8::5", "port": 8443},
"remote": {"addr": "2001:db8:1::9", "port": 52801}},
]
def test_the_wildcard_listeners_keep_their_literal_wildcard_addresses() -> None:
"""The wildcard **is** the fact the inbound join keys on, so it is never nullled or starred.
haproxy's `bind :::80 v4v6` and nginx's bare `listen 80` both mean "any address on this
port", and the page reaches that fact by comparing the socket's address to the config's; a
`*`, a `null` or a dropped field would each lose every wildcard frontend on the gateway.
"""
v4_listens, _ = host_agent.parse_proc_net_tcp(REFERENCE_PROC_NET_TCP, "v4")
v6_listens, _ = host_agent.parse_proc_net_tcp(REFERENCE_PROC_NET_TCP6, "v6")
assert v4_listens[0]["addr"] == "0.0.0.0"
assert v6_listens[0]["addr"] == "::"
assert "*" not in str(v4_listens) and "*" not in str(v6_listens)
def test_an_endpoint_that_is_not_an_address_is_null_and_never_a_guess() -> None:
"""A short word, a non-hex word and a non-numeric port each cost a field, not a row."""
assert host_agent._proc_net_endpoint("0100007F:1F92", "v4") == ("127.0.0.1", 8082)
assert host_agent._proc_net_endpoint("0100007:1F92", "v4") == (None, 8082)
assert host_agent._proc_net_endpoint("010000ZZ:1F92", "v4") == (None, 8082)
assert host_agent._proc_net_endpoint("0100007F:nope", "v4") == ("127.0.0.1", None)
assert host_agent._proc_net_endpoint("local_address", "v4") == (None, None)
listens, _ = host_agent.parse_proc_net_tcp(
" 0: local_address rem_address st rest\n"
" 1: 00000000:0050 00000000:0000 0A rest\n", "v4")
assert listens == [{"family": "v4", "addr": "0.0.0.0", "port": 80}], \
"/proc's own header line must not become a socket"
def test_an_address_this_platform_cannot_render_is_a_gap_and_not_a_crash(
monkeypatch: pytest.MonkeyPatch) -> None:
"""`socket.inet_ntop` is the one call here the agent does not control.
It exists on every Linux the agent runs on, and this pins what happens if it ever refuses:
the address becomes `null` and the socket row stays, because a snapshot path may not raise
over a rendering question, and the port — the half the join actually needs — is still true.
"""
def refuse(_family: int, _bytes: bytes) -> str:
raise OSError("no inet_ntop here")
monkeypatch.setattr(host_agent.socket, "inet_ntop", refuse)
listens, _ = host_agent.parse_proc_net_tcp(REFERENCE_PROC_NET_TCP, "v4")
assert listens[0] == {"family": "v4", "addr": None, "port": 80}
def test_the_socket_tables_are_read_from_the_proc_tree(tmp_path: Path) -> None:
"""The section over a tree, not just over text: `--root` resolves `/proc/net/tcp`."""
root = tmp_path / "root"
sockets_tree(root)
errors: list[str] = []
section = host_agent.Collector(str(root)).sockets(errors)
assert section["available"] is True
assert section["listens"] == [
{"family": "v4", "addr": "0.0.0.0", "port": 80},
{"family": "v4", "addr": "127.0.0.1", "port": 8082},
{"family": "v6", "addr": "::", "port": 443},
{"family": "v6", "addr": "::ffff:127.0.0.1", "port": 8082},
], "the v4 rows come first and the family label says which table each came from"
assert len(section["established"]) == 2
def test_a_tree_without_the_socket_tables_is_null_with_one_errors_line(tmp_path: Path) -> None:
"""Neither table answering is the section's own `null` plus **one** line, nothing else.
A snapshot is not the place to discover that `/proc` is missing, and one line is the whole
report: the other sections still have to be readable and the system source still has to be
`ok` (the `sockets` prefix is in the dashboard's add-on list for exactly that).
"""
root = tmp_path / "root"
sockets_tree(root, tcp="", tcp6=None)
(root / "proc" / "net" / "tcp").unlink()
errors: list[str] = []
section = host_agent.Collector(str(root)).sockets(errors)
assert section == {"available": None, "listens": [], "established": []}
assert len(errors) == 1, errors
assert errors[0].startswith("sockets:")
assert "/proc/net/tcp" in errors[0] and "/proc/net/tcp6" in errors[0]
assert "--no-proxy" in errors[0]
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_the_socket_tables_reach_a_snapshot_and_nothing_else_moves(tmp_path: Path) -> None:
"""The section is additive in the strongest sense: adding the tables changes no other field.
A snapshot is a dozen independent reads, and the cheapest proof that a new one is a guest
rather than a renovation is to take the same tree twice — once with the socket tables, once
without — and demand every other key byte-identical. `ts` and `duration_ms` are the two
fields a snapshot owns, so they are the only ones allowed to differ besides `sockets`.
"""
with_tables = tmp_path / "with"
sockets_tree(with_tables)
without = tmp_path / "without"
sockets_tree(without)
(without / "proc" / "net" / "tcp").unlink()
(without / "proc" / "net" / "tcp6").unlink()
first = host_agent.Collector(str(with_tables)).snapshot()
second = host_agent.Collector(str(without)).snapshot()
assert first["sockets"]["available"] is True
assert second["sockets"]["available"] is None
for key in first:
if key in ("ts", "duration_ms", "sockets", "errors"):
continue
assert first[key] == second[key], f"{key} moved when the socket section appeared"
# the gap line is the add-on's own voice: it reaches `errors` and nothing else
assert any(line.startswith("sockets:") for line in second["errors"])
assert not any(line.startswith("sockets:") for line in first["errors"])
def test_an_unreadable_socket_table_costs_one_line_while_the_other_still_reports(
tmp_path: Path) -> None:
"""Partial coverage is `available: true` plus one line: the host said something worth saying."""
root = tmp_path / "root"
sockets_tree(root)
(root / "proc" / "net" / "tcp6").chmod(0)
errors: list[str] = []
section = host_agent.Collector(str(root)).sockets(errors)
assert section["available"] is True
assert [row["family"] for row in section["listens"]] == ["v4", "v4"]
assert len(errors) == 1, errors
assert errors[0].startswith("sockets:") and "/proc/net/tcp6" in errors[0]
assert "not found" not in errors[0], "a denied file is not a missing one"
def test_a_missing_second_table_costs_nothing_when_the_first_answered(tmp_path: Path) -> None:
"""A kernel built without IPv6 has no `/proc/net/tcp6` at all.
That is the host telling us what it has, so the section is `available: true` and the pass is
silent — the nginx rule that an include naming a file this host does not have contributes
nothing and no line, and the reason a fleet-wide agent rollout does not put a line in every
IPv4-only container's snapshot.
"""
root = tmp_path / "root"
sockets_tree(root, tcp6=None)
errors: list[str] = []
section = host_agent.Collector(str(root)).sockets(errors)
assert section["available"] is True
assert [row["family"] for row in section["listens"]] == ["v4", "v4"]
assert errors == []
def test_the_socket_read_is_bounded_and_says_so(tmp_path: Path) -> None:
"""A busy gateway writes one line per socket; the bound is what keeps that out of the poll.
Truncation is reported, not hidden: `available` stays `true`, the first rows are the rows the
section carries, and one line names the bound so the difference between "4096 sockets" and
"more sockets than this read takes" is on the wire (the SMART pass-budget shape).
"""
root = tmp_path / "root"
rows = [" sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt"
" uid timeout inode"]
rows += [f" {n}: 00000000:{n:04X} 00000000:0000 0A 00000000:00000000 00:00000000"
" 00000000 0 0 1 1 0000000000000000 100 0 0 10 0"
for n in range(1, 5000)]
sockets_tree(root, tcp="\n".join(rows) + "\n", tcp6=None)
errors: list[str] = []
section = host_agent.Collector(str(root)).sockets(errors)
assert section["available"] is True
assert len(section["listens"]) == host_agent.PROC_NET_MAX_LINES - 1, \
"the bound counts the lines read, /proc's own header among them"
assert [row["port"] for row in section["listens"][:2]] == [1, 2]
assert section["listens"][-1]["port"] == host_agent.PROC_NET_MAX_LINES - 1
assert len(errors) == 1, errors
assert errors[0].startswith("sockets:") and "/proc/net/tcp" in errors[0]
assert str(host_agent.PROC_NET_MAX_LINES) in errors[0]
def test_no_proxy_removes_the_sockets_section_and_says_nothing(tmp_path: Path) -> None:
"""A box with no proxy has nothing to watch from either side, so opting out is silent."""
root = tmp_path / "root"
sockets_tree(root)
snap = host_agent.Collector(str(root), proxy=False).snapshot()
assert "sockets" not in snap, "--no-proxy removes the key, not just its contents"
assert core_errors(snap) == []
def test_a_sockets_failure_inside_the_collector_still_yields_a_section(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""The `_safe` wrapper is the section's last line of defence, like the two configs'"""
root = tmp_path / "root"
sockets_tree(root)
collector = host_agent.Collector(str(root))
def explode(errors: list[str]) -> dict:
raise RuntimeError("kaboom")
monkeypatch.setattr(collector, "sockets", explode)
snap = collector.snapshot()
assert snap["sockets"] == {"available": None, "listens": [], "established": []}
assert any(line.startswith("sockets:") for line in snap["errors"])
def test_the_sockets_section_reaches_the_command_line(tmp_path: Path, capsys) -> None:
"""`--once` proves the CLI path, including the opt-out the operator actually uses."""
root = tmp_path / "root"
sockets_tree(root)
assert host_agent.main(["--once", "--root", str(root)]) == 0
snap = json.loads(capsys.readouterr().out)
assert snap["ok"] is True
assert snap["sockets"]["available"] is True
assert {"family": "v6", "addr": "::", "port": 443} in snap["sockets"]["listens"]
assert host_agent.main(["--once", "--root", str(root), "--no-proxy"]) == 0
assert "sockets" not in json.loads(capsys.readouterr().out)
def test_the_sockets_section_is_documented_in_the_schema() -> None:
"""`GET /schema` is how an install documents itself; the two configs set the precedent."""
assert schema_documents("sockets")
doc = host_agent.__doc__ or ""
block = doc.split(" sockets ", 1)[-1].split(" errors ", 1)[0]
assert block, "the sockets section has no SCHEMA block to document it"
for key in ("available", "listens", "established", "family", "addr", "port",
"local", "remote"):
assert key in block, f"sockets.{key} missing from the SCHEMA block"
assert "--no-proxy" in block, "the flag that decides whether this section exists belongs here"
for table in ("/proc/net/tcp", "/proc/net/tcp6"):
assert table in block, f"{table} is what the section reads; name it in the schema"
@linux_only
def test_the_real_proc_net_tcp_parses_on_the_box_running_pytest() -> None:
"""The parser against the kernel that is actually here.
The shape is what this pins: every row carries its family, a port that parsed is a plausible
one, and a wildcard listener is still a wildcard. Whether this machine has any listeners at
all is not a claim a test may make, so it does not.
"""
collector = host_agent.Collector("/")
errors: list[str] = []
section = collector.sockets(errors)
assert section["available"] is True, "a Linux box with /proc mounted must answer for sockets"
assert errors == []
for row in section["listens"]:
assert row["family"] in ("v4", "v6")
assert row["port"] is None or 0 <= row["port"] <= 65535
if row["port"] is not None:
assert row["addr"] is not None, "a decoded port from an undecodable address is drift"
for row in section["established"]:
assert set(row) == {"family", "local", "remote"}
assert set(row["local"]) == set(row["remote"]) == {"addr", "port"}
# ---- 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", "haproxy", "nginx",
"sockets", "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.
# So are the proxy ones — most boxes have no haproxy, and saying so is the add-on's
# `available: false` arm, not a degraded section. So is `sockets:` on a box whose /proc has
# no net tables mounted, which is the same "unconfigured, not broken" sentence.
hard_errors = core_errors(snap)
assert hard_errors == [], f"degraded sections must be empty on a healthy box: {hard_errors}"
# `files` is the config's own paths: the main config first when it was read, empty when not
if snap["nginx"]["available"] is True:
assert snap["nginx"]["files"][0] == snap["nginx"]["config"]
else:
assert snap["nginx"]["files"] == []
# the add-on is present by default, and its own verdict is one of the three arms
assert snap["haproxy"]["available"] in (True, False, None)
assert snap["haproxy"]["config"] == host_agent.HAPROXY_CONFIG_PATH
assert snap["nginx"]["available"] in (True, False, None)
assert snap["nginx"]["config"] == host_agent.NGINX_CONFIG_PATH
# the socket table is present by default too, and answers for itself: /proc/net/tcp is not
# something a host can leave uninstalled, so the arms here are "read" and "did not read"
assert snap["sockets"]["available"] in (True, None)
for row in snap["sockets"]["listens"]:
assert row["family"] in ("v4", "v6")
assert row["port"] is None or 0 <= row["port"] <= 65535
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}"
# `fnmatch` + `glob` (phase 28) join the list on purpose: an nginx `include` line is a glob,
# and both are as old as the 3.10 floor and as dependency-free as `re`.
assert top <= {"__future__", "argparse", "collections", "contextlib", "dataclasses", "fnmatch",
"glob", "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_tls_fields_are_documented_in_the_schema() -> None:
"""The phase-15 fields are documented in the docstring, which is what `GET /schema` answers.
The version constant itself is pinned by the newest phase's test below — one place names the
number, so a bump moves one assertion and not a file's worth of history.
"""
assert schema_documents("tls") and schema_documents("bundle_rev")
assert "--cert-bundle" in (host_agent.__doc__ or "")
def test_1_2_0_is_the_version_that_ships_the_three_proxy_sections() -> None:
"""The version bump is the fleet's way of saying "this file has the gateway sections".
Phase 28 added three sections and one flag to the payload, so an install that predates them
is distinguishable by one string: `--status`, `/health` and this test all read the same
constant. The SCHEMA block is `GET /schema`'s body, so it names all three, each with the
three-way `available` the dashboard reads, and `--no-proxy`, the flag that removes the family.
"""
assert host_agent.AGENT_VERSION == "1.2.0"
doc = host_agent.__doc__ or ""
for section in ("haproxy", "nginx", "sockets"):
assert schema_documents(section), f"{section} is not documented in the SCHEMA block"
block = doc.split(f" {section} ", 1)[-1]
assert "available" in block, f"{section}'s block does not document its three-way answer"
assert "--no-proxy" in doc and "--haproxy-config" in doc and "--nginx-config" in doc
# ---- the recorded gateway: the captures themselves (phase 28 task 04) -------------------- #
#: Every document `scripts/capture-gateway-config.sh` records, at the tree path the agent reads
#: it from and with the fixture it came from. This is the phase's ground truth: the synthetic
#: shapes above are edge cases with pinned labels, and these are the box (invariant 7).
GATEWAY_CAPTURES = {
"etc/haproxy/haproxy.cfg": "gateway_haproxy.cfg",
"etc/nginx/nginx.conf": "gateway_nginx_main.conf",
"etc/nginx/conf.d/http.conf": "gateway_nginx_conf_d_http.conf",
"etc/nginx/stream.d/stream.conf": "gateway_nginx_stream_d_stream.conf",
"proc/net/tcp": "gateway_proc_net_tcp.txt",
"proc/net/tcp6": "gateway_proc_net_tcp6.txt",
}
#: What each captured document was read from, so a test can say which file to re-record.
GATEWAY_SOURCES = {
"gateway_haproxy.cfg": "/etc/haproxy/haproxy.cfg",
"gateway_nginx_main.conf": "/etc/nginx/nginx.conf",
"gateway_nginx_conf_d_http.conf": "/etc/nginx/conf.d/http.conf",
"gateway_nginx_stream_d_stream.conf": "/etc/nginx/stream.d/stream.conf",
"gateway_proc_net_tcp.txt": "/proc/net/tcp",
"gateway_proc_net_tcp6.txt": "/proc/net/tcp6",
}
def gateway_tree(root: Path) -> Path:
"""A fake root carrying the recorded gateway: the six captures at their real paths.
`mime.types` is deliberately absent, because it is absent on the host too — the main config
includes it and the agent's answer for that is one include line that costs nothing.
"""
write_file(root, "proc/stat",
"cpu 100 0 50 800 0 0 0 0 0 0\ncpu0 100 0 50 800 0 0 0 0 0 0\n")
write_file(root, "proc/meminfo", "MemTotal: 8000000 kB\nMemFree: 1000000 kB\n"
"MemAvailable: 4000000 kB\n")
for tree_path, fixture in GATEWAY_CAPTURES.items():
write_file(root, tree_path, (FIXTURES / fixture).read_text())
return root
def gateway_collector(root: Path) -> Any:
return host_agent.Collector(str(root), smartctl="/nonexistent/smartctl")
def test_the_gateway_captures_are_captures_and_say_so() -> None:
"""Each fixture names the remote path it came from and the script that read it.
That header is the difference between a capture and a file somebody typed, and the parsers
skip it — `#` is a comment in both config formats, and a `/proc` line whose 4th field is not
a kernel state is not a socket row — so provenance costs nothing on either side.
"""
for fixture, source in GATEWAY_SOURCES.items():
path = FIXTURES / fixture
assert path.is_file(), f"{fixture} is missing: run ./scripts/capture-gateway-config.sh"
head = path.read_text().splitlines()[:10]
assert head[0].startswith(f"# Verbatim {source} from "), head[:1]
assert any("capture-gateway-config.sh" in line for line in head), head
def test_the_recorded_gateway_nginx_tree_merges_the_documents_it_names(tmp_path: Path) -> None:
"""The real main config plus the real vhosts, through the real include pass.
`files` is the answer an operator can check against `nginx -t`, and the map is the SNI table
the whole point of the stream block: the map's *values* are the outbound targets, and the
server beside it has no target at all because the map decides per connection.
"""
errors: list[str] = []
section = gateway_collector(gateway_tree(tmp_path / "root")).nginx(errors)
assert section["available"] is True, errors
assert errors == [], f"the recorded tree should cost no error line: {errors}"
assert section["files"] == ["/etc/nginx/nginx.conf",
"/etc/nginx/stream.d/stream.conf",
"/etc/nginx/conf.d/http.conf"], \
"the merge set changed: re-run ./scripts/capture-gateway-config.sh"
assert len(section["servers"]) >= 10, "a gateway with a handful of vhosts is not this gateway"
maps = {cast("dict[str, Any]", m)["variable"]: m for m in section["maps"]}
assert "targetBackend" in maps and "connection_upgrade" in maps, list(maps)
entries = {cast("str", cast("dict[str, Any]", e)["key"]): cast("dict[str, Any]", e)
for e in maps["targetBackend"]["entries"]}
brain = entries["brain.reeseapps.com"]["target"]
assert brain == {"host": "reeseapps-brain-of-reese.reeselink.com", "port": 443}, brain
assert entries["default"]["target"] == {"host": "127.0.0.1", "port": 4443}, entries["default"]
# `map $http_upgrade` answers header words, and a word is not a host (rule 2).
upgrade = {cast("str", cast("dict[str, Any]", e)["key"]): cast("dict[str, Any]", e)["target"]
for e in maps["connection_upgrade"]["entries"]}
assert upgrade == {"default": None, "": None}, upgrade
def test_the_recorded_gateway_routes_by_sni_without_inventing_a_target(tmp_path: Path) -> None:
"""One vhost and the stream server, named: what the phase-29 page draws its pipes from."""
errors: list[str] = []
section = gateway_collector(gateway_tree(tmp_path / "root")).nginx(errors)
servers = [cast("dict[str, Any]", s) for s in section["servers"]]
def ports(server: dict[str, Any]) -> set[int | None]:
return {cast("dict[str, Any]", listen)["port"] for listen in server["listen"]}
nextcloud = [s for s in servers
if "nextcloud.reeseapps.com" in cast("list[str]", s["server_names"])
and 4443 in ports(s)]
assert nextcloud, "the recorded vhost for nextcloud on 4443 is gone"
passes = [cast("dict[str, Any]", p) for p in nextcloud[0]["proxy_passes"]]
assert {"host": "nextcloud.reeselink.com", "port": 11000} in [
cast("dict[str, Any]", p)["target"] for p in passes], passes
stream = [s for s in servers
if ports(s) == {443} and not cast("list[str]", s["server_names"])]
assert stream, "the stream server that terminates :443 and routes by SNI is gone"
variable_pass = [cast("dict[str, Any]", p) for p in stream[0]["proxy_passes"]
if p["variable"] == "targetBackend"]
assert variable_pass and variable_pass[0]["target"] is None, \
"`proxy_pass $targetBackend` must travel as the variable, never as a merged target"
def in_documentation_space(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
"""RFC 5737 (`203.0.113.0/24`) or RFC 3849 (`2001:db8::/32`) — the ranges the capture uses."""
return (address in ipaddress.ip_network("203.0.113.0/24")
or address in ipaddress.ip_network("2001:db8::/32"))
def proc_data_lines(text: str) -> list[list[str]]:
"""The socket rows of a `/proc/net/tcp` capture: no kernel header, no provenance header.
The kernel numbers its rows (`0:`, `1:` …), which is the one test that separates a socket from
a line of comment or a column title — both of which have four fields too.
"""
rows = []
for line in text.splitlines():
fields = line.split()
if len(fields) > 4 and fields[0].endswith(":") and fields[0][:-1].isdigit():
rows.append(fields)
return rows
def test_the_recorded_socket_tables_decode_every_row_they_hold() -> None:
"""Real `/proc` bytes, every address and port of them resolved.
The agent's decoder is being read against a kernel's own output rather than a hand-written
fragment: a row whose address comes back `None` means the spelling changed under us, which is
exactly what a fixture is for. And the rows the parser must *not* report are here in volume —
a proxy's `TIME_WAIT` pile — so their absence from the answer is measured, not assumed.
"""
transit = 0
for fixture, family in (("gateway_proc_net_tcp.txt", "v4"), ("gateway_proc_net_tcp6.txt", "v6")):
text = (FIXTURES / fixture).read_text()
rows = proc_data_lines(text)
assert rows, f"{fixture} holds no socket row at all"
listens, established = host_agent.parse_proc_net_tcp(text, family)
parsed = [*listens, *established]
assert parsed, f"{fixture} produced no socket at all"
broken = [row for row in parsed
if any(endpoint.get("addr") is None or endpoint.get("port") is None
for endpoint in ([row] if "addr" in row
else [row["local"], row["remote"]]))]
assert not broken, f"{fixture} has a row /proc's spelling does not explain: {broken[:3]}"
states = [fields[3] for fields in rows]
kept = states.count(host_agent.PROC_NET_STATE_LISTEN) \
+ states.count(host_agent.PROC_NET_STATE_ESTABLISHED)
assert len(parsed) == kept, (
f"{fixture}: the parser returned {len(parsed)} rows for {kept} LISTEN/ESTABLISHED "
f"sockets (states present: {sorted(set(states))})")
transit += len(states) - kept
def test_the_recorded_configs_and_the_recorded_sockets_agree(tmp_path: Path) -> None:
"""The config claims frontends; the kernel agrees, and one listener belongs to neither.
Two independent reads of one box at one moment: haproxy's `bind`s and nginx's `listen`s are
ports the gateway is answering on, and the agent's own port is a listener no proxy config
names. That last one is the case phase 29 must keep visible rather than drop — a socket with
no claim is a fact, not a join failure.
"""
collector = gateway_collector(gateway_tree(tmp_path / "root"))
errors: list[str] = []
haproxy = collector.haproxy(errors)
nginx = collector.nginx(errors)
sockets = collector.sockets(errors)
assert errors == [], errors
assert haproxy["available"] is True and nginx["available"] is True
assert sockets["available"] is True
claimed: set[int] = set()
for frontend in cast("list[dict[str, Any]]", haproxy["frontends"]):
claimed |= {cast("int", b["port"]) for b in cast("list[dict[str, Any]]", frontend["binds"])}
for server in cast("list[dict[str, Any]]", nginx["servers"]):
claimed |= {cast("int", ls["port"]) for ls in cast("list[dict[str, Any]]", server["listen"])
if ls["port"] is not None}
listening = {cast("int", row["port"]) for row in
cast("list[dict[str, Any]]", sockets["listens"]) if row["port"] is not None}
missing = claimed - listening
assert not missing, (
f"the configs claim ports {sorted(missing)} nothing is listening on: the two captures "
"were taken apart, or a service was down — re-run ./scripts/capture-gateway-config.sh")
assert 9120 in listening - claimed, \
"the agent's own listener should be in the socket table and claimed by neither config"
def test_the_recorded_socket_tables_publish_no_globally_routable_endpoint() -> None:
"""AGENTS.md invariant 6, on the artifacts: identifiers never go in the repo.
The socket table is the one capture here that can hold somebody else's address, and on Linux
a global IPv6 is EUI-64 — it encodes the NIC's MAC, which is the identifier class
`capture-fixtures.sh` already refuses to record for a drive. Private, loopback, wildcard and
documentation addresses are all expected here; a globally routable one is a capture that got
away.
"""
offenders: list[str] = []
documented = 0
for fixture in ("gateway_proc_net_tcp.txt", "gateway_proc_net_tcp6.txt"):
for fields in proc_data_lines((FIXTURES / fixture).read_text()):
for column in (fields[1], fields[2]):
word = column.rpartition(":")[0]
packed = bytes.fromhex(word)
raw = b"".join(packed[i:i + 4][::-1] for i in range(0, len(packed), 4))
address = ipaddress.ip_address(raw)
# Only an IPv6Address has `ipv4_mapped` (3.14 added `ipv6_mapped` beside it).
subject = getattr(address, "ipv4_mapped", None) or address
if subject.is_global:
offenders.append(f"{fixture}: {subject}")
if in_documentation_space(subject):
documented += 1
assert not offenders, ("globally routable endpoints are in a tracked fixture: "
+ ", ".join(offenders)
+ " — scripts/capture-gateway-config.sh is the thing that stops them")
assert documented, ("no documentation-space address either, which means this fixture no "
"longer pins the substitution the capture makes")
def test_the_recorded_gateway_answers_all_three_sections_at_once(tmp_path: Path) -> None:
"""The phase's end state in one harness: one 1.2.0-shaped read of the captured gateway.
`haproxy` and `nginx` from the config tree, `sockets` from the kernel tables, all three in
one snapshot, nothing in `errors`, and the old sections exactly as they were — which is what
phase 29 builds its page on.
"""
snap = gateway_collector(gateway_tree(tmp_path / "root")).snapshot()
assert snap["agent_version"] == host_agent.AGENT_VERSION
for section in ("haproxy", "nginx", "sockets"):
assert snap[section]["available"] is True, snap["errors"]
assert len(snap["haproxy"]["frontends"]) == 6, "the recorded gateway has six frontends"
assert len(snap["haproxy"]["backends"]) == 6, "and one backend each"
assert snap["nginx"]["maps"], "the SNI map is the point of the nginx capture"
assert snap["sockets"]["listens"], "the socket capture must answer listeners"
assert core_errors(snap) == [], snap["errors"]
# The old sections are untouched by the add-on, which is the whole compatibility claim.
for section in ("host", "cpu", "memory", "filesystems", "disk_io", "network",
"temperatures", "processes", "errors"):
assert section in snap, f"the pre-1.2.0 payload lost {section}"