dashboards/tests/test_deploy_scripts.py

"""Contract tests for the deployment scripts in `deploy/`.

There is no shell test framework here — and adding one would buy nothing a subprocess does
not — but phase 10 rests on a sentence that lives in **two** files: the sudoers grant in
`deploy/agent-sudoers`, and the hint `deploy/install-agent.sh` prints when it is run without
root by the unprivileged `ops` account. Two copies of one instruction is exactly what drifts
silently, and the failure is an install that cannot happen on a host nobody is sitting at.
So these tests read the file, run the script the way the operator does (unprivileged, which
exits before the script can touch anything), and require the two lines to be the same line
apart from the path the script was invoked through.
"""

from __future__ import annotations

import difflib
import json
import os
import re
import shutil
import stat
import subprocess
import threading
from http.server import ThreadingHTTPServer
from pathlib import Path
from typing import Any

import pytest

from dashboards import certs

ROOT = Path(__file__).resolve().parent.parent
DEPLOY = ROOT / "deploy"
INSTALLER = DEPLOY / "install-agent.sh"
SUDOERS = DEPLOY / "agent-sudoers"
BASH = shutil.which("bash") or "/bin/bash"

#: The grant as the file documents it. `/home/ops/agent-install` is the canonical staging
#: path: a monitored host has no repo checkout, so `deploy/` and `agent/` are copied there
#: (phase 10 task 02) and every host ends up with the same absolute path to grant.
CANONICAL_GRANT = "ops ALL=(root) NOPASSWD: /home/ops/agent-install/deploy/install-agent.sh"

#: The prefix a correct hint line starts with — user, run-as, and NOPASSWD, everything but
#: the path, which is the one thing the installer is allowed to substitute.
GRANT_PREFIX = "ops ALL=(root) NOPASSWD: "

#: Everything a non-root operator needs to hear, besides the grant line itself.
HINT_MUST_SAY = (
    "visudo -cf",  # check the syntax before installing it
    "install -m 0440 -o root -g root",  # and install it the way sudoers files are installed
    "/etc/sudoers.d/dashboards-agent",  # the destination both files name
    "deploy/agent-sudoers",  # the pointer back to the documented file
)


def _run(args: list[str], cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
    return subprocess.run(args, capture_output=True, text=True, cwd=cwd, check=False)


def _grant_lines(text: str) -> list[str]:
    """The sudoers *rules* in a file — comments start with `#`, so this never matches prose."""
    return [line for line in text.splitlines() if line.startswith(GRANT_PREFIX)]


def _installer_without_root(*flags: str) -> subprocess.CompletedProcess[str]:
    """Run the installer the way `ops` does.

    Skipped as root on purpose: root would go past `need_root` and really install a systemd
    unit on the machine running pytest.
    """
    if hasattr(os, "geteuid") and os.geteuid() == 0:
        pytest.skip("running as root — the installer would install a real unit here")
    return _run([BASH, str(INSTALLER), *flags])


# --- deploy/agent-sudoers ---------------------------------------------------------------


def test_agent_sudoers_exists_and_is_a_config_file_not_a_script() -> None:
    assert SUDOERS.is_file(), "deploy/agent-sudoers is missing (phase 10 task 01)"
    assert not os.access(SUDOERS, os.X_OK), "a sudoers file is data; only install-agent.sh runs"


def test_agent_sudoers_is_exactly_one_grant_line_at_the_canonical_path() -> None:
    """One rule, the installer, the staging path — nothing else may be granted."""
    lines = _grant_lines(SUDOERS.read_text())
    assert lines == [CANONICAL_GRANT], (
        "deploy/agent-sudoers must grant exactly one command, the staged installer:\n" + "\n".join(lines)
    )
    assert "/home/ops/agent-install" in SUDOERS.read_text()


def test_agent_sudoers_grants_no_shell_editor_or_second_path() -> None:
    """The 'deliberately NOT granted' block has to stay a comment, not become a rule."""
    text = SUDOERS.read_text()
    not_granted = text.split("deliberately NOT granted", 1)[-1]
    for never in ("visudo", "sh", "tee", "systemctl", "usermod", "upscmd", "upsrw"):
        assert never in not_granted, f"the file must name {never!r} as deliberately not granted"
    assert not_granted.lstrip().startswith("-"), "the block heading is followed by the list"
    # Every line of that block is a comment: a stray rule there would be a second grant.
    assert _grant_lines(not_granted) == []
    # And it names the token file, because "we did not grant reading it" is the point.
    assert "/etc/dashboards-agent.token" in not_granted


def test_agent_sudoers_documents_the_why_and_the_two_stage_install() -> None:
    """House style (deploy/nut-sudoers): why it is needed, then check-first install steps."""
    text = SUDOERS.read_text()
    assert "WHY THIS IS NEEDED" in text
    for step in HINT_MUST_SAY:
        assert step in text, f"deploy/agent-sudoers must document {step!r}"
    # The installed copy is checked too — the file being valid is not the same as the
    # installed file being valid (nut-sudoers' third step, and the reason stays).
    assert text.count("visudo -cf") >= 2


@pytest.mark.skipif(shutil.which("visudo") is None, reason="visudo is not installed on this box")
def test_agent_sudoers_passes_the_syntax_check_it_documents() -> None:
    """The file tells the operator to run `visudo -cf` on it, so it must survive that."""
    result = _run([shutil.which("visudo") or "/usr/sbin/visudo", "-cf", str(SUDOERS)])
    assert result.returncode == 0, f"visudo -cf failed:\n{result.stdout}{result.stderr}"


# --- deploy/install-agent.sh without root -----------------------------------------------


def test_installer_without_root_exits_1_and_prints_the_grant_line() -> None:
    result = _installer_without_root("--bind", "127.0.0.1")
    assert result.returncode == 1, "the non-root refusal keeps its exit code (1)"
    assert result.stdout == "", "the refusal is a failure: it goes to stderr, not stdout"
    lines = _grant_lines(result.stderr)
    assert lines == [f"{GRANT_PREFIX}{Path(os.path.realpath(INSTALLER))}"], (
        "the hint must print one grant line naming this script's resolved path:\n" + result.stderr
    )


def test_hint_and_the_sudoers_file_are_the_same_line_apart_from_the_path() -> None:
    """The pinned version of the diff in data/logs/NOTES-sudoers.md."""
    printed = _installer_without_root("--bind", "127.0.0.1").stderr
    hint = _grant_lines(printed)[0]
    canonical = _grant_lines(SUDOERS.read_text())[0]
    assert hint.startswith(GRANT_PREFIX) and canonical.startswith(GRANT_PREFIX)
    assert hint.removeprefix(GRANT_PREFIX).endswith("/deploy/install-agent.sh")
    assert hint.removeprefix(GRANT_PREFIX).removeprefix(str(DEPLOY.parent)) == (
        canonical.removeprefix(GRANT_PREFIX).removeprefix("/home/ops/agent-install")
    ), "same user, same (root), same NOPASSWD, same script — only the checkout path differs"


def test_hint_resolves_a_relative_invocation_to_an_absolute_path() -> None:
    """`bash deploy/install-agent.sh` from the repo root still prints a path sudo can match."""
    result = _run([BASH, "deploy/install-agent.sh", "--bind", "10.0.0.5"], cwd=ROOT)
    assert result.returncode == 1
    assert _grant_lines(result.stderr) == [f"{GRANT_PREFIX}{Path(os.path.realpath(INSTALLER))}"]


def test_installer_names_the_two_stage_install_and_points_at_the_file() -> None:
    stderr = _installer_without_root("--bind", "127.0.0.1").stderr
    for phrase in HINT_MUST_SAY:
        assert phrase in stderr, f"the non-root hint must say {phrase!r}, not just 'run as root'"
    # The pointer is the file next to *this* script, wherever the checkout lives.
    assert str(Path(os.path.realpath(SUDOERS))) in stderr


def test_installer_hint_repeats_the_command_that_was_asked_for() -> None:
    """Copy-pasteable: the hint ends with the same invocation, with the flags it was given."""
    stderr = _installer_without_root("--uninstall").stderr
    assert f"sudo {Path(os.path.realpath(INSTALLER))} --uninstall" in stderr


def test_installer_help_points_at_the_sudoers_grant() -> None:
    """The header comment *is* the help text, so the grant is discoverable from --help."""
    result = _run([BASH, str(INSTALLER), "--help"])
    assert result.returncode == 0
    assert "deploy/agent-sudoers" in result.stdout


# --- the tunnel machinery stays retired (phase 10) ---------------------------------------

#: Names and ports that only ever existed to serve the ssh-tunnel dev transport, which phase
#: 10 deleted. Every pattern is written as a character class (`[d]ev-…`) so that **this file
#: cannot match its own grep**: the phase criterion demands the repo stay free of those
#: literals, and a guard that reintroduces one to forbid it is the oldest trick there is.
#: Scope excludes `.agents/` (phase files and run reports are history — rewriting a finished
#: phase's record to satisfy a grep would falsify it) and `tests/fixtures/` (verbatim
#: upstream captures: a sector counter or a wheel hash may contain those digits by accident,
#: and rule 7 forbids editing a capture to please a grep).
TUNNEL_PATTERNS = ("[d]ev-tunnels", "DEV_[T]UNNELS", "DEV_[F]ORWARDS", r"912[123]\b", r"[3]494\b")

#: Where those names must not appear: every file that *tells someone what to run*.
#: `tests/fixtures/` is excluded (verbatim captures, see above) and so is `.agents/` —
#: phase files and run reports are history, and rewriting a finished phase's record to
#: satisfy a grep would falsify it.
TUNNEL_FREE_ROOTS = ("scripts", "deploy", "src", "agent", "web")


def _tunnel_scope_files() -> list[Path]:
    files: list[Path] = []
    for top in TUNNEL_FREE_ROOTS:
        for path in (ROOT / top).rglob("*"):
            if path.is_file() and path.suffix in (".py", ".sh", ".md", ".css", ".js", ".html"):
                files.append(path)
    files.append(ROOT / "README.md")
    files.append(ROOT / "AGENTS.md")
    files.append(ROOT / ".env.example")
    files.append(ROOT / ".env.dev.example")
    for path in (ROOT / "tests").glob("*.py"):
        files.append(path)
    return files


def test_dev_tunnel_script_is_deleted() -> None:
    """The tunnel script is gone, not shelved: git history keeps it if it is ever needed."""
    # Name built at runtime so this file never carries the literal the grep forbids.
    script = ROOT / "scripts" / ("dev-" + "tunnels.sh")
    assert not script.exists(), (
        "the retired ssh-tunnel dev script is back — phase 10 deleted it deliberately"
    )


def test_no_operational_file_mentions_the_tunnel_machinery() -> None:
    """A stale tunnel-env instruction is a setup step nobody can complete any more."""
    offenders: list[str] = []
    for path in _tunnel_scope_files():
        text = path.read_text(errors="replace")
        for pattern in TUNNEL_PATTERNS:
            if re.search(pattern, text):
                offenders.append(f"{path.relative_to(ROOT)}: /{pattern}/")
    assert not offenders, "tunnel-era names/ports are back:\n" + "\n".join(offenders)


# --- shell syntax (the only gate these scripts get) -------------------------------------


#: Names a shell function must never take. Bash resolves a **function** before an external
#: command of the same name, so defining `install()` in a script that then calls
#: `install -d /somewhere` is not a directory creation — it is the function calling itself,
#: forever. That is not hypothetical: `deploy/install-agent.sh` did exactly this, and the
#: first real install of phase 10 spun on the SMART block for two minutes and installed
#: nothing (no unit, no agent file, no token) while looking like a working script to `bash -n`.
SHADOW_FORBIDDEN = frozenset(
    {
        "install",
        "test",
        "time",
        "date",
        "sort",
        "grep",
        "sed",
        "tee",
        "type",
        "ulimit",
        "wait",
        "kill",
        "cd",
        "pwd",
    }
)

_FUNCTION_DEF_RE = r"^([A-Za-z_][A-Za-z0-9_]*)\s*\(\)"


def _function_names(script: Path) -> list[str]:
    return re.findall(_FUNCTION_DEF_RE, script.read_text(), flags=re.MULTILINE)


@pytest.mark.parametrize(
    "script",
    sorted([*DEPLOY.glob("*.sh"), *(ROOT / "scripts").glob("*.sh")]),
    ids=lambda p: str(p.relative_to(ROOT)),
)
def test_shell_script_has_valid_syntax(script: Path) -> None:
    result = _run([BASH, "-n", str(script)])
    assert result.returncode == 0, f"bash -n failed for {script}:\n{result.stderr}"


@pytest.mark.parametrize(
    "script",
    sorted([*DEPLOY.glob("*.sh"), *(ROOT / "scripts").glob("*.sh")]),
    ids=lambda p: str(p.relative_to(ROOT)),
)
def test_no_shell_function_is_named_after_a_coreutil_it_could_be_asked_to_run(script: Path) -> None:
    """`install()` + `install -d` is infinite recursion, and `bash -n` cannot see it."""
    shadowed = SHADOW_FORBIDDEN.intersection(_function_names(script))
    assert not shadowed, (
        f"{script.relative_to(ROOT)} defines function(s) {sorted(shadowed)} that shadow an "
        "external command: bash runs the function, so any call meant for the program recurses"
    )


# --- the install path, run for real against a sandboxed root boundary --------------------

#: The commands of `install()` that would touch the machine running pytest. Each one is a
#: PATH stub that records its arguments and succeeds — the same trick `ScriptedNutClient` is
#: for the UPS wire: fake the privileged boundary, run the real logic.
ROOT_BOUNDARY_STUBS = ("id", "getent", "useradd", "usermod", "chown", "systemctl")

#: The three absolute destinations the installer writes to, rewritten into a temp directory
#: so the test exercises the body instead of the operator's `/usr` and `/etc`.
SANDBOX_REWRITES = {
    'AGENT_DIR="/usr/local/lib/dashboards-agent"': 'AGENT_DIR="{root}/lib"',
    'TOKEN_FILE="/etc/dashboards-agent.token"': 'TOKEN_FILE="{root}/token"',
    'CERT_FILE="/etc/dashboards-agent.certs"': 'CERT_FILE="{root}/certs"',
    'UNIT="/etc/systemd/system/$SERVICE.service"': 'UNIT="{root}/dashboards-agent.service"',
}


def _sandbox_installer(tmp_path: Path) -> tuple[Path, Path, dict[str, str]]:
    """A copy of the installer whose writes land in `tmp_path`, plus a stubbed PATH.

    Returns (script, stub call log, env). `agent/host_agent.py` is stubbed next to the copy
    because the script resolves the agent as `../agent/host_agent.py`.
    """
    text = INSTALLER.read_text()
    target = tmp_path / "target"
    for old, new in SANDBOX_REWRITES.items():
        assert old in text, f"the installer moved {old!r} — update SANDBOX_REWRITES with it"
        text = text.replace(old, new.format(root=target))

    script = tmp_path / "deploy" / "install-agent.sh"
    script.parent.mkdir(parents=True)
    script.write_text(text)
    (tmp_path / "agent").mkdir()
    (tmp_path / "agent" / "host_agent.py").write_text("# stub agent source\n", encoding="utf-8")

    stubs = tmp_path / "stubs"
    stubs.mkdir()
    call_log = tmp_path / "stub-calls.log"
    for name in ROOT_BOUNDARY_STUBS:
        stub = stubs / name
        # `id -u` must answer 0 or need_root() refuses before the body runs; every other call
        # just records itself, which is what the assertions below read.
        body = ['#!/bin/sh', f'printf "%s %s\\n" "{name}" "$*" >> "$STUB_LOG"']
        if name == "id":
            body.append('[ "$1" = "-u" ] && echo 0')
        body.append("exit 0")
        stub.write_text("\n".join(body) + "\n", encoding="utf-8")
        stub.chmod(0o755)

    env = {**os.environ, "PATH": f"{stubs}{os.pathsep}{os.environ['PATH']}", "STUB_LOG": str(call_log)}
    return script, call_log, env


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_install_path_installs_into_its_own_paths_and_prints_the_token_once(tmp_path: Path) -> None:
    """The install body end to end: files, unit, token, and no recursion.

    The exit code is the regression test for the `install()` shadow: while the function was
    named after `/usr/bin/install`, this never returned at all.
    """
    script, call_log, env = _sandbox_installer(tmp_path)
    try:
        result = subprocess.run(
            [BASH, str(script), "--bind", "10.0.0.5", "--port", "9120"],
            capture_output=True,
            text=True,
            env=env,
            timeout=60,
            check=False,
        )
    except subprocess.TimeoutExpired as exc:
        pytest.fail(f"the installer did not return (recursion? partial output: {exc.stdout!r})")
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"

    target = tmp_path / "target"
    assert (target / "lib" / "host_agent.py").is_file(), "install -m never landed the agent"
    unit = (target / "dashboards-agent.service").read_text()
    token_path = target / "token"
    token = token_path.read_text().strip()

    assert "User=dashboards-agent" in unit
    assert f"--token-file {token_path}" in unit, "the unit must use the file, never --token"
    assert "--host 10.0.0.5 --port 9120" in unit
    # A token-only install runs no TLS, so it gets neither the bundle flag nor the directory
    # the bundle unpacks into (the agent falls back to a private mkdtemp without one).
    assert "--cert-bundle" not in unit, "no bundle installed means no TLS flag on a token install"
    assert "RuntimeDirectory=dashboards-agent" not in unit
    # No device filter, on purpose (phase 17): a strict device policy denies every node the
    # allow-list fails to match, and systemd matches those globs by *name* against
    # /proc/devices — where kernel 7.2 lists NVMe as `blkext`, so the old `/dev/nvme*` entry
    # locked the whole device namespace on driveripper. Device access is DAC (the `disk` group
    # the installer grants) plus SELinux, the standard SMART-monitoring arrangement.
    assert "DevicePolicy" not in unit, (
        "a strict device policy with a name-matched allow-list is the deny-all device filter "
        "that locked driveripper's SMART reads — the unit must carry none"
    )
    assert "DeviceAllow" not in unit, "no device filter means no allow-list either"
    # PrivateDevices=false survives as the explicit statement of the systemd default: a
    # private /dev with no device list dies at spawn (status 208/STDIN), because systemd
    # opens /dev/null for the service stdin before it execs python.
    assert "PrivateDevices=false" in unit, (
        "the explicit default is what keeps a private-/dev sandbox from dying at STDIN"
    )
    assert token and len(token) >= 32, "a generated token is expected"
    assert token in result.stdout, "first install prints the token, and never again"
    assert f'HOST_TARGETS="hostname=http://{token}@10.0.0.5:9120"' in result.stdout

    calls = call_log.read_text()
    assert "systemctl daemon-reload" in calls
    assert "systemctl enable --now dashboards-agent" in calls
    assert "chown root:dashboards-agent" in calls, "the token file must become group-readable"


#: A line the unit heredoc may legitimately write: a comment, a `[Section]` header, or a
#: systemd directive assignment. Anything else is the unquoted heredoc having *expanded*
#: something — see `test_the_written_unit_is_only_comments_and_directives`.
_UNIT_LINE_RE = re.compile(r"^(?:#|\[[A-Za-z]+\]|[A-Za-z]+=)")


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_the_written_unit_is_only_comments_and_directives(tmp_path: Path) -> None:
    """The unit heredoc is unquoted so `$TLS_UNIT_LINES` and `$TLS_EXEC_ARGS` can expand.

    That is also what executes its **comments**: writing the phase-17 rationale with backticks
    around `disk`, `blkext` and `/dev/nvme*` made the install run them (`disk: command not
    found`, smartctl's version banner) and paste the output into the unit file, with `bash -n`
    and every text assertion still happy. Quote the heredoc and phase 15's interpolated lines
    stop working, so the guard is on the *result*: a unit line is a comment, a section header,
    or a directive assignment — never the output of something the prose mentioned.
    """
    script, _call_log, env = _sandbox_installer(tmp_path)
    result = subprocess.run(
        [BASH, str(script), "--bind", "127.0.0.1", "--port", "9199"],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
        check=False,
    )
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"
    unit = (tmp_path / "target" / "dashboards-agent.service").read_text()
    offenders = [line for line in unit.splitlines() if line.strip() and not _UNIT_LINE_RE.match(line)]
    assert not offenders, "the unit heredoc expanded its own text into the unit file:\n" + "\n".join(
        offenders
    )
    assert result.stderr == "", "installing must not execute anything the unit's comments mention"


# --- the template's unit: no device filter, and the whole unit pinned (phase 17 task 04) ------

#: The ten lines the pre-phase template wrote between ``PrivateDevices=false`` and
#: ``ProtectSystem=strict`` — the device filter phase 17 removes. Pinned as the *old* text, in
#: order, because the fix is the absence of exactly these lines: re-adding
#: ``DeviceAllow=/dev/nvme* rw`` with ``DevicePolicy=strict`` is the deny-all returning, and a
#: test that only asserted "`DevicePolicy` is absent today" would pass while somebody added
#: the allow-list back under a default `auto` policy that ignores it.
DEVICE_FILTER_LINES = (
    "DevicePolicy=strict",
    "DeviceAllow=/dev/null rw",
    "DeviceAllow=/dev/zero rw",
    "DeviceAllow=/dev/full rw",
    "DeviceAllow=/dev/random rw",
    "DeviceAllow=/dev/urandom rw",
    "DeviceAllow=/dev/nvme* rw",
    "DeviceAllow=/dev/sd* rw",
    "DeviceAllow=/dev/vd* rw",
    "DeviceAllow=/dev/sg* rw",
)

#: The unit the template wrote **before** phase 17, rendered for a default token install
#: (bind `10.0.0.5`, port `9120`, no bundle), with the sandbox root spelled `<ROOT>`.
#:
#: This is the whole unit and not a highlight reel: it is the reference the default install is
#: compared against, so a template line added, dropped, reordered or reworded *outside* this
#: phase's remit fails a test instead of landing in the next host's service file. The comment
#: block is included verbatim (it is what the operator reads in `systemctl cat`) even though
#: phase 17 task 01 rewrote that prose on purpose — which is why the comparison below reads
#: directives, and the new rationale is pinned separately for the two facts it must carry.
PRE_PHASE_DEFAULT_UNIT = """[Unit]
Description=Homelab host agent (read-only system metrics)
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=dashboards-agent
ExecStart=/usr/bin/env python3 <ROOT>/target/lib/host_agent.py --host 10.0.0.5 --port 9120 --token-file <ROOT>/target/token
Restart=on-failure
RestartSec=5
Environment=PYTHONUNBUFFERED=1
SyslogIdentifier=dashboards-agent

# The agent reads /proc, /sys and statvfs() and writes nothing, so everything writable can
# be taken away. smartctl is the one thing that needs a real /dev node, so instead of
# PrivateDevices the unit closes all devices and reopens only the disk nodes, read/write
# at the cgroup layer while the code (and smartctl -i -A -H) stays read-only by design.
#
# The pseudo nodes are not decoration: DevicePolicy=strict denies every node that is not
# listed, and systemd needs /dev/null for the service's stdin before it ever execs python —
# without it the unit dies at spawn with "Failed at step STDIN" (status 208), which is how
# the first phase-10 install failed on a live host running systemd 257. /dev/zero, full,
# random and urandom are the rest of the set a PrivateDevices= sandbox would have provided,
# and none of them is a disk.
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=false
DevicePolicy=strict
DeviceAllow=/dev/null rw
DeviceAllow=/dev/zero rw
DeviceAllow=/dev/full rw
DeviceAllow=/dev/random rw
DeviceAllow=/dev/urandom rw
DeviceAllow=/dev/nvme* rw
DeviceAllow=/dev/sd* rw
DeviceAllow=/dev/vd* rw
DeviceAllow=/dev/sg* rw
ProtectSystem=strict
ProtectHome=read-only
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
RestrictSUIDSGID=true
RestrictNamespaces=true
RestrictRealtime=true
LockPersonality=true
MemoryDenyWriteExecute=true
CapabilityBoundingSet=
SystemCallFilter=@system-service

[Install]
WantedBy=multi-user.target
"""


def _unit_heredoc() -> str:
    """The template's unit heredoc, exactly as the script holds it.

    Scoped on purpose: the rest of the script is prose about flags, and a pin that read the
    whole file for `DeviceAllow` would be satisfied by deleting a sentence rather than by the
    absence of a *unit line*.
    """
    text = INSTALLER.read_text()
    assert "<<UNIT_FILE" in text, "the unit is still written from a heredoc"
    heredoc = text.split("<<UNIT_FILE\n", 1)[1].split("\nUNIT_FILE\n", 1)[0]
    assert "[Service]" in heredoc and "CapabilityBoundingSet" in heredoc, (
        "the heredoc region extracted from the script is not the unit"
    )
    return heredoc


def _unit_directives(unit: str) -> list[str]:
    """The lines systemd acts on — every line that is neither a comment nor blank.

    Comments carry the phase's reasoning and *are* part of the file, but they are also the one
    thing phase 17 task 01 rewrote deliberately (the old block justified the filter; the new
    one explains its absence). Directives are what a host runs, so those are compared exactly.
    """
    return [line for line in unit.splitlines() if line.strip() and not line.startswith("#")]


def test_the_template_unit_has_no_device_filter_lines() -> None:
    """The fix is the absence, so the absence is what is pinned.

    Three layers, because the regression has three shapes: a `DevicePolicy=strict` or
    `DeviceAllow=` line back in the heredoc (the deny-all returning on kernel 7.2, where
    systemd matches those globs by *name* against `/proc/devices` and lists NVMe as `blkext`,
    so `/dev/nvme*` matched nothing and strict denied the whole namespace);
    `PrivateDevices=false` removed as "redundant" because it is the systemd default (a private
    `/dev` with no device list dies at spawn with `status=208/STDIN`); and a rationale comment
    that stops naming the mechanism, which is how the next operator re-adds the filter while
    believing they are hardening the unit.
    """
    text = INSTALLER.read_text()
    # The phase's completion criterion is this exact grep, so it is pinned as one: neither
    # directive may appear anywhere in the script, prose included.
    assert "DevicePolicy" not in text, "no device policy line anywhere in the installer"
    assert "DeviceAllow" not in text, "no device allow-list anywhere in the installer"

    heredoc = _unit_heredoc()
    offenders = [line for line in heredoc.splitlines() if re.match(r"^\s*(DevicePolicy|DeviceAllow)\s*=", line)]
    assert offenders == [], "the unit heredoc grew a device filter line again:\n" + "\n".join(offenders)
    assert "PrivateDevices=false" in heredoc, (
        "the explicit default is the line that keeps the unit from dying at STDIN; the filter "
        "going away does not make it redundant"
    )
    # The rewritten rationale has to be an explanation, not a shrug (task 01's own criterion).
    comment = "\n".join(line for line in heredoc.splitlines() if line.startswith("#"))
    assert "/proc/devices" in comment and "blkext" in comment, (
        "the comment must name the mechanism, or the next reader cannot tell an omitted filter "
        "from a forgotten one"
    )
    assert "208" in comment, "and the STDIN failure that PrivateDevices=false still prevents"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_a_default_install_unit_is_the_pre_phase_unit_minus_the_filter_lines(tmp_path: Path) -> None:
    """No flags writes the unit this script has always written, minus exactly ten lines.

    `AGENTS.md` invariant 3 in deployment clothes: a host that needs nothing new must not get a
    new unit from an unrelated phase. The reference is the whole pre-phase unit recorded above,
    so every directive is compared (not a keyword spot-check) and the only allowed difference is
    the device filter's absence — asserted as the ordered list comparison *and* as each of the ten
    lines being absent by name, so a filter line silently renamed rather than removed fails too.
    """
    script, _call_log, env = _sandbox_installer(tmp_path)
    result = _install(script, env, "--bind", "10.0.0.5", "--port", "9120")
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"
    installed = (tmp_path / "target" / "dashboards-agent.service").read_text()

    reference = PRE_PHASE_DEFAULT_UNIT.replace("<ROOT>", str(tmp_path))
    expected = [line for line in _unit_directives(reference) if line not in DEVICE_FILTER_LINES]
    assert len(expected) == len(_unit_directives(reference)) - len(DEVICE_FILTER_LINES), (
        "DEVICE_FILTER_LINES must match ten real lines of the reference unit — a filter line "
        "this list no longer names is a filter that would pass this test"
    )
    got = _unit_directives(installed)
    drift = "\n".join(
        difflib.unified_diff(
            expected, got, fromfile="pre-phase unit minus the filter", tofile="installed", lineterm=""
        )
    )
    assert got == expected, f"the default unit drifted from the pre-phase unit by more than the filter:\n{drift}"

    # And the ten lines are gone rather than reworded into something equivalent.
    assert not [line for line in installed.splitlines() if line.startswith(DEVICE_FILTER_LINES)]


# --- the kernel-gate unit: --as-root / --caps (phase 17 task 02) ---------------------------

#: The exact sentence `--as-root` alone prints. It is not a refusal — a capability-less root is
#: a legitimate if unusual choice — but it is the fact gap 10 paid for: the first root-user
#: attempt on the kernel-7.2 host still read `n/a` on all 16 drives, because the bounding set
#: bounds even root (`CapEff 0`). Pinned verbatim so a paraphrase that drops "not uid" is a
#: test failure rather than an operator rediscovering it.
AS_ROOT_NOTE = (
    "note: --as-root with the default empty CapabilityBoundingSet leaves the process "
    "capability-less (the kernel gates check capabilities, not uid); pair it with --caps "
    "for a kernel-gated host"
)

#: `--caps` on the dedicated service user is not the kernel-gate fix (a non-root process cannot
#: hold these capabilities across exec), so the combination is refused rather than written.
CAPS_REFUSAL = "--caps needs --as-root: a capability grant on the service user is not the kernel-gate fix"

#: The two capabilities the kernel names for the two gates (NVMe SMART log / SG_IO), and the
#: `driveripper` recipe as `--help` must teach it (phase 17 completion criterion).
KERNEL_CAPS = "CAP_SYS_RAWIO,CAP_SYS_ADMIN"
KERNEL_GATE_RECIPE = f"--as-root --caps {KERNEL_CAPS}"


def _install(script: Path, env: dict[str, str], *flags: str) -> subprocess.CompletedProcess[str]:
    """One sandboxed installer run: fake root boundary, real script body."""
    try:
        return subprocess.run(
            [BASH, str(script), *flags],
            capture_output=True,
            text=True,
            env=env,
            timeout=60,
            check=False,
        )
    except subprocess.TimeoutExpired as exc:  # pragma: no cover - the recursion regression
        pytest.fail(f"the installer did not return (flags {flags}, partial output: {exc.stdout!r})")


def _directive(unit: str, key: str) -> list[str]:
    """Every line of the unit assigning `key` — a duplicate assignment is a heredoc bug."""
    return [line for line in unit.splitlines() if line.split("=", 1)[0] == key]


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_the_default_unit_keeps_the_service_user_and_an_empty_bounding_set(tmp_path: Path) -> None:
    """No flags = the unit this script wrote before the flags existed, byte-for-byte.

    `CapabilityBoundingSet=` with nothing after the `=` is the capability-less default and what
    every host whose drives answer through the `disk` group runs; it is also the line
    `--caps` replaces, so its exact spelling is pinned here rather than as `in unit`.
    """
    script, _call_log, env = _sandbox_installer(tmp_path)
    result = _install(script, env, "--bind", "127.0.0.1", "--port", "9199")
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"
    unit = (tmp_path / "target" / "dashboards-agent.service").read_text()
    assert _directive(unit, "User") == ["User=dashboards-agent"]
    assert _directive(unit, "CapabilityBoundingSet") == ["CapabilityBoundingSet="]
    assert AS_ROOT_NOTE not in result.stdout, "the note belongs to --as-root alone"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_as_root_with_caps_writes_the_kernel_gate_unit(tmp_path: Path) -> None:
    """The two lines that reproduce the working `driveripper` unit, exactly.

    Comma on the command line, space in the unit: that is systemd's own spelling for
    `CapabilityBoundingSet`, and it is the whole reason the flag takes a list rather than the
    line. Everything else in the unit — the token file, its `0640 root:dashboards-agent`, the
    hardening block — is unchanged by these flags.
    """
    script, call_log, env = _sandbox_installer(tmp_path)
    kernel_gate = ["--bind", "10.0.0.5", "--port", "9120", "--as-root", "--caps", KERNEL_CAPS]
    result = _install(script, env, *kernel_gate)
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"
    unit = (tmp_path / "target" / "dashboards-agent.service").read_text()
    assert _directive(unit, "User") == ["User=root"]
    assert _directive(unit, "CapabilityBoundingSet") == ["CapabilityBoundingSet=CAP_SYS_RAWIO CAP_SYS_ADMIN"]
    assert AS_ROOT_NOTE not in result.stdout, "--caps answers the note: the grant is there"
    # The token file's logic is deliberately untouched by --as-root: root reads it either way,
    # and re-deriving the mode per service user would be a second rule for no new safety.
    token = tmp_path / "target" / "token"
    assert stat.S_IMODE(token.stat().st_mode) == 0o640
    assert "chown root:dashboards-agent" in call_log.read_text()
    assert "--host 10.0.0.5 --port 9120" in unit


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_as_root_and_caps_write_the_kernel_gate_unit_lines(tmp_path: Path) -> None:
    """The whole flag contract in one place, plus the part only a comparison can pin.

    The three flag shapes (`--as-root --caps LIST`, `--caps` alone, `--as-root` alone) each have
    their own detailed pin nearby; what none of them shows is **how much** the recipe moves. So:
    two installs in two sandboxes at the same bind and port, one default and one kernel-gated, and
    their directive lists must differ by exactly the two lines this phase promises and nothing
    else — the hardening block, the token file's path, the `ExecStart` and the `[Install]` section
    are not these flags' business, and an `--as-root` that also slipped a `ProtectSystem` change in
    would be a privilege change nobody asked for. The two half-recipes are then run in their own
    sandboxes (one host re-run with different flags is the guard's decision, not the parser's).
    """
    plain_script, _plain_log, plain_env = _sandbox_installer(tmp_path / "plain")
    plain = _install(plain_script, plain_env, "--bind", "10.1.0.126", "--port", "9120")
    assert plain.returncode == 0, f"default install failed:\n{plain.stdout}\n{plain.stderr}"

    gate_script, gate_log, gate_env = _sandbox_installer(tmp_path / "gate")
    recipe = _install(
        gate_script, gate_env, "--bind", "10.1.0.126", "--port", "9120", "--as-root", "--caps", KERNEL_CAPS
    )
    assert recipe.returncode == 0, f"the driveripper recipe failed:\n{recipe.stdout}\n{recipe.stderr}"

    def directives(where: str) -> list[str]:
        root = tmp_path / where
        unit = (root / "target" / "dashboards-agent.service").read_text()
        return _unit_directives(unit.replace(str(root), "<ROOT>"))

    changed = [
        f"{a} -> {b}"
        for a, b in zip(directives("plain"), directives("gate"), strict=True)
        if a != b
    ]
    assert changed == [
        "User=dashboards-agent -> User=root",
        "CapabilityBoundingSet= -> CapabilityBoundingSet=CAP_SYS_RAWIO CAP_SYS_ADMIN",
    ], "--as-root --caps may write these two lines and nothing else"
    assert "chown root:dashboards-agent" in gate_log.read_text(), (
        "--as-root leaves the token file's ownership alone: root reads it either way, and the "
        "grant must not widen because of a flag that is about capabilities"
    )

    # The two half-recipes, in their own sandboxes (a differing re-run of one host is the
    # guard's business, not the flag parser's).
    caps_only_script, caps_only_log, caps_only_env = _sandbox_installer(tmp_path / "caps-only")
    caps_only = _install(
        caps_only_script, caps_only_env, "--bind", "127.0.0.1", "--port", "9199", "--caps", KERNEL_CAPS
    )
    assert caps_only.returncode == 1, "--caps without --as-root is refused"
    assert CAPS_REFUSAL in caps_only.stderr
    assert not caps_only_log.exists(), "and refused before a single privileged command"

    root_only_script, _root_only_log, root_only_env = _sandbox_installer(tmp_path / "root-only")
    root_only = _install(root_only_script, root_only_env, "--bind", "127.0.0.1", "--port", "9199", "--as-root")
    assert root_only.returncode == 0, f"--as-root alone must proceed:\n{root_only.stderr}"
    assert AS_ROOT_NOTE in root_only.stdout, "with the capability-less note"
    root_unit = (tmp_path / "root-only" / "target" / "dashboards-agent.service").read_text()
    assert _directive(root_unit, "User") == ["User=root"]
    assert _directive(root_unit, "CapabilityBoundingSet") == ["CapabilityBoundingSet="]


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_as_root_alone_installs_and_prints_the_capability_note(tmp_path: Path) -> None:
    """A capability-less root proceeds — with the reminder, on stdout, and no refusal."""
    script, _call_log, env = _sandbox_installer(tmp_path)
    result = _install(script, env, "--bind", "127.0.0.1", "--port", "9199", "--as-root")
    assert result.returncode == 0, f"--as-root alone must proceed:\n{result.stdout}\n{result.stderr}"
    assert AS_ROOT_NOTE in result.stdout, "the note is the operator's only warning here"
    unit = (tmp_path / "target" / "dashboards-agent.service").read_text()
    assert _directive(unit, "User") == ["User=root"]
    assert _directive(unit, "CapabilityBoundingSet") == ["CapabilityBoundingSet="], (
        "--as-root alone writes no capability grant: the note is the whole effect"
    )


@pytest.mark.parametrize(
    ("caps", "refusal"),
    [
        ("CAP_SYS_RAWIO", CAPS_REFUSAL),
        ("CAP_SYS_RAWIO,CAP_SYS_ADMIN", CAPS_REFUSAL),
        ("", "--caps needs at least one capability name"),
        (" , ", "--caps needs at least one capability name"),
    ],
    ids=["one-cap", "two-caps", "empty", "punctuation-only"],
)
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_caps_without_as_root_is_refused_before_anything_is_written(
    tmp_path: Path, caps: str, refusal: str
) -> None:
    """Exit 1, the reason on stderr, and no side effect at all.

    The check sits *before* `need_root` (the `--cert-bundle` validator's reasoning): a wrong
    flag combination gets its own sentence instead of a sudoers hint, and nothing after it —
    user, group membership, token, unit — can have run.
    """
    script, call_log, env = _sandbox_installer(tmp_path)
    result = _install(script, env, "--bind", "127.0.0.1", "--port", "9199", "--caps", caps)
    assert result.returncode == 1, "the refusal keeps exit 1"
    assert result.stdout == "", "a refusal is a failure: nothing on stdout"
    assert refusal in result.stderr
    assert GRANT_PREFIX not in result.stderr, "a flag mistake is not a sudoers problem"
    target = tmp_path / "target"
    assert not (target / "dashboards-agent.service").exists(), "refused means nothing written"
    assert not (target / "token").exists()
    assert not call_log.exists(), "no privileged command ran either"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_caps_without_an_argument_is_refused_rather_than_crashing(tmp_path: Path) -> None:
    """`--caps` last on the command line: exit 1 naming the flag, not an `unbound variable`."""
    script, _call_log, env = _sandbox_installer(tmp_path)
    result = _install(script, env, "--bind", "127.0.0.1", "--port", "9199", "--as-root", "--caps")
    assert result.returncode == 1
    assert result.stdout == ""
    assert "--caps needs a capability list" in result.stderr


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_the_caps_list_is_normalised_to_systemd_spelling(tmp_path: Path) -> None:
    """Empty entries and stray whitespace never reach the unit as a double space…

    …and a `*` stays a literal: the list is split with pathname expansion switched off, so an
    operator typo cannot substitute the working directory's file names into `User=`'s partner.
    """
    script, _call_log, env = _sandbox_installer(tmp_path)
    messy = ["--bind", "127.0.0.1", "--port", "9199", "--as-root", "--caps", "CAP_SYS_RAWIO,, CAP_SYS_ADMIN "]
    result = _install(script, env, *messy)
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"
    unit = (tmp_path / "target" / "dashboards-agent.service").read_text()
    assert _directive(unit, "CapabilityBoundingSet") == ["CapabilityBoundingSet=CAP_SYS_RAWIO CAP_SYS_ADMIN"]

    script_glob, _log, env_glob = _sandbox_installer(tmp_path / "glob")
    star = ["--bind", "127.0.0.1", "--port", "9199", "--as-root", "--caps", "*"]
    result_glob = _install(script_glob, env_glob, *star)
    assert result_glob.returncode == 0, f"install failed:\n{result_glob.stdout}\n{result_glob.stderr}"
    unit_glob = (tmp_path / "glob" / "target" / "dashboards-agent.service").read_text()
    assert _directive(unit_glob, "CapabilityBoundingSet") == ["CapabilityBoundingSet=*"], (
        "the capability list is words, not a glob"
    )


def test_help_teaches_the_kernel_gate_recipe() -> None:
    """Completion criterion: the `driveripper` flags are learnable from `--help` alone.

    `--help` *is* the header comment, so the sentence that explains why a root service needs
    its capabilities named is pinned here too — it is the one thing an executor on a fresh
    clone cannot infer from a unit file.
    """
    result = _run([BASH, str(INSTALLER), "--help"])
    assert result.returncode == 0
    assert KERNEL_GATE_RECIPE in result.stdout, "the recipe line must be copy-pasteable"
    assert "--as-root      run the service as root" in result.stdout
    assert "--caps LIST    CapabilityBoundingSet line for the unit" in result.stdout
    assert "capabilities, not uid" in result.stdout, "the kernel gate is the why of both flags"
    assert "capability-less" in result.stdout, "--as-root without --caps has to say what it is"


def test_the_status_probe_is_blind_to_who_the_service_runs_as() -> None:
    """`--status` probes the agent over the wire; the unit's user is invisible to it, on purpose.

    A status path that read `AS_ROOT`/`CAPS` would make a health probe depend on flags the
    operator may not have passed on this invocation — the probe answers about the port.
    """
    status_body = INSTALLER.read_text().split("status() {", 1)[1].split("\n}", 1)[0]
    assert "AS_ROOT" not in status_body
    assert "CAPS" not in status_body


def test_token_file_is_readable_by_the_service_user_and_nobody_else(tmp_path: Path) -> None:
    """`--token-file` is a path the *agent* opens, as its own user.

    systemd does not read the secret for the service, so a 0600 root-owned file makes the
    agent exit before it binds. 0640 root:dashboards-agent is the smallest mode that works,
    and it is still unreadable by `ops` (which is why deploy/agent-sudoers does not grant it).
    """
    script, _call_log, env = _sandbox_installer(tmp_path)
    result = subprocess.run(
        [BASH, str(script), "--bind", "10.0.0.5"],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
        check=False,
    )
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"
    mode = stat.S_IMODE((tmp_path / "target" / "token").stat().st_mode)
    assert mode == 0o640, f"token file mode is {oct(mode)}, expected 0o640 (group = the agent user)"
    assert not mode & 0o007, "the token file must never be world-readable"


# --- the re-run guard: no silent clobber (phase 17 task 03) ----------------------------------

def _refusal(unit: Path) -> str:
    """The sentence a refused re-run prints, with the unit path substituted.

    Pinned verbatim because it is the operator's only explanation of why a routine re-run
    stopped, and because the diff follows it on **stdout** rather than stderr: the diff is the
    payload of the answer (an operator pipes it into a pager or a file), and splitting a
    sentence from its diff across two streams only interleaves them.
    """
    return f"refusing to overwrite {unit} (generated unit differs); run with --force to apply:"


#: One line the operator can plausibly have edited on the host — `RestartSec` is a real knob,
#: it is not a line any flag writes, and it sits far from every other assertion in these tests.
OPERATOR_EDIT = ("RestartSec=5", "RestartSec=7  # hand-tuned on the host after a flapping start")


def _hand_edit(unit_path: Path) -> str:
    """Edit the installed unit the way an operator does, and return what it now holds."""
    text = unit_path.read_text()
    assert OPERATOR_EDIT[0] in text, "the template unit lost the line the operator edits"
    edited = text.replace(*OPERATOR_EDIT)
    unit_path.write_text(edited, encoding="utf-8")
    return edited


def _systemctl_calls(call_log: Path) -> list[str]:
    """Only the `systemctl` lines of the stub log — the ones that touch the running service.

    The privileged calls before the guard (`id`, `getent`, `usermod`, `chown`) do run again on
    a refused re-run: they are the idempotent ones the guard's reasoning accepts. `systemctl`
    is what a refused run must not do — no `daemon-reload`, no `enable --now` — so that is the
    half compared here.
    """
    return [line for line in call_log.read_text().splitlines() if line.startswith("systemctl ")]


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_a_rerun_that_would_change_the_installed_unit_refuses(tmp_path: Path) -> None:
    """install → hand-edit → re-run: exit 1, the diff on stdout, nothing applied.

    This is the incident the guard exists for: `driveripper`'s unit was fixed by hand, and the
    installer rewrote it unconditionally. A refused run must leave the unit on disk as the
    operator left it and must not reload or re-enable anything, or the refusal arrives after
    the thing it is refusing.
    """
    script, call_log, env = _sandbox_installer(tmp_path)
    unit_path = tmp_path / "target" / "dashboards-agent.service"
    token_path = tmp_path / "target" / "token"

    first = _install(script, env, "--bind", "10.0.0.5", "--port", "9120")
    assert first.returncode == 0, f"first install failed:\n{first.stdout}\n{first.stderr}"
    edited = _hand_edit(unit_path)
    token_before = token_path.read_bytes()
    reloads_before = _systemctl_calls(call_log)
    assert reloads_before, "the first install did reload and enable the service"

    result = _install(script, env, "--bind", "10.0.0.5", "--port", "9120")
    assert result.returncode == 1, "a differing re-run refuses with exit 1"
    assert result.stderr == "", "the refusal and its diff are one answer: stdout, not split"
    assert _refusal(unit_path) in result.stdout
    assert "--- " in result.stdout and "+++ " in result.stdout, "the diff is unified"
    assert f"-{OPERATOR_EDIT[1]}" in result.stdout, "the diff shows the operator's line going away"
    assert "+RestartSec=5" in result.stdout, "and the template's line coming back"
    assert unit_path.read_text() == edited, "a refused run leaves the installed unit byte-identical"
    # The reuse path re-reads the existing secret and rewrites the same bytes with the same
    # mode; its *value* — what invariant 6 and the HOST_TARGETS entry depend on — never moves,
    # which is the guarantee here (mtime is not: the write happens before the guard).
    assert token_path.read_bytes() == token_before, "the reused token must not change"
    assert stat.S_IMODE(token_path.stat().st_mode) == 0o640
    assert _systemctl_calls(call_log) == reloads_before, (
        "a refused run must not daemon-reload or enable: that would restart the service the "
        "refusal exists to leave alone"
    )


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_force_prints_the_diff_then_writes_the_generated_unit(tmp_path: Path) -> None:
    """`--force` is the opt-in, and the diff still arrives: an operator approves what they saw."""
    script, call_log, env = _sandbox_installer(tmp_path)
    unit_path = tmp_path / "target" / "dashboards-agent.service"
    _install(script, env, "--bind", "10.0.0.5", "--port", "9120")
    _hand_edit(unit_path)

    result = _install(script, env, "--bind", "10.0.0.5", "--port", "9120", "--force")
    assert result.returncode == 0, f"--force must proceed:\n{result.stdout}\n{result.stderr}"
    assert _refusal(unit_path) not in result.stdout, "--force is not a refusal"
    assert f"-{OPERATOR_EDIT[1]}" in result.stdout and "+RestartSec=5" in result.stdout, (
        "the diff is printed even when the answer is yes"
    )
    unit = unit_path.read_text()
    assert "RestartSec=5" in unit and OPERATOR_EDIT[1] not in unit, "the generated unit is written"
    assert _systemctl_calls(call_log).count("systemctl enable --now dashboards-agent") == 2, (
        "the overwrite is followed by the usual reload/enable"
    )


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_an_unchanged_rerun_is_silent_and_proceeds(tmp_path: Path) -> None:
    """Idempotence is the normal path: same flags, same unit, no diff, no complaint, exit 0."""
    script, call_log, env = _sandbox_installer(tmp_path)
    unit_path = tmp_path / "target" / "dashboards-agent.service"
    first = _install(script, env, "--bind", "10.0.0.5", "--port", "9120")
    assert first.returncode == 0, f"first install failed:\n{first.stdout}\n{first.stderr}"
    installed = unit_path.read_text()

    result = _install(script, env, "--bind", "10.0.0.5", "--port", "9120")
    assert result.returncode == 0, f"an identical re-run must proceed:\n{result.stdout}\n{result.stderr}"
    assert _refusal(unit_path) not in result.stdout
    assert "+++ " not in result.stdout, "nothing differs, so nothing is printed about differences"
    assert "reusing the existing token" in result.stdout, "the token idiom is unchanged"
    assert unit_path.read_text() == installed
    assert _systemctl_calls(call_log).count("systemctl enable --now dashboards-agent") == 2


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_a_re_run_that_differs_refuses_without_force_and_applies_with_it(tmp_path: Path) -> None:
    """The refuse → force → silent sequence on **one** host, in that order.

    Each leg has its own detailed pin beside it; what those cannot show is the transition — a
    guard that refuses correctly the first time and then never lets an identical re-run through
    silently is the same bug seen from the other side (an operator who has to type `--force`
    every deploy stops reading the diff). So: install, hand-edit, re-run (refused, unit left
    exactly as edited, no new `systemctl` call), re-run with `--force` (diff first, generated
    unit on disk, service re-enabled), then re-run plain once more (silent, unit unmoved).
    """
    script, call_log, env = _sandbox_installer(tmp_path)
    unit_path = tmp_path / "target" / "dashboards-agent.service"
    flags = ["--bind", "10.1.0.126", "--port", "9120", "--as-root", "--caps", KERNEL_CAPS]

    first = _install(script, env, *flags)
    assert first.returncode == 0, f"first install failed:\n{first.stdout}\n{first.stderr}"
    edited = _hand_edit(unit_path)
    reloads = len(_systemctl_calls(call_log))

    refused = _install(script, env, *flags)
    assert refused.returncode == 1, "leg 1: a differing re-run refuses"
    assert _refusal(unit_path) in refused.stdout and "+++ " in refused.stdout
    assert unit_path.read_text() == edited, "leg 1: the hand edit survives"
    assert len(_systemctl_calls(call_log)) == reloads, "leg 1: nothing was reloaded or re-enabled"

    applied = _install(script, env, *flags, "--force")
    assert applied.returncode == 0, f"leg 2: --force failed:\n{applied.stdout}\n{applied.stderr}"
    assert "+++ " in applied.stdout, "leg 2: the diff is printed even on a yes"
    assert unit_path.read_text() != edited, "leg 2: the generated unit replaced the hand edit"
    assert "RestartSec=5" in unit_path.read_text()
    generated = unit_path.read_text()
    # three calls per applied install: daemon-reload, enable --now, and the is-active check
    assert len(_systemctl_calls(call_log)) == reloads + 3, "leg 2: reload + enable, as a normal install"

    settled = _install(script, env, *flags)
    assert settled.returncode == 0, f"leg 3: a silent re-run failed:\n{settled.stdout}\n{settled.stderr}"
    assert "refusing to overwrite" not in settled.stdout and "+++ " not in settled.stdout
    assert "reusing the existing token" in settled.stdout
    assert unit_path.read_text() == generated, "leg 3: the silent path leaves the applied unit in place"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_a_fresh_install_never_meets_the_guard(tmp_path: Path) -> None:
    """No installed unit → no comparison, no output about one, and `--force` changes nothing.

    The guard is about a *re*-run. A first install must stay exactly the run it was before this
    task, with or without `--force` — hence the two sandboxes writing byte-identical units.
    """
    plain_script, _plain_log, plain_env = _sandbox_installer(tmp_path / "plain")
    forced_script, _forced_log, forced_env = _sandbox_installer(tmp_path / "forced")

    plain = _install(plain_script, plain_env, "--bind", "10.0.0.5", "--port", "9120")
    forced = _install(forced_script, forced_env, "--bind", "10.0.0.5", "--port", "9120", "--force")
    assert plain.returncode == 0, f"a first install failed:\n{plain.stdout}\n{plain.stderr}"
    assert forced.returncode == 0, f"--force broke a first install:\n{forced.stdout}\n{forced.stderr}"
    for stdout in (plain.stdout, forced.stdout):
        assert "refusing to overwrite" not in stdout
        assert "+++ " not in stdout

    def rendered(where: str) -> str:
        """The unit that sandbox wrote, with its own sandbox root name blanked out.

        The two runs live in different directories, so the absolute paths in `ExecStart` and
        `--token-file` differ while every *decision* in the unit is the same — and the decision
        is what `--force` must not touch.
        """
        root = tmp_path / where
        return (
            (root / "target" / "dashboards-agent.service")
            .read_text()
            .replace(str(root), "<sandbox>")
        )

    assert rendered("plain") == rendered("forced"), "--force writes the unit the default install writes"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_the_kernel_gate_reproduction_refuses_once_and_then_holds(tmp_path: Path) -> None:
    """The `driveripper` case end to end: a token-mode unit is *not* the kernel-gate unit.

    Switching a host to `--as-root --caps` is a real change to a real unit, so it asks first,
    applies with `--force`, and is silent afterwards — the flag line in `--help` is a
    reproduction command only if running it twice lands the same unit and says nothing twice.
    """
    script, _call_log, env = _sandbox_installer(tmp_path)
    unit_path = tmp_path / "target" / "dashboards-agent.service"
    kernel_gate = ["--bind", "10.1.0.126", "--port", "9120", "--as-root", "--caps", KERNEL_CAPS]
    _install(script, env, "--bind", "10.1.0.126", "--port", "9120")
    token_mode = unit_path.read_text()

    asked = _install(script, env, *kernel_gate)
    assert asked.returncode == 1, "a flag change to an installed unit is a change, so it asks"
    assert _refusal(unit_path) in asked.stdout
    assert "-User=dashboards-agent" in asked.stdout and "+User=root" in asked.stdout
    assert "+CapabilityBoundingSet=CAP_SYS_RAWIO CAP_SYS_ADMIN" in asked.stdout
    assert unit_path.read_text() == token_mode, "the question must not also be the answer"

    applied = _install(script, env, *kernel_gate, "--force")
    assert applied.returncode == 0, f"--force failed:\n{applied.stdout}\n{applied.stderr}"
    applied_unit = unit_path.read_text()
    assert _directive(applied_unit, "User") == ["User=root"]
    assert _directive(applied_unit, "CapabilityBoundingSet") == [
        "CapabilityBoundingSet=CAP_SYS_RAWIO CAP_SYS_ADMIN"
    ]

    again = _install(script, env, *kernel_gate, "--force")
    assert again.returncode == 0, f"the same command twice failed:\n{again.stdout}\n{again.stderr}"
    assert "+++ " not in again.stdout, "once applied, the same flags are the zero-noise path"
    assert unit_path.read_text() == applied_unit


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_uninstall_is_still_deliberate_and_never_asks(tmp_path: Path) -> None:
    """The guard belongs to `do_install`: `--uninstall` removes a hand-edited unit without one.

    A guard on the teardown path would be a guard that keeps a unit nobody wants, and
    `uninstall()` does not render a unit at all.
    """
    script, call_log, env = _sandbox_installer(tmp_path)
    unit_path = tmp_path / "target" / "dashboards-agent.service"
    _install(script, env, "--bind", "10.0.0.5", "--port", "9120")
    _hand_edit(unit_path)

    result = _install(script, env, "--uninstall")
    assert result.returncode == 0, f"uninstall refused:\n{result.stdout}\n{result.stderr}"
    assert "refusing to overwrite" not in result.stdout
    assert not unit_path.exists(), "uninstall still removes the unit, edited or not"
    assert "systemctl disable --now dashboards-agent" in _systemctl_calls(call_log)[-3:]


def test_the_guard_sits_between_the_render_and_the_unit_write() -> None:
    """Text pin for the ordering the behaviour depends on.

    The comparison must read the *rendered* unit (every flag, bundle and capability decision
    applied) and must come before the write, the `daemon-reload` and the `enable`; a `cat >
    "$UNIT" <<UNIT_FILE` would put the clobber back while still passing every behavioural test
    above on a host with no installed unit.
    """
    text = INSTALLER.read_text()
    assert "do_install() {" in text, "the install body is still the function that writes the unit"
    # Scoped to do_install: `uninstall()` also reloads the daemon, and an unscoped index would
    # compare offsets across two different functions.
    body = text.split("do_install() {", 1)[1]
    render = body.index('cat > "$TMP_UNIT" <<UNIT_FILE')
    guard = body.index('! cmp -s "$UNIT" "$TMP_UNIT"')
    write = body.index('cat "$TMP_UNIT" > "$UNIT"')
    reload = body.index("systemctl daemon-reload")
    enable = body.index('systemctl enable --now "$SERVICE"')
    assert render < guard < write < reload < enable
    assert 'cat > "$UNIT" <<' not in text, "the template may only write $UNIT after comparing it"
    assert 'if [ "$FORCE" -eq 1 ]; then' in text, "--force is the guard's own opt-in branch"
    assert "rm -f \"$TMP_UNIT\"" in text, "the rendered temp file is cleaned up on every path"
    assert 'mktemp "${TMPDIR:-/tmp}/dashboards-agent-unit.XXXXXX"' in text, (
        "the diff's +++ side must name the generated unit, not an anonymous temp file"
    )


#: The `--force` help line's sentence (phase 17 completion criterion: the flags are learnable
#: from `--help`, and the diff-first promise is the reason an operator trusts `--force`).
FORCE_HELP_SENTENCE = "overwrite an existing unit even if the generated unit differs"


def test_help_documents_force_and_keeps_the_flag_block_aligned() -> None:
    stdout = _run([BASH, str(INSTALLER), "--help"]).stdout
    lines = [line for line in stdout.splitlines() if FORCE_HELP_SENTENCE in line]
    assert len(lines) == 1, f"exactly one --force flag line is expected, got {lines}"
    assert lines[0].lstrip().startswith("--force")
    assert "(prints the" in lines[0] and "diff first)" in stdout
    # One column for every flag's description: the block is a table, and `--force` landing at
    # its own column is how `--help` stays readable as a reference rather than a paragraph.
    as_root = next(line for line in stdout.splitlines() if line.lstrip().startswith("--as-root"))
    assert lines[0].index(FORCE_HELP_SENTENCE) == as_root.index("run the service as root")


# --- the certificate bundle (phase 15 task 06) --------------------------------------------

#: Where the installed bundle lives — one constant in the script, pinned here so a rename
#: fails a test instead of stranding every already-written unit file.
CERT_FILE_PATH = "/etc/dashboards-agent.certs"


def _fake_bundle_text(revision: str = "3") -> str:
    """A syntactically real bundle: agent cert, agent key, root CA, and the rev line.

    Shape only. The installer's validator greps PEM blocks by text exactly like
    `resolve_bundle` does (a monitored host has no `cryptography`), so a well-formed *fake*
    is all an unprivileged validation run needs; the functional probe tests below use a
    real bundle minted by `dashboards.certs`.
    """

    def block(kind: str, body: str) -> str:
        return f"-----BEGIN {kind}-----\n{body}\n-----END {kind}-----\n"

    return (
        f"# BUNDLE-REV {revision}\n"
        + block("CERTIFICATE", "AGENTCERT")
        + block("PRIVATE KEY", "AGENTKEY")
        + block("CERTIFICATE", "ROOTCA")
    )


def _installer_status(stdout: str) -> str:
    """The line `status()` prints as `health: …`, parsed as JSON when it is JSON."""
    for line in stdout.splitlines():
        if line.startswith("health: "):
            body = line.removeprefix("health: ").strip()
            try:
                return str(json.loads(body)["ok"])
            except (json.JSONDecodeError, KeyError):
                return body
    pytest.fail(f"the probe printed no health line:\n{stdout}")


@pytest.mark.parametrize(
    ("shape", "reason"),
    [
        ("missing", "cannot be read"),
        ("empty", "is empty"),
        ("no-revision", "BUNDLE-REV"),
        ("two-keys", "the agent certificate, the agent key and the root CA"),
    ],
)
def test_a_bad_cert_bundle_is_refused_with_its_reason_before_anything_else(
    tmp_path: Path, shape: str, reason: str
) -> None:
    """Completion criterion: `--cert-bundle /nonexistent` refuses with the reason, touches nothing.

    The validation runs *before* `need_root`, so the failure names the file rather than
    sending the operator to add a sudoers line for a typo in the flag's argument — and
    nothing after the validation (user, files, unit) can have run.
    """
    if shape == "missing":
        path = tmp_path / "absent.bundle"
    elif shape == "empty":
        path = tmp_path / "empty.bundle"
        path.write_text("\n", encoding="ascii")
    elif shape == "no-revision":
        path = tmp_path / "revless.bundle"
        path.write_text(_fake_bundle_text().replace("# BUNDLE-REV 3\n", ""), encoding="ascii")
    else:  # a bundle with the root *key* in it is exactly what the shape check refuses
        path = tmp_path / "leaky.bundle"
        extra = "-----BEGIN PRIVATE KEY-----\nROOTKEY\n-----END PRIVATE KEY-----\n"
        path.write_text(_fake_bundle_text() + extra, encoding="ascii")
    result = _installer_without_root("--cert-bundle", str(path))
    assert result.returncode == 2, "the bundle refusal keeps its own exit code (2)"
    assert result.stdout == "", "a refusal goes to stderr, not stdout"
    assert reason in result.stderr
    assert str(path) in result.stderr, "the message names the file the operator passed"
    assert GRANT_PREFIX not in result.stderr, "a bad bundle is not a sudoers problem"


def test_a_well_formed_bundle_passes_validation_and_reaches_the_root_check(
    tmp_path: Path,
) -> None:
    """Validation is shape-only and never dials: the next refusal is the usual sudoers one."""
    path = tmp_path / "agent.bundle"
    path.write_text(_fake_bundle_text(), encoding="ascii")
    result = _installer_without_root("--bind", "10.0.0.5", "--cert-bundle", str(path))
    assert result.returncode == 1
    assert _grant_lines(result.stderr), "the bundle validated; what remains is the root boundary"


def test_installer_help_documents_the_cert_bundle_and_its_path() -> None:
    result = _run([BASH, str(INSTALLER), "--help"])
    assert result.returncode == 0
    assert "--cert-bundle" in result.stdout
    assert CERT_FILE_PATH in result.stdout


def test_the_bundle_lands_at_the_token_file_idiom_in_the_script() -> None:
    """`0640 root:$AGENT_USER` + chown: the agent opens the bundle itself and re-reads it."""
    text = INSTALLER.read_text()
    assert f'CERT_FILE="{CERT_FILE_PATH}"' in text
    assert 'install -m 0640 "$BUNDLE_SRC" "$CERT_FILE"' in text
    assert 'chown root:"$AGENT_USER" "$CERT_FILE"' in text
    # 0600 would fail the unit the way it would fail the token file: the service user opens
    # the bundle at startup and re-reads the denylist live. Pin the reasoning, not just the mode.
    assert "0600 root-owned file would fail the unit" in text
    assert "re-reads it when its stamp changes" in text


def test_the_tls_unit_lines_come_from_one_bundle_branch() -> None:
    """Flag + RuntimeDirectory are emitted together, or the unit is a token-mode unit."""
    text = INSTALLER.read_text()
    match = re.search(r'\n  if \[ -f "\$CERT_FILE" \]; then(.*?)\n  fi\n', text, re.DOTALL)
    assert match, "the TLS unit lines must come from one `if [ -f \"$CERT_FILE\" ]` branch"
    block = match.group(1)
    assert 'TLS_EXEC_ARGS=" --cert-bundle $CERT_FILE"' in block
    assert "RuntimeDirectory=dashboards-agent" in block
    assert "${TLS_EXEC_ARGS}" in text, "ExecStart must carry the flag from that branch"


def test_status_probes_tls_with_the_bundle_ca_and_no_client_cert() -> None:
    """The status branch: unit says --cert-bundle → python3 ssl probe, never openssl, no cert."""
    status_body = INSTALLER.read_text().split("status() {", 1)[1].split("\n}", 1)[0]
    assert "grep -q -- '--cert-bundle'" in status_body
    assert "PROTOCOL_TLS_CLIENT" in status_body
    assert "load_verify_locations" in status_body
    assert "https://" in status_body
    assert "load_cert_chain" not in status_body, "the installer holds no client certificate"
    assert 'http://$BIND:$PORT/health' in status_body, "token mode keeps the plain probe"
    assert "openssl" not in INSTALLER.read_text(), "no new binary dependency on a monitored host"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_a_bundle_install_lands_the_file_the_unit_flag_and_the_runtime_directory(
    tmp_path: Path,
) -> None:
    """The install body with a bundle: byte-identical file at 0640, TLS unit, https hint."""
    bundle = tmp_path / "agent.bundle"
    bundle.write_text(_fake_bundle_text(), encoding="ascii")
    script, call_log, env = _sandbox_installer(tmp_path)
    result = subprocess.run(
        [BASH, str(script), "--bind", "10.0.0.5", "--port", "9120", "--cert-bundle", str(bundle)],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
        check=False,
    )
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"

    cert_path = tmp_path / "target" / "certs"
    assert cert_path.is_file(), "install -m never landed the bundle"
    assert cert_path.read_text(encoding="ascii") == _fake_bundle_text(), (
        "the installed bundle must be byte-identical to the downloaded one"
    )
    mode = stat.S_IMODE(cert_path.stat().st_mode)
    assert mode == 0o640, f"bundle mode is {oct(mode)}, expected 0o640 (group = the agent user)"
    assert not mode & 0o007, "the bundle holds the agent's key: never world-readable"

    unit = (tmp_path / "target" / "dashboards-agent.service").read_text()
    assert f"--cert-bundle {cert_path}" in unit
    assert "RuntimeDirectory=dashboards-agent" in unit, (
        "ProtectSystem=strict leaves the agent no writable file but this, and the bundle "
        "unpacks its PEM blocks there"
    )
    assert "--token-file" in unit, "the dual-mode line stays — rollout is host by host"
    assert 'HOST_TARGETS="hostname=https://10.0.0.5:9120"' in result.stdout, (
        "a TLS install points the dashboard at https, with no token in the URL"
    )

    calls = call_log.read_text()
    assert calls.count("chown root:dashboards-agent") == 2, "token and bundle both chowned"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_an_unrelated_rerun_reuses_the_installed_bundle(tmp_path: Path) -> None:
    """The token-reuse idiom: a re-run without the flag neither rewrites nor drops the bundle."""
    script, _call_log, env = _sandbox_installer(tmp_path)
    target = tmp_path / "target"
    target.mkdir(parents=True)
    cert_path = target / "certs"
    cert_path.write_text(_fake_bundle_text(revision="7"), encoding="ascii")
    result = subprocess.run(
        [BASH, str(script), "--bind", "10.0.0.5"],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
        check=False,
    )
    assert result.returncode == 0, f"re-run failed:\n{result.stdout}\n{result.stderr}"
    assert cert_path.read_text(encoding="ascii") == _fake_bundle_text(revision="7")
    assert "reusing the installed certificate bundle" in result.stdout
    unit = (target / "dashboards-agent.service").read_text()
    assert f"--cert-bundle {cert_path}" in unit, "an installed bundle keeps the unit in TLS mode"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_uninstall_removes_the_bundle_with_the_token(tmp_path: Path) -> None:
    """An uninstalled agent leaves no private key behind."""
    script, call_log, env = _sandbox_installer(tmp_path)
    target = tmp_path / "target"
    (target / "lib").mkdir(parents=True)
    leftovers = {
        target / "token": "tok\n",
        target / "certs": _fake_bundle_text(),
        target / "lib" / "host_agent.py": "# stub\n",
        target / "dashboards-agent.service": "[Unit]\n",
    }
    for path, body in leftovers.items():
        path.write_text(body, encoding="ascii")
    result = subprocess.run(
        [BASH, str(script), "--uninstall"],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
        check=False,
    )
    assert result.returncode == 0, f"uninstall failed:\n{result.stdout}\n{result.stderr}"
    for path in leftovers:
        assert not path.exists(), f"uninstall left {path} on disk"
    assert "systemctl disable --now dashboards-agent" in call_log.read_text()


# --- the --status probe, end to end against the real TLS agent -----------------------------


class DictMeta:
    """The three-method `certs.MetaStore` slice, backed by a dict (the `test_auth.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 _mint_bundle() -> str:
    """A real bundle from the dashboard's own CA — the artifact the operator downloads."""
    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
    return entry.bundle


def _serve_tls_agent(host_agent: Any, bundle_text: str, where: Path) -> tuple[Any, int]:
    """The real agent handler on a TLS-wrapped loopback port, exactly as the unit runs it."""
    path = where / "installed.certs"
    path.write_text(bundle_text, encoding="ascii")
    bundle, error = host_agent.TlsBundle.load(str(path))
    assert error is None and bundle is not None
    host_agent.Handler.collector = host_agent.Collector("/", smart=False)
    host_agent.Handler.token = None
    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)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    return httpd, httpd.server_address[1]


def _run_status(script: Path, port: int, env: dict[str, str]) -> str:
    result = subprocess.run(
        [BASH, str(script), "--status", "--bind", "127.0.0.1", "--port", str(port)],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
        check=False,
    )
    assert result.returncode == 0, f"--status exited {result.returncode}:\n{result.stderr}"
    return result.stdout


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_the_tls_status_probe_reads_health_with_the_ca_and_no_client_cert(
    tmp_path: Path, host_agent_module: Any
) -> None:
    """The task-02 ASSUMPTION, proven through the installer: CERT_OPTIONAL + the /health
    exemption let a probe that holds only the bundle's CA read /health on a TLS agent."""
    bundle = _mint_bundle()
    httpd, port = _serve_tls_agent(host_agent_module, bundle, tmp_path)
    try:
        script, _call_log, env = _sandbox_installer(tmp_path / "installer")
        target = tmp_path / "installer" / "target"
        target.mkdir(parents=True)
        (target / "dashboards-agent.service").write_text(
            f"ExecStart=python3 host_agent.py --cert-bundle {target / 'certs'}\n", encoding="ascii"
        )
        (target / "certs").write_text(bundle, encoding="ascii")
        assert _installer_status(_run_status(script, port, env)) == "True"
    finally:
        httpd.shutdown()
        httpd.server_close()
        host_agent_module.Handler.bundle = None


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_the_tls_status_probe_verifies_the_chain_and_says_no_to_another_authority(
    tmp_path: Path, host_agent_module: Any
) -> None:
    """The probe loads the CA for a reason: a server signed by a different root is not health."""
    httpd, port = _serve_tls_agent(host_agent_module, _mint_bundle(), tmp_path)
    try:
        script, _call_log, env = _sandbox_installer(tmp_path / "installer")
        target = tmp_path / "installer" / "target"
        target.mkdir(parents=True)
        (target / "dashboards-agent.service").write_text(
            f"ExecStart=python3 host_agent.py --cert-bundle {target / 'certs'}\n", encoding="ascii"
        )
        (target / "certs").write_text(_mint_bundle(), encoding="ascii")  # a *different* CA
        stdout = _run_status(script, port, env)
    finally:
        httpd.shutdown()
        httpd.server_close()
        host_agent_module.Handler.bundle = None
    line = _installer_status(stdout)
    assert line != "True" and line.startswith("(no answer"), f"the probe must refuse: {line}"


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_the_tls_status_probe_names_the_bundle_when_it_cannot_read_it(tmp_path: Path) -> None:
    """The bundle is 0640 root:agent — an unprivileged --status cannot read it, and says so."""
    script, _call_log, env = _sandbox_installer(tmp_path)
    target = tmp_path / "target"
    target.mkdir(parents=True)
    (target / "dashboards-agent.service").write_text(
        f"ExecStart=python3 host_agent.py --cert-bundle {target / 'certs'}\n", encoding="ascii"
    )
    cert = target / "certs"
    cert.write_text(_fake_bundle_text(), encoding="ascii")
    cert.chmod(0o000)
    line = _installer_status(_run_status(script, 1, env))
    assert "cannot read" in line and "run --status as root" in line, line


@pytest.mark.skipif(shutil.which("curl") is None, reason="curl is not installed on this box")
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_a_token_mode_status_probe_stays_plain_http(tmp_path: Path) -> None:
    """The other branch: no bundle in the unit → the unchanged plain probe, not https."""
    script, _call_log, env = _sandbox_installer(tmp_path)
    target = tmp_path / "target"
    target.mkdir(parents=True)
    (target / "dashboards-agent.service").write_text(
        "ExecStart=python3 host_agent.py --token-file /etc/dashboards-agent.token\n",
        encoding="ascii",
    )
    result = subprocess.run(
        [BASH, str(script), "--status", "--bind", "127.0.0.1", "--port", "1"],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
        check=False,
    )
    assert result.returncode == 0
    assert _installer_status(result.stdout).startswith("(no answer"), (
        "token mode must probe plain http — nothing listens here, and the refusal is the point"
    )