felhom-agent/configs/test_felhom_os_apply.py

#!/usr/bin/env python3
"""Tests for configs/felhom-os-apply (`11` §5.4.1). A fake runner plays the host and the guest: nothing is executed
for real except the local `dpkg --compare-versions` (pure, no network). Every refusal R1–R13 has a test; the red-proof
(each test fails when its rule is removed) is `audits/os-guest-lane-2026-10-04/partB/redproof.txt`.

Run: python3 configs/test_felhom_os_apply.py   (also run by internal/osupdate's Go test)
"""
import importlib.machinery
import importlib.util
import json
import os
import pathlib
import re
import stat as statmod
import subprocess
import unittest

HERE = pathlib.Path(__file__).resolve().parent
_loader = importlib.machinery.SourceFileLoader("osapply", os.environ.get("OSAPPLY_UNDER_TEST", str(HERE / "felhom-os-apply")))  # red-proof seam
_spec = importlib.util.spec_from_loader("osapply", _loader)
osapply = importlib.util.module_from_spec(_spec)
_loader.exec_module(osapply)

PLAN = "/var/lib/felhom-agent/os/plan-t1.json"
CONF_OK = ("arch: amd64\nmp0: local-lvm:vm-9201-disk-1,mp=/var/lib/felhom,backup=1,size=70G\n"
           "mp8: /mnt/felhom-drives,mp=/mnt/felhom-drives\nrootfs: local-lvm:vm-9201-disk-0,size=32G\n")
DEB = "Debian:13.7/stable"
SEC = "Debian-Security:13/stable-security"


def dpkg_cmp(a, op, b):
    return subprocess.run(["dpkg", "--compare-versions", a, op, b]).returncode == 0


class St:
    def __init__(self, mode=statmod.S_IFREG | 0o600, uid=999, size=100):
        self.st_mode, self.st_uid, self.st_size = mode, uid, size


class Fake:
    """The host + one guest. `installed` / `live` (name -> versions in the live archive) / `snapshot` (versions
    the snapshot archive adds) / `extra_sim` (lines the simulation adds) / `dpkg_audit` / `free`."""

    def __init__(self):
        self.plan = {"release_id": "os-t1", "layer": "guest", "lane": "fast", "vmid": 9201, "mode": "apply",
                     "snapshot": "20261004T080000Z",
                     "packages": [{"name": "libc6", "version": "2.41-12+deb13u4", "origin": "Debian"},
                                  {"name": "openssl", "version": "3.5.7-1~deb13u3", "origin": "Debian-Security"}]}
        self.files = {"/etc/pve/lxc/9201.conf": CONF_OK}
        self.stats = {PLAN: St()}
        self.installed = {"libc6": "2.41-12+deb13u3", "openssl": "3.5.6-1~deb13u1", "bash": "5.2.37-2+b9"}
        self.live = {"libc6": {"2.41-12+deb13u4"}, "openssl": {"3.5.7-1~deb13u3"}}
        self.snapshot = {}
        self.snap_active = False
        self.extra_sim = []
        self.dpkg_audit = ""
        self.free = 10 * 1024 ** 3
        self.install_rc = 0
        self.calls = []
        self.logs = []
        self.written = {}
        self.status = "status: running"
        self.lock_held = False
        self.services = {}
        self.files[osapply.INSTALL_STATE] = json.dumps({"mode": "appliance"})
        self.stats[osapply.INSTALL_STATE] = St(mode=statmod.S_IFREG | 0o644, uid=0)
        # Docker / trust / facts state (v0.142.0)
        self.live_restore = "true"
        self.ids = ["aaa111", "bbb222"]
        self.engine = "29.7.2"
        self.daemon_json = '{"log-driver": "json-file"}'
        self.reload_enables = True
        self.sig_rc = 0
        self.nonces = {}
        self.clock = 1791115200.0  # 2026-10-04T12:00:00Z
        self.saved_reports = []  # R-868: (plan path, report) the wrapper kept on disk
        self.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False})
        self.stats[osapply.TRUST_FILE] = St(mode=statmod.S_IFREG | 0o644, uid=0)
        self.files[osapply.TRUST_SIGNERS] = 'felhom-op-1 namespaces="felhom-op-v1" ssh-ed25519 AAAA\n'
        self.stats[osapply.TRUST_SIGNERS] = St(mode=statmod.S_IFREG | 0o644, uid=0)

    def now(self):
        return self.clock

    def sleep(self, s):
        self.sleeps = getattr(self, "sleeps", []) + [(len(self.calls), s)]  # (calls made before it, seconds)

    def verify_sig(self, signers, key_id, ns, blob, sig):
        self.verified = (signers, key_id, ns, blob, sig)
        return self.sig_rc

    def read_nonces(self):
        return dict(self.nonces)

    def write_nonces(self, d):
        self.nonces = dict(d)

    # Runner interface
    def read_file(self, p):
        if p == PLAN:
            return json.dumps(self.plan)
        if p not in self.files:
            raise OSError("no such file")
        return self.files[p]

    def stat(self, p):
        if p not in self.stats:
            raise OSError("no such file")
        return self.stats[p]

    def agent_uid(self):
        return 999

    def log(self, line):
        self.logs.append(line)

    def save_report(self, plan_path, report):
        self.saved_reports.append((plan_path, json.loads(json.dumps(report))))
        return plan_path.replace("/plan-", "/report-")

    def host(self, argv, timeout=600, stdin=None):
        self.calls.append(("host", argv))
        if argv[0] == "/usr/sbin/pct" and argv[1] == "status":
            return 0, self.status + "\n", ""
        if argv[0] == "dpkg" and argv[1] == "--compare-versions":
            return (0 if dpkg_cmp(argv[2], argv[3], argv[4]) else 1), "", ""
        if argv[0] == "systemctl" and argv[1] == "is-active":
            return 0, "\n".join(self.services.get(s, "active") for s in argv[2:]) + "\n", ""
        return self.emulate(argv)

    def write_file(self, layer, vmid, path, body):
        self.written[path] = body
        self.write_layer = layer
        if path == osapply.SNAPSHOT_LIST:
            self.snap_active = True

    def avail(self, n):
        v = set(self.live.get(n, set()))
        if self.snap_active:
            v |= self.snapshot.get(n, set())
        return v

    def guest(self, vmid, argv, timeout=1800):
        self.calls.append(("guest", vmid, argv))
        return self.emulate(argv)

    def emulate(self, argv):
        a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"]
        cmd = a[0]
        if cmd == "dpkg-query":
            return 0, "".join(f"{n}\t{v}\tii \n" for n, v in self.installed.items()), ""
        if cmd == "dpkg" and a[1] == "--compare-versions":
            return (0 if dpkg_cmp(a[2], a[3], a[4]) else 1), "", ""
        if cmd == "dpkg" and a[1] == "--audit":
            return 0, self.dpkg_audit, ""
        if cmd == "dpkg" and a[1] == "--configure":
            self.configured_calls = getattr(self, "configured_calls", 0) + 1
            self.dpkg_journal = []  # `dpkg --configure -a` replays and empties the update journal
            return 0, "", ""
        if cmd == "fuser":
            return (0, " 123", "") if self.lock_held else (1, "", "")
        if cmd == "apt-cache" and a[1] == "madison":
            self.madison_calls = getattr(self, "madison_calls", 0) + 1
            return 0, "".join(f" {n} | {v} | http://deb.debian.org trixie/main amd64 Packages\n" for n in a[2:] for v in self.avail(n)), ""
        if cmd == "apt-cache" and a[1] == "policy":
            out = ""
            for n in a[2:]:
                out += f"{n}:\n  Installed: {self.installed.get(n)}\n  Version table:\n *** {self.installed.get(n)} 500\n        500 http://deb.debian.org/debian trixie/main amd64 Packages\n"
            return 0, out, ""
        if cmd == "apt-get":
            if "update" in a:
                return 0, "", ""
            if "clean" in a:
                return 0, "", ""
            if "-f" in a:
                self.dpkg_audit = ""
                return 0, "Setting up x (1) ...\n" if getattr(self, "repaired", False) else "", ""
            if "--print-uris" in a or "-s" in a:
                return self.sim(a)  # --print-uris prints and installs nothing, with or without -s (9202, 2026-10-05)
            if "install" in a:
                if getattr(self, "dpkg_journal", []):
                    # real apt (demo-hp 2026-10-05): a non-empty update journal refuses every install
                    return 100, "", "E: dpkg was interrupted, you must manually run 'sudo dpkg --configure -a' to correct the problem.\n"
                if self.install_rc:
                    return self.install_rc, "", "E: boom"
                for x in a:
                    if "=" in x and not x.startswith("-") and "::" not in x:
                        n, v = x.split("=", 1)
                        self.installed[n] = v
                return 0, getattr(self, "install_out", "Setting up libc6 ...\n"), ""
        if cmd == "df":
            return 0, f"Avail\n{self.free}\n", ""
        if cmd == "docker" and a[1] == "info":
            return 0, self.live_restore + "\n", ""
        if cmd == "docker" and a[1] == "version":
            return 0, self.engine + "\n", ""
        if cmd == "docker" and a[1:3] == ["ps", "-q"]:
            return 0, "".join(i + "\n" for i in self.ids), ""
        # R-528: the memory-kill check's engine (oom_image None = unreadable; oom_state / oom_event the engine's answer)
        if cmd == "date" and a[1:] == ["+%s"]:
            # each read is 3 s later than the last, so the order of the reads is visible in the values
            self.oom_epochs = getattr(self, "oom_epochs", []) + [int(self.clock) + 3 * len(getattr(self, "oom_epochs", []))]
            return 0, f"{self.oom_epochs[-1]}\n", ""
        if cmd == "docker" and a[1:4] == ["inspect", "-f", "{{.Config.Image}}"]:
            img = getattr(self, "oom_image", "gitea.dooplex.hu/admin/felhom-controller:0.300.0")
            return (0, img + "\n", "") if img is not None else (1, "", "Error: No such object: felhom-controller")
        if cmd == "docker" and a[1] == "run":
            self.oom_runs = getattr(self, "oom_runs", []) + [a]
            return 137, "", ""
        if cmd == "docker" and a[1:4] == ["inspect", "-f", "{{.State.OOMKilled}} {{.State.ExitCode}}"]:
            if getattr(self, "oom_inspect_raises", False):
                raise subprocess.TimeoutExpired(a, 10)
            return 0, getattr(self, "oom_state", "true 137") + "\n", ""
        if cmd == "docker" and a[1] == "events":
            self.oom_events_argv = a
            return 0, ("oom\n" if getattr(self, "oom_event", True) else ""), ""
        if cmd == "docker" and a[1:3] == ["rm", "-f"]:
            self.oom_removed = getattr(self, "oom_removed", []) + a[3:]
            return 0, a[3] + "\n", ""
        if cmd == "docker" and a[1] == "inspect":
            mounts = {"aaa111": "/felhom-controller|/var/run/docker.sock;/app/data;", "bbb222": "/app|/data;"}
            return 0, mounts.get(a[-1], "/other|;") + "\n", ""
        if cmd == "docker" and a[1] == "restart":
            self.restarted_containers = a[2:]
            return 0, "", ""
        if cmd == "docker" and a[1] == "exec":
            return (1, "", "Cannot connect to the Docker daemon") if getattr(self, "controller_blind", False) else (0, self.engine + "\n", "")
        if cmd == "docker":
            ids = self.ids + ["x"] * 2
            return 0, f"felhom-controller\trunning\tUp 1 hour (healthy)\t{ids[0]}\napp\trunning\tUp 1 hour (healthy)\t{ids[1]}\n", ""
        if cmd == "cat" and a[1] == osapply.DAEMON_JSON:
            return (0, self.daemon_json, "") if self.daemon_json is not None else (1, "", "No such file")
        if cmd == "cat" and a[1] == "/etc/debian_version":
            return 0, "13.7\n", ""
        if cmd == "pveversion":
            return 0, f"pve-manager/{self.installed.get('pve-manager', '9.2.2')}/abcdef (running kernel: 7.0.14-20-pve)\n", ""
        if cmd == "uname":
            return 0, "7.0.14-20-pve\n", ""
        if cmd == "apt-mark":
            return 0, getattr(self, "held", ""), ""
        if cmd == "systemctl" and a[1] == "reload":
            self.reloads = getattr(self, "reloads", 0) + 1
            if self.reload_enables and '"live-restore": true' in self.written.get(osapply.DAEMON_JSON, ""):
                self.live_restore = "true"
            return 0, "", ""
        if cmd == "systemctl" and a[1] == "restart":
            self.restarted = True
            return 0, "", ""
        if cmd == "getent":
            return 0, "1.2.3.4 deb.debian.org\n", ""
        if cmd == "sh" and a[2] == osapply.DPKG_STATE_SCRIPT:
            return 0, self.dpkg_audit + osapply.JOURNAL_MARK + "\n" + "".join(j + "\n" for j in getattr(self, "dpkg_journal", [])), ""
        if cmd == "sh":
            if "vmlinuz" in a[2]:
                return 0, "/boot/vmlinuz-7.0.2-6-pve\n/boot/vmlinuz-7.0.14-20-pve\n", ""
            if "engine=" in a[2]:
                return 0, f"debian=13.7\nengine={self.engine}\ncontainerd=2.3.3-1~debian.13~trixie\nlive={self.live_restore}\n", ""
            if "os-release" in a[2]:
                return 0, "trixie\n", ""
            if "(deleted)" in a[2]:
                return 0, getattr(self, "restart_out", ""), ""
            return 0, "", ""
        if cmd == "rm":
            self.snap_active = False
            return 0, "", ""
        return 1, "", f"unexpected guest call {a}"

    def sim(self, a):
        if "--print-uris" in a:
            if "-s" in a:
                # real apt (measured 9202 2026-10-05): with -s it prints the SIMULATION, no URI list
                return 0, "Inst libc6 [2.41-12+deb13u4] (2.41-12+deb13u4 Debian:13.7/stable [amd64])\n", ""
            # real apt's line shape, verbatim from 9202 2026-10-05 (audits/night-fixes-2026-10-05/partC/)
            return 0, ("Need to get 4347 kB of archives.\n"
                       "'http://deb.debian.org/debian/pool/main/b/bash/bash_5.2.37-2%2bb10_amd64.deb' bash_5.2.37-2+b10_amd64.deb 1500792 MD5Sum:27b11721fea83d73b96e0f7023863771\n"
                       "'http://deb.debian.org/debian/pool/main/g/glibc/libc6_2.41-12%2bdeb13u4_amd64.deb' libc6_2.41-12+deb13u4_amd64.deb 2846580 MD5Sum:5559581916477ef1f57ea9f82cecf22e\n"
                       + getattr(self, "extra_uris", "")), ""
        if "dist-upgrade" in a:
            if getattr(self, "pending_sim", None) is not None and not getattr(self, "_pending_used", False):
                self._pending_used = True
                return 0, "\n".join(self.pending_sim) + "\n", ""
            return 0, "Inst bash [5.2.37-2+b9] (5.2.37-2+b10 Debian:13.7/stable [amd64])\n", ""
        out = ""
        for x in a:
            if "=" in x and not x.startswith("-") and "::" not in x:
                n, v = x.split("=", 1)
                if v not in self.avail(n):
                    return 100, "", f"E: Version '{v}' for '{n}' was not found"
                origin = getattr(self, "origins", {}).get(n) or ("Docker CE:trixie" if n in osapply.DOCKER_NAMES else SEC if n == "openssl" else DEB)
                out += f"Inst {n} [{self.installed[n]}] ({v} {origin} [amd64])\n"
        out += "".join(l + "\n" for l in self.extra_sim)
        return 0, out, ""


def run(f):
    import io
    import contextlib
    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        rc = osapply.main(["felhom-os-apply", "--plan", PLAN], runner=f)
    line = [l for l in buf.getvalue().splitlines() if l.startswith("OSAPPLY-REPORT ")][-1]
    return rc, json.loads(line[len("OSAPPLY-REPORT "):])


class Happy(unittest.TestCase):
    def test_apply_installs_exactly_the_plan(self):
        f = Fake()
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(f.installed["libc6"], "2.41-12+deb13u4")
        self.assertEqual(f.installed["openssl"], "3.5.7-1~deb13u3")
        self.assertEqual(f.installed["bash"], "5.2.37-2+b9", "a package outside the plan was changed")
        self.assertEqual(rep["plan"]["upgrade"], 2)
        self.assertIn("installed", rep)
        self.assertEqual(rep["pending"][0]["name"], "bash")
        self.assertTrue(any(l.startswith("os-apply: REPAIR ") for l in f.logs), "the repair line must always print")
        self.assertTrue(any(l.startswith("os-apply: DONE rc=0") for l in f.logs))
        inst = [c for c in f.calls if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2] and "--print-uris" not in c[2]]
        self.assertTrue(inst and "Dpkg::Options::=--force-confold" in inst[0][2], "must keep existing config files")

    def test_already_current_is_a_no_op(self):
        f = Fake()
        f.installed.update(libc6="2.41-12+deb13u4", openssl="3.5.7-1~deb13u3")
        rc, rep = run(f)
        self.assertEqual(rc, 0)
        self.assertEqual(rep["plan"]["upgrade"], 0)

    def test_inventory_installs_nothing(self):
        f = Fake()
        f.plan["mode"] = "inventory"
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(f.installed["libc6"], "2.41-12+deb13u3")
        self.assertIn("installed", rep)

    def test_health_mode(self):
        f = Fake()
        f.plan["mode"] = "health"
        rc, rep = run(f)
        self.assertEqual(rc, 0)
        self.assertEqual(rep["health"]["controller"], "healthy")


class Repair(unittest.TestCase):
    def test_repair_runs_first_and_is_reported(self):
        f = Fake()
        f.dpkg_audit = "The following packages have been unpacked but not yet configured.\n perl  Larry Wall's\n"
        f.repaired = True
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["repair"]["half_configured_before"], 1)
        self.assertEqual(rep["repair"]["fixed"], 1)
        order = [i for i, c in enumerate(f.calls) if c[0] == "guest" and c[2][-1:] != ["update"]]
        first_cfg = next(i for i, c in enumerate(f.calls) if c[0] == "guest" and "--configure" in c[2])
        first_upd = next(i for i, c in enumerate(f.calls) if c[0] == "guest" and "update" in c[2])
        self.assertLess(first_cfg, first_upd, "the repair must run before anything else touches apt")
        self.assertTrue(order)


class Snapshot(unittest.TestCase):
    def test_a_replaced_version_comes_from_the_snapshot(self):
        f = Fake()
        f.live["openssl"] = {"3.5.7-1~deb13u4"}  # Debian moved on
        f.snapshot["openssl"] = {"3.5.7-1~deb13u3"}
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["plan"]["from_snapshot"], 1)
        self.assertEqual(f.installed["openssl"], "3.5.7-1~deb13u3", "must install the APPROVED version, not the newer one")
        body = f.written[osapply.SNAPSHOT_LIST]
        self.assertIn("snapshot.debian.org/archive/debian/20261004T080000Z trixie main", body)
        self.assertIn("debian-security/20261004T080000Z trixie-security main", body)
        self.assertFalse(f.snap_active, "the temporary snapshot sources must be removed after the run")

    def test_snapshot_does_not_have_it_either(self):
        f = Fake()
        f.live["openssl"] = set()
        rc, rep = run(f)
        self.assertEqual((rc, rep["refused"]["code"]), (2, "R7"))
        self.assertFalse(f.snap_active)


class Refusals(unittest.TestCase):
    def refused(self, f, code):
        rc, rep = run(f)
        self.assertEqual(rc, 2, rep)
        self.assertEqual(rep["refused"]["code"], code, rep)
        self.assertTrue(any(l.startswith(f"os-apply: REFUSED: {code} ") for l in f.logs), f.logs)
        inst = [c for c in f.calls if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2] and "--print-uris" not in c[2]]
        self.assertEqual(inst, [], "a refusal must install nothing")
        return rep

    def test_R1_usage(self):
        import io
        import contextlib
        f = Fake()
        with contextlib.redirect_stdout(io.StringIO()):
            self.assertEqual(osapply.main(["felhom-os-apply", "--plan", PLAN, "--extra"], runner=f), 2)
            self.assertEqual(osapply.main(["felhom-os-apply", "--plan"], runner=f), 2)

    def test_R1_path_outside_the_plan_dir(self):
        import io
        import contextlib
        f = Fake()
        with contextlib.redirect_stdout(io.StringIO()):
            rc = osapply.main(["felhom-os-apply", "--plan", "/tmp/plan-x.json"], runner=f)
        self.assertEqual(rc, 2)
        self.assertTrue(any("R1" in l for l in f.logs))

    def test_R1_symlink(self):
        f = Fake()
        f.stats[PLAN] = St(mode=statmod.S_IFLNK | 0o777)
        self.refused(f, "R1")

    def test_R1_not_owned_by_the_agent(self):
        f = Fake()
        f.stats[PLAN] = St(uid=0)
        self.refused(f, "R1")

    def test_R2_non_debian_origin_in_the_plan(self):
        f = Fake()
        f.plan["packages"][0]["origin"] = "Proxmox"
        self.refused(f, "R2")

    def test_R2_non_debian_origin_in_the_simulation(self):
        f = Fake()
        f.installed["libc6"] = "2.41-12+deb13u3"
        orig = f.sim

        def sim(a):
            rc, out, err = orig(a)
            return rc, out.replace("Debian:13.7/stable", "Proxmox Debian Repository:stable"), err
        f.sim = sim
        self.refused(f, "R2")

    def test_R3_slow_lane(self):
        f = Fake()
        f.plan["lane"] = "slow"
        self.refused(f, "R3")

    def test_R4_removal(self):
        f = Fake()
        f.extra_sim = ["Remv bash [5.2.37-2+b9]"]
        self.refused(f, "R4")

    def test_R5_downgrade(self):
        f = Fake()
        f.installed["libc6"] = "2.41-12+deb13u4"
        f.plan["packages"] = [{"name": "openssl", "version": "3.5.7-1~deb13u3", "origin": "Debian-Security"}]
        f.extra_sim = ["Inst openssl [3.5.6-1~deb13u1] (3.5.5-1 Debian:13.7/stable [amd64])"]
        orig = f.sim

        def sim(a):  # the simulation answers with a LOWER version than installed
            rc, out, err = orig(a)
            return rc, "\n".join(l for l in out.splitlines() if not l.startswith("Inst openssl [3.5.6-1~deb13u1] (3.5.7")) + "\n", err
        f.sim = sim
        f.plan["packages"][0]["version"] = "3.5.7-1~deb13u3"
        rep = run(f)[1]
        # The plan asks 3.5.7; the simulation goes to 3.5.5: that is BOTH a wrong version (R6) and a downgrade.
        self.assertIn(rep["refused"]["code"], ("R5", "R6"))

    def test_R5_downgrade_exact(self):
        f = Fake()
        f.plan["packages"] = [{"name": "openssl", "version": "3.5.7-1~deb13u3", "origin": "Debian-Security"}]
        f.installed["openssl"] = "3.5.6-1~deb13u1"
        orig = f.sim

        def sim(a):
            rc, out, err = orig(a)
            return rc, out.replace("[3.5.6-1~deb13u1]", "[3.5.8-1]"), err
        f.sim = sim
        self.refused(f, "R5")

    def test_R6_new_package(self):
        f = Fake()
        f.extra_sim = ["Inst newthing (1.0 Debian:13.7/stable [amd64])"]
        self.refused(f, "R6")

    def test_R6_unlisted_package(self):
        f = Fake()
        f.extra_sim = ["Inst bash [5.2.37-2+b9] (5.2.37-2+b10 Debian:13.7/stable [amd64])"]
        self.refused(f, "R6")

    def test_R6_allow_new_is_slow_lane(self):
        f = Fake()
        f.plan["allow_new"] = ["proxmox-kernel-x"]
        self.refused(f, "R6")

    def test_R7_not_downloadable_and_no_snapshot(self):
        f = Fake()
        f.live["openssl"] = set()
        f.plan["snapshot"] = ""
        self.refused(f, "R7")

    def test_R8_free_space(self):
        f = Fake()
        f.free = 100 * 1024 * 1024
        self.refused(f, "R8")

    # R-865: the download is the real one. 2 GB of URIs, 5 GB free: 5 GB < 3 x 2 GB -> R8, with the size in the line.
    # COMPANION RED-PROOF: put "-s" back into download_bytes -> 0 B -> no refusal -> this test fails.
    def test_R8_counts_the_real_download(self):
        f = Fake()
        f.free = 5 * 1024 ** 3
        f.extra_uris = "'http://deb.debian.org/debian/pool/main/b/big/big_1_amd64.deb' big_1_amd64.deb 2000000000 MD5Sum:x\n"
        rep = self.refused(f, "R8")
        self.assertIn("download 2004347372 B", str(rep))

    def test_download_bytes_never_simulates(self):
        f = Fake()
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        calls = [c[2] for c in f.calls if c[0] == "guest" and "--print-uris" in c[2]]
        self.assertTrue(calls, "download_bytes was never called")
        for c in calls:
            self.assertNotIn("-s", c, f"--print-uris with -s prints no URIs: {c}")

    def test_R9_guest_locked_by_a_backup(self):
        f = Fake()
        f.files["/etc/pve/lxc/9201.conf"] = CONF_OK + "lock: backup\n"
        self.refused(f, "R9")

    def test_R9_apt_lock_held(self):
        f = Fake()
        f.lock_held = True
        self.refused(f, "R9")

    def test_R10_not_the_boxs_own_guest(self):
        f = Fake()
        f.files["/etc/pve/lxc/9201.conf"] = CONF_OK.replace("mp8: /mnt/felhom-drives,", "mp8: /mnt/hdd_1/scratch,")
        self.refused(f, "R10")

    def test_R10_reserved_vmid(self):
        f = Fake()
        f.plan["vmid"] = 990003
        self.refused(f, "R10")

    def test_R10_bind_only_in_a_snapshot_section(self):
        f = Fake()
        f.files["/etc/pve/lxc/9201.conf"] = "rootfs: x\n[snap1]\nmp8: /mnt/felhom-drives,mp=/mnt/felhom-drives\n"
        self.refused(f, "R10")

    def test_R10_not_running(self):
        f = Fake()
        f.status = "status: stopped"
        self.refused(f, "R10")

    def test_R11_duplicate(self):
        f = Fake()
        f.plan["packages"].append(dict(f.plan["packages"][0]))
        self.refused(f, "R11")

    def test_R11_bad_version_string(self):
        f = Fake()
        f.plan["packages"][0]["version"] = "1.0; rm -rf /"
        self.refused(f, "R11")

    def test_R11_bad_name(self):
        f = Fake()
        f.plan["packages"][0]["name"] = "--purge"
        self.refused(f, "R11")

    def test_R12_unknown_layer(self):
        f = Fake()
        f.plan["layer"] = "vm"
        self.refused(f, "R12")

    def test_R12_host_on_a_byo_box(self):
        f = Fake()
        f.plan["layer"] = "host"
        f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"})
        self.refused(f, "R12")

    def test_R12_host_without_an_install_record(self):
        f = Fake()
        f.plan["layer"] = "host"
        del f.stats[osapply.INSTALL_STATE]
        self.refused(f, "R12")

    def test_R12_host_record_not_root_owned(self):
        # the agent can write agent.json's deployment_mode; only a ROOT-owned record proves anything
        f = Fake()
        f.plan["layer"] = "host"
        f.stats[osapply.INSTALL_STATE] = St(mode=statmod.S_IFREG | 0o644, uid=999)
        self.refused(f, "R12")

    def test_R14_kernel_package_in_a_host_plan(self):
        f = Fake()
        f.plan["layer"] = "host"
        f.plan["packages"].append({"name": "linux-image-amd64", "version": "6.12.1-1", "origin": "Debian"})
        self.refused(f, "R14")

    def test_R14_kernel_package_pulled_by_the_simulation(self):
        f = Fake()
        f.plan["layer"] = "host"
        f.extra_sim = ["Inst grub-common [2.12-9] (2.12-10 Debian:13.7/stable [amd64])"]
        f.installed["grub-common"] = "2.12-9"
        f.plan["packages"].append({"name": "grub-common", "version": "2.12-10", "origin": "Debian"})
        self.refused(f, "R14")

    def test_R13_repair_does_not_fix_it(self):
        f = Fake()
        f.dpkg_audit = "The following packages are broken\n perl\n"
        orig = f.guest

        def guest(vmid, argv, timeout=1800):
            rc, out, err = orig(vmid, argv, timeout)
            if "-f" in argv:
                f.dpkg_audit = "The following packages are broken\n perl\n"
            return rc, out, err
        f.guest = guest
        self.refused(f, "R13")


class Conffiles(unittest.TestCase):
    # dpkg's two shapes, measured live 2026-10-04: an unchanged file is UPDATED; a locally changed one is KEPT.
    def test_updated_vs_kept(self):
        f = Fake()
        f.install_out = ("Installing new version of config file /etc/debian_version ...\n"
                         "Configuration file '/etc/ssh/sshd_config'\n ==> Modified (by you or by a script) since installation.\n"
                         " ==> Keeping old config file as default.\n")
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertIn("os-apply: CONFFILE updated /etc/debian_version (it was not changed locally)", f.logs)
        self.assertTrue(any(l.startswith("os-apply: CONFFILE kept /etc/ssh/sshd_config") for l in f.logs), f.logs)
        self.assertEqual(rep["conffiles_kept"], ["/etc/ssh/sshd_config"])
        self.assertFalse(any("kept /etc/debian_version" in l for l in f.logs), "an updated file must not be reported as kept")


class HostLayer(unittest.TestCase):
    def test_host_runs_on_the_host_not_in_the_guest(self):
        f = Fake()
        f.plan["layer"] = "host"
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        inst = [c for c in f.calls if c[0] == "host" and "install" in c[1] and "-s" not in c[1] and "-f" not in c[1]]
        self.assertTrue(inst, "the host install must run on the host")
        self.assertFalse([c for c in f.calls if c[0] == "guest" and "install" in c[2]], "nothing installed in the guest")
        self.assertEqual(sorted(rep["health_after"]["host_services"]), sorted(osapply.HOST_SERVICES))
        self.assertTrue(rep["health_after"]["guest_running"])

    def test_pending_fast_skips_proxmox_docker_and_kernel(self):
        f = Fake()
        f.plan["layer"] = "host"
        f.plan["select"] = "pending-fast"
        f.plan["packages"] = []
        f.installed.update({"pve-manager": "9.2.2", "linux-image-amd64": "6.12.1", "docker-ce": "29.7"})
        f.pending_sim = [
            "Inst libc6 [2.41-12+deb13u3] (2.41-12+deb13u4 Debian:13.7/stable [amd64])",
            "Inst openssl [3.5.6-1~deb13u1] (3.5.7-1~deb13u3 Debian:13.7/stable, Debian-Security:13/stable-security [amd64])",
            "Inst pve-manager [9.2.2] (9.2.21 Proxmox Debian Repository:stable [amd64])",
            "Inst linux-image-amd64 [6.12.1] (6.12.9 Debian:13.7/stable [amd64])",
            "Inst docker-ce [29.7] (29.8 Docker CE:trixie [amd64])",
            "Inst brand-new (1.0 Debian:13.7/stable [amd64])",
        ]
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        got = sorted(u["name"] for u in rep["upgraded"])
        self.assertEqual(got, ["libc6", "openssl"], "pending-fast must take only installed, Debian-origin, non-kernel packages")

    def test_reboot_needed_when_pid1_or_lxc_start(self):
        f = Fake()
        f.plan["layer"] = "host"
        f.restart_out = "1 systemd\n2101 lxc-start\n530 sshd\n"
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertTrue(rep["reboot_needed"])
        self.assertIn("lxc-start", rep["restart_needed"])

    def test_reboot_needed_for_lxc_start_alone(self):
        # lxc-start runs the guest; only a guest restart (or a host reboot) replaces it
        f = Fake()
        f.plan["layer"] = "host"
        f.restart_out = "2101 lxc-start\n530 sshd\n"
        rc, rep = run(f)
        self.assertTrue(rep["reboot_needed"], rep)

    def test_every_layer_scans_every_pass(self):
        # R-849 (v0.142.0): host AND guest are scanned on every pass, so a reboot / restart clears the flag.
        for layer in ("host", "guest"):
            f = Fake()
            f.plan["layer"] = layer
            f.plan["mode"] = "inventory"
            f.restart_out = "2101 lxc-start\n" if layer == "host" else "1 systemd\n"
            rc, rep = run(f)
            self.assertEqual(rc, 0, rep)
            self.assertTrue(rep["reboot_scanned"] and rep["reboot_needed"], (layer, rep))
            f = Fake()
            f.plan["layer"] = layer
            f.plan["mode"] = "inventory"
            f.restart_out = ""
            rc, rep = run(f)
            self.assertTrue(rep["reboot_scanned"] and rep["reboot_needed"] is False, (layer, rep))

    def test_no_reboot_for_ordinary_daemons(self):
        f = Fake()
        f.plan["layer"] = "host"
        f.restart_out = "530 sshd\n611 cron\n"
        rc, rep = run(f)
        self.assertFalse(rep["reboot_needed"], rep)


class Speed(unittest.TestCase):
    # R-845: no `pct exec` per package — version checks on the host, madison once, the restart scan only after an install.
    def test_no_per_package_guest_calls(self):
        f = Fake()
        for i in range(40):
            f.installed[f"pkg{i}"] = "1.0-1"
            f.live[f"pkg{i}"] = {"1.0-2"}
            f.plan["packages"].append({"name": f"pkg{i}", "version": "1.0-2", "origin": "Debian"})
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        guest_cmp = [c for c in f.calls if c[0] == "guest" and "--compare-versions" in c[2]]
        self.assertEqual(guest_cmp, [], "version comparisons must run on the host")
        self.assertEqual(f.madison_calls, 1, "madison must run once for all packages")
        guest_calls = len([c for c in f.calls if c[0] == "guest"])
        self.assertLess(guest_calls, 30, f"{guest_calls} guest calls for 42 packages — something is per-package again")


class Failure(unittest.TestCase):
    def test_install_failure_is_rc3_with_dpkg_state(self):
        f = Fake()
        f.install_rc = 100
        rc, rep = run(f)
        self.assertEqual(rc, 3)
        self.assertEqual(rep["failed"]["rc"], 100)
        self.assertTrue(any(l.startswith("os-apply: FAILED rc=100 step=install") for l in f.logs))



class RestartSkipPattern(unittest.TestCase):
    """The cgroup filter runs as `grep -q PATTERN /proc/<pid>/cgroup`; check it with grep itself against the cgroup
    lines measured on demo-felhom 2026-10-04."""

    def grep(self, pattern, line):
        return subprocess.run(["grep", "-q", pattern], input=line + "\n", text=True).returncode == 0

    def test_restart_skip_patterns_against_real_cgroups(self):
        host = osapply.RESTART_SKIP_CGROUP["host"]
        self.assertTrue(self.grep(host, "0::/lxc/9201/ns/system.slice/docker.service"), "a guest process must be skipped")
        self.assertFalse(self.grep(host, "0::/lxc.monitor/9201"), "lxc-start must NOT be skipped (it runs the guest)")
        self.assertFalse(self.grep(host, "0::/system.slice/pve-cluster.service"), "a host daemon must NOT be skipped")
        guest = osapply.RESTART_SKIP_CGROUP["guest"]
        self.assertTrue(self.grep(guest, "0::/system.slice/docker-0123abcd.scope"))
        self.assertFalse(self.grep(guest, "0::/system.slice/cron.service"))


DOCKER_SET = [{"name": "docker-ce", "version": "5:29.8.2-1~debian.13~trixie", "origin": "Docker CE"},
              {"name": "containerd.io", "version": "2.3.6-1~debian.13~trixie", "origin": "Docker CE"}]


def docker_fake(signed=None, undo=False, ring0=False):
    f = Fake()
    f.installed.update({"docker-ce": "5:29.7.2-1~debian.13~trixie", "containerd.io": "2.3.3-1~debian.13~trixie"})
    f.live["docker-ce"] = {"5:29.8.2-1~debian.13~trixie", "5:29.7.2-1~debian.13~trixie"}
    f.live["containerd.io"] = {"2.3.6-1~debian.13~trixie", "2.3.3-1~debian.13~trixie"}
    f.plan = {"release_id": "os-docker-t1", "layer": "docker", "lane": "slow", "vmid": 9201, "mode": "apply",
              "packages": [dict(p) for p in DOCKER_SET]}
    if undo:
        f.plan["undo"] = True
    if ring0:
        f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": True})
    if signed is not None:
        f.plan["signed"] = signed
    return f


def signed_job(packages=DOCKER_SET, host="demo-hp-bb76ea", op="os_docker_step", undo=False, nonce="n1",
               issued="2026-10-04T11:50:00Z", expires="2026-10-04T12:30:00Z"):
    import base64
    params = {"packages": packages, "undo": undo}
    blob = json.dumps({"expires_at": expires, "issued_at": issued, "key_id": "felhom-op-1", "nonce": nonce, "op": op,
                       "params": params, "target": {"guest_id": "", "host_id": host}}, sort_keys=True).encode()
    return {"blob_b64": base64.b64encode(blob).decode(), "sig": "-----BEGIN SSH SIGNATURE-----\nx\n-----END SSH SIGNATURE-----\n"}


class DockerLane(unittest.TestCase):
    """`11` §5.8, agent v0.142.0. Each test names the refusal it pins; the red-proof file mutates each one."""

    def refused(self, f, code):
        rc, rep = run(f)
        self.assertEqual(rc, 2, rep)
        self.assertEqual(rep["refused"]["code"], code, rep)
        self.assertEqual(f.installed.get("docker-ce", "5:29.7.2-1~debian.13~trixie"), "5:29.7.2-1~debian.13~trixie")
        return rep

    def test_docker_package_in_a_fast_plan_is_refused(self):
        f = Fake()
        f.plan["packages"].append({"name": "docker-ce", "version": "5:29.8.2-1~debian.13~trixie", "origin": "Debian"})
        self.refused(f, "R2")

    def test_docker_layer_in_the_fast_lane_is_refused(self):
        f = docker_fake(ring0=True)
        f.plan["lane"] = "fast"
        self.refused(f, "R3")

    def test_no_authority_is_refused(self):
        self.refused(docker_fake(), "R3")

    def test_ring0_mark_allows_pending_docker(self):
        f = docker_fake(ring0=True)
        f.plan["select"], f.plan["packages"] = "pending-docker", []
        f.pending_sim = ["Inst docker-ce [5:29.7.2-1~debian.13~trixie] (5:29.8.2-1~debian.13~trixie Docker CE:trixie [amd64])",
                         "Inst containerd.io [2.3.3-1~debian.13~trixie] (2.3.6-1~debian.13~trixie Docker CE:trixie [amd64])",
                         "Inst bash [5.2.37-2+b9] (5.2.37-2+b10 Debian:13.7/stable [amd64])"]
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["authority"], "ring0")
        self.assertEqual(f.installed["docker-ce"], "5:29.8.2-1~debian.13~trixie")
        self.assertEqual(f.installed["bash"], "5.2.37-2+b9", "a Debian package must not ride a Docker step")
        self.assertFalse(getattr(f, "restarted", False), "never a docker restart")

    def test_signed_job_applies_exactly_its_packages(self):
        f = docker_fake(signed=signed_job())
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["authority"], "signed")
        self.assertEqual(f.installed["containerd.io"], "2.3.6-1~debian.13~trixie")
        self.assertEqual(f.verified[0], osapply.TRUST_SIGNERS, "the ROOT-OWNED signers file, not the agent's config")
        self.assertIn("n1", f.nonces)

    def test_bad_signature_is_refused(self):
        f = docker_fake(signed=signed_job())
        f.sig_rc = 255
        self.refused(f, "R3")
        self.assertEqual(f.nonces, {}, "a bad signature must not burn a nonce")

    def test_signed_job_for_another_host_is_refused(self):
        self.refused(docker_fake(signed=signed_job(host="demo-felhom-8363b5")), "R3")

    def test_signed_job_other_op_is_refused(self):
        self.refused(docker_fake(signed=signed_job(op="agent_update")), "R3")

    def test_expired_signed_job_is_refused(self):
        self.refused(docker_fake(signed=signed_job(expires="2026-10-04T11:55:00Z")), "R3")

    def test_replayed_signed_job_is_refused(self):
        f = docker_fake(signed=signed_job())
        f.nonces = {"n1": f.clock + 600}
        self.refused(f, "R3")

    def test_plan_must_equal_the_signed_packages(self):
        f = docker_fake(signed=signed_job(packages=DOCKER_SET[:1]))
        self.refused(f, "R3")

    def test_agent_writable_trust_file_is_refused(self):
        f = docker_fake(ring0=True)
        f.stats[osapply.TRUST_FILE] = St(mode=statmod.S_IFREG | 0o644, uid=999)
        self.refused(f, "R3")

    def test_live_restore_off_is_refused(self):
        f = docker_fake(signed=signed_job())
        f.live_restore = "false"
        self.refused(f, "R15")

    def test_undo_needs_a_signed_job(self):
        self.refused(docker_fake(undo=True, ring0=True), "R3")

    def test_signed_undo_downgrades(self):
        old = [{"name": "docker-ce", "version": "5:29.7.2-1~debian.13~trixie", "origin": "Docker CE"}]
        f = docker_fake(signed=signed_job(packages=old, undo=True), undo=True)
        f.plan["packages"] = [dict(p) for p in old]
        f.installed["docker-ce"] = "5:29.8.2-1~debian.13~trixie"
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(f.installed["docker-ce"], "5:29.7.2-1~debian.13~trixie")
        self.assertTrue(rep["undo"])

    def test_unsigned_downgrade_is_refused(self):
        f = docker_fake(ring0=True)
        f.plan["packages"] = [{"name": "docker-ce", "version": "5:29.6.0-1~debian.13~trixie", "origin": "Docker CE"}]
        f.live["docker-ce"].add("5:29.6.0-1~debian.13~trixie")
        rc, rep = run(f)
        self.assertEqual(rep["plan"]["upgrade"], 0, "an older version on an unsigned step is 'already', never installed")

    def test_step_restarts_only_the_socket_users(self):
        # R-858: after an engine step, ONLY the container that mounts the docker socket is restarted (here the controller)
        f = docker_fake(signed=signed_job())
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(f.restarted_containers, ["felhom-controller"])
        self.assertEqual(rep["socket_restarted"], ["felhom-controller"])

    def test_no_install_restarts_nothing(self):
        f = docker_fake(signed=signed_job())
        f.installed.update({"docker-ce": "5:29.8.2-1~debian.13~trixie", "containerd.io": "2.3.6-1~debian.13~trixie"})
        rc, rep = run(f)
        self.assertFalse(hasattr(f, "restarted_containers"), "nothing installed -> no container restart")

    def test_health_says_when_the_controller_cannot_reach_docker(self):
        f = docker_fake(signed=signed_job())
        f.controller_blind = True
        rc, rep = run(f)
        self.assertFalse(rep["health_after"]["controller_docker_ok"])

    def test_health_carries_container_ids(self):
        f = docker_fake(signed=signed_job())
        rc, rep = run(f)
        self.assertEqual(rep["health_after"]["containers"]["app"]["id"], "bbb222")
        self.assertEqual(rep["docker_engine"], "29.7.2")


class LiveRestore(unittest.TestCase):
    def lr(self):
        f = Fake()
        f.plan = {"release_id": "lr", "layer": "guest", "lane": "fast", "vmid": 9201, "mode": "live-restore-on", "packages": []}
        f.live_restore = "false"
        return f

    def test_turns_it_on_with_a_reload_never_a_restart(self):
        f = self.lr()
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(json.loads(f.written[osapply.DAEMON_JSON]), {"log-driver": "json-file", "live-restore": True})
        self.assertEqual(f.reloads, 1)
        self.assertFalse(getattr(f, "restarted", False))
        self.assertEqual(rep["live_restore"]["result"], "on")
        self.assertTrue(rep["live_restore"]["same_ids"])

    def test_already_on_writes_nothing(self):
        f = self.lr()
        f.live_restore = "true"
        rc, rep = run(f)
        self.assertEqual(rc, 0)
        self.assertNotIn(osapply.DAEMON_JSON, f.written)

    def test_invalid_daemon_json_is_left_alone(self):
        f = self.lr()
        f.daemon_json = "{not json"
        rc, rep = run(f)
        self.assertEqual(rep["refused"]["code"], "R16")
        self.assertNotIn(osapply.DAEMON_JSON, f.written)

    def test_reload_that_does_not_enable_puts_the_file_back(self):
        f = self.lr()
        f.reload_enables = False
        rc, rep = run(f)
        self.assertEqual(rc, 3, rep)
        self.assertEqual(f.written[osapply.DAEMON_JSON], '{"log-driver": "json-file"}')
        self.assertFalse(getattr(f, "restarted", False))


class Facts(unittest.TestCase):
    def facts(self, f=None):
        f = f or Fake()
        f.plan = {"release_id": "facts", "layer": "host", "lane": "fast", "vmid": 9201, "mode": "facts", "packages": []}
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        return f, rep["facts"]

    def test_reads_host_and_guest(self):
        f = Fake()
        f.files["/proc/sys/kernel/tainted"] = "4225\n"  # 4096 + 128 (D: oops) + 1
        f.files["/proc/sys/kernel/panic"] = "10\n"
        f.held = "tzdata\n"
        f.files["/etc/default/grub"] = "GRUB_DEFAULT=saved\n"
        f.files["/boot/grub/grubenv"] = "# GRUB Environment Block\nsaved_entry=gnulinux-advanced-x>gnulinux-7.0.14-20-pve-advanced-x\n"
        _, fa = self.facts(f)
        h, g = fa["host"], fa["guest"]
        self.assertEqual((h["debian"], h["kernel_running"], h["kernel_next_boot"]), ("13.7", "7.0.14-20-pve", "7.0.14-20-pve"))
        self.assertEqual(h["held"], ["tzdata"])
        self.assertTrue(h["oops_this_boot"])
        self.assertEqual(h["kernel_panic"], 10)
        self.assertEqual((g["debian"], g["docker_engine"], g["live_restore"]), ("13.7", "29.7.2", "on"))
        self.assertEqual(g["containerd"], "2.3.3-1~debian.13~trixie")

    def test_next_entry_wins_and_default_zero_is_the_newest(self):
        f = Fake()
        f.files["/etc/default/grub"] = "GRUB_DEFAULT=saved\n"
        f.files["/boot/grub/grubenv"] = "saved_entry=gnulinux-advanced-x>gnulinux-7.0.2-6-pve-advanced-x\nnext_entry=gnulinux-advanced-x>gnulinux-7.0.14-20-pve-advanced-x\n"
        _, fa = self.facts(f)
        self.assertEqual(fa["host"]["kernel_next_boot"], "7.0.14-20-pve")
        self.assertIn("next_entry", fa["host"]["kernel_next_boot_source"])
        f = Fake()
        f.files["/etc/default/grub"] = "GRUB_DEFAULT=0\n"
        _, fa = self.facts(f)
        self.assertEqual(fa["host"]["kernel_next_boot"], "7.0.14-20-pve", "dpkg order, not string order (7.0.2 < 7.0.14)")

    def test_stopped_guest_is_unknown_never_guessed(self):
        f = Fake()
        f.status = "status: stopped"
        _, fa = self.facts(f)
        self.assertEqual(fa["guest"]["docker_engine"], "unknown")
        self.assertIn("R10", fa["guest"]["unknown_reason"])
        self.assertEqual(fa["host"]["tainted"], None, "unreadable -> None, not 0")


class RealSignatureCheck(unittest.TestCase):
    """The REAL Runner.verify_sig with the real ssh-keygen and a throwaway key, in the installer's allowed_signers form
    (`<key_id> namespaces="felhom-op-v1" <type> <b64> <comment>`). Nothing leaves the temp dir."""

    def setUp(self):
        import shutil
        if not shutil.which("ssh-keygen"):
            self.skipTest("ssh-keygen not available")
        import tempfile
        self.d = tempfile.mkdtemp()
        self.key = os.path.join(self.d, "k")
        subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "felhom-op-1", "-f", self.key], check=True)
        pub = open(self.key + ".pub").read().strip()
        self.signers = os.path.join(self.d, "signers")
        open(self.signers, "w").write(f'felhom-op-1 namespaces="felhom-op-v1" {pub}\n')

    def sign(self, blob, ns="felhom-op-v1"):
        bp = os.path.join(self.d, "blob")
        open(bp, "wb").write(blob)
        if os.path.exists(bp + ".sig"):
            os.remove(bp + ".sig")  # ssh-keygen -Y sign asks before overwriting (it would wait on stdin)
        subprocess.run(["ssh-keygen", "-q", "-Y", "sign", "-f", self.key, "-n", ns, bp], check=True, stdin=subprocess.DEVNULL, timeout=30)
        return open(bp + ".sig").read()

    def test_good_signature_verifies(self):
        blob = b'{"op":"os_docker_step"}'
        self.assertEqual(osapply.Runner().verify_sig(self.signers, "felhom-op-1", "felhom-op-v1", blob, self.sign(blob)), 0)

    def test_changed_blob_wrong_namespace_or_wrong_principal_fail(self):
        blob = b'{"op":"os_docker_step"}'
        sig = self.sign(blob)
        r = osapply.Runner()
        self.assertNotEqual(r.verify_sig(self.signers, "felhom-op-1", "felhom-op-v1", blob + b" ", sig), 0)
        self.assertNotEqual(r.verify_sig(self.signers, "someone-else", "felhom-op-v1", blob, sig), 0)
        self.assertNotEqual(r.verify_sig(self.signers, "felhom-op-1", "felhom-op-v1", blob, self.sign(blob, ns="other-ns")), 0)



class UnsentReport(unittest.TestCase):
    """R-868 (v0.144.0): an apply pass keeps its report on disk until the agent has sent it.
    COMPANION RED-PROOF: drop the r.save_report call in main() -> test_apply_keeps_a_copy fails."""

    def test_apply_keeps_a_copy_with_the_agents_ids(self):
        f = Fake()
        f.plan.update(run_id="20261005T0257-ab12", trigger="debug", ring=0)
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(len(f.saved_reports), 1, "an apply pass must keep its report on disk")
        path, saved = f.saved_reports[0]
        self.assertEqual(path, PLAN)
        self.assertEqual(saved, rep, "the copy is the report the agent would have read")
        self.assertEqual((saved["run_id"], saved["trigger"], saved["ring"]), ("20261005T0257-ab12", "debug", 0))

    def test_a_refusal_is_kept_too(self):
        f = Fake()
        f.free = 1
        rc, rep = run(f)
        self.assertEqual(rc, 2)
        self.assertEqual(f.saved_reports[0][1]["refused"]["code"], "R8")

    def test_other_modes_keep_nothing(self):
        f = Fake()
        f.plan["mode"] = "health"
        run(f)
        self.assertEqual(f.saved_reports, [])

    def test_odd_ids_are_dropped_not_trusted(self):
        f = Fake()
        f.plan.update(run_id="../../etc/x", trigger="Night; rm", ring=True)
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        for k in ("run_id", "trigger", "ring"):
            self.assertNotIn(k, rep)


class SaveReportOnDisk(unittest.TestCase):
    """The real Runner.save_report on a temp dir: root writes into the AGENT's directory, so a symlink must never
    be followed — neither for the directory nor for the file name."""

    def setUp(self):
        import tempfile
        self.tmp = tempfile.mkdtemp()
        self.dir = os.path.join(self.tmp, "os")
        os.mkdir(self.dir)
        self.prev = osapply.PLAN_DIR
        osapply.PLAN_DIR = self.dir
        self.r = osapply.Runner()
        self.r.agent_uid = lambda: os.getuid()

    def tearDown(self):
        import shutil
        osapply.PLAN_DIR = self.prev
        shutil.rmtree(self.tmp)

    def test_writes_0600_next_to_the_plan(self):
        p = self.r.save_report(os.path.join(self.dir, "plan-r1-guest-apply.json"), {"mode": "apply"})
        self.assertEqual(p, os.path.join(self.dir, "report-r1-guest-apply.json"))
        st = os.stat(p)
        self.assertEqual(statmod.S_IMODE(st.st_mode), 0o600)
        with open(p) as fh:
            self.assertEqual(json.load(fh), {"mode": "apply"})

    def test_a_symlink_at_the_name_is_replaced_not_followed(self):
        victim = os.path.join(self.tmp, "victim")
        with open(victim, "w") as fh:
            fh.write("untouched")
        os.symlink(victim, os.path.join(self.dir, "report-r2-guest-apply.json"))
        self.r.save_report(os.path.join(self.dir, "plan-r2-guest-apply.json"), {"mode": "apply"})
        with open(victim) as fh:
            self.assertEqual(fh.read(), "untouched")
        self.assertFalse(os.path.islink(os.path.join(self.dir, "report-r2-guest-apply.json")))

    def test_a_symlinked_directory_is_refused(self):
        real = os.path.join(self.tmp, "elsewhere")
        os.mkdir(real)
        link = os.path.join(self.tmp, "linked")
        os.symlink(real, link)
        osapply.PLAN_DIR = link
        with self.assertRaises(OSError):
            self.r.save_report(os.path.join(link, "plan-r3-guest-apply.json"), {"mode": "apply"})
        self.assertEqual(os.listdir(real), [])


class AgentDiesMidPass(unittest.TestCase):
    """R-868, MEASURED LIVE 2026-10-05 05:45 UTC on demo-hp (agent v0.144.0): the agent was kill -9-ed while apt-get
    ran; apt finished all 13 packages, but the wrapper's next log line went to a stderr pipe nobody reads any more ->
    BrokenPipeError -> the wrapper died before save_report: no DONE in the journal, no kept copy, no report.
    COMPANION RED-PROOF: let Runner.log write to stderr unguarded -> this test fails (BrokenPipeError)."""

    def test_a_dead_reader_does_not_stop_the_report_copy(self):
        import io, sys, contextlib

        class DeadPipe(io.TextIOBase):
            dead = False
            def write(self, s):
                if DeadPipe.dead:
                    raise BrokenPipeError(32, "Broken pipe")
                return len(s)
            def flush(self):
                if DeadPipe.dead:
                    raise BrokenPipeError(32, "Broken pipe")

        class DyingFake(Fake):
            def emulate(self, argv):
                a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"]
                if a and a[0] == "apt-get" and "install" in a and "-s" not in a and "--print-uris" not in a:
                    DeadPipe.dead = True  # the agent is killed while apt-get runs
                return super().emulate(argv)

        f = DyingFake()
        journal = []
        f.log = lambda line: osapply.Runner.log(f, line)  # the REAL log(): stderr, then the journal
        prev_run = osapply.subprocess.run
        osapply.subprocess.run = lambda argv, *a, **kw: (journal.append(argv[-1]) if argv[0] == "logger"
                                                         else prev_run(argv, *a, **kw))  # never the real `logger`
        prev_err, prev_out = sys.stderr, sys.stdout
        sys.stderr, sys.stdout = DeadPipe(), DeadPipe()
        try:
            rc = osapply.main(["felhom-os-apply", "--plan", PLAN], runner=f)
        finally:
            sys.stderr, sys.stdout = prev_err, prev_out
            osapply.subprocess.run = prev_run
            DeadPipe.dead = False
        self.assertEqual(rc, 0)
        self.assertEqual(len(f.saved_reports), 1, "the kept copy is the only way this report reaches the hub")
        self.assertEqual(len(f.saved_reports[0][1]["upgraded"]), 2)
        self.assertTrue(any(l.startswith("os-apply: DONE") for l in journal), "the journal must still get the DONE line")


class CrashLeftTheJournal(unittest.TestCase):
    """R-876 — THE MEASURED SHAPE (demo-hp 2026-10-05, a crash while dpkg unpacked): `dpkg --audit` CLEAN, but
    /var/lib/dpkg/updates holds 3 files; apt refuses every install ("dpkg was interrupted"). v0.144.1 logged
    `REPAIR configured=0 fixed=0`, then `FAILED rc=100`, every pass, until a person ran `dpkg --configure -a`.
    COMPANION RED-PROOFS: drop `or journal` from repair()'s condition -> test 1 fails; drop the interrupted-retry
    -> test 3 fails; make repair() always run -> test 2 fails (R-845's speed)."""

    def test_the_next_pass_repairs_by_itself_and_finishes(self):
        f = Fake()
        f.dpkg_audit = ""
        f.dpkg_journal = ["0000", "0001", "0002"]
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["repair"]["journal_before"], 3)
        self.assertTrue(rep["repair"]["clean_after"])
        self.assertEqual(len(rep["upgraded"]), 2)
        cfg = next(i for i, c in enumerate(f.calls) if c[0] == "guest" and "--configure" in c[2])
        inst = next(i for i, c in enumerate(f.calls) if c[0] == "guest" and "install" in c[2] and "-s" not in c[2]
                    and "-f" not in c[2] and "--print-uris" not in c[2])
        self.assertLess(cfg, inst, "the repair must run before the install")
        self.assertFalse(any("INTERRUPTED" in l for l in f.logs), "the journal check must catch it BEFORE apt refuses")

    def test_a_clean_pass_still_costs_one_state_call(self):
        f = Fake()
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        state = [c for c in f.calls if c[0] == "guest" and c[2][:2] == ["sh", "-c"] and c[2][2] == osapply.DPKG_STATE_SCRIPT]
        self.assertEqual(len(state), 1, "R-845: a clean pass reads dpkg's state ONCE and runs no repair")
        self.assertEqual(getattr(f, "configured_calls", 0), 0)

    def test_apt_interrupted_is_repaired_and_retried_once(self):
        """The belt: the journal probe saw nothing (a race, an odd layout), apt still says interrupted."""
        f = Fake()
        orig = f.emulate
        state = {"first": True}

        def emulate(argv):
            a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"]
            if a[0] == "apt-get" and "install" in a and "-s" not in a and "-f" not in a and "--print-uris" not in a and state["first"]:
                state["first"] = False
                return 100, "", "E: dpkg was interrupted, you must manually run 'sudo dpkg --configure -a' to correct the problem.\n"
            return orig(argv)
        f.emulate = emulate
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertTrue(any("INTERRUPTED" in l for l in f.logs), f.logs)
        self.assertTrue(any(l.startswith("os-apply: REPAIR ") and l.endswith("forced") for l in f.logs), f.logs)


class OOMCheck(unittest.TestCase):
    """R-528 (`09` decision 157): after a Docker engine step the wrapper proves the engine reports a memory kill
    (OOMKilled=true AND the `oom` event). Reported only; the hub decides. Red-proofs: audits/readback-2026-10-07/F/."""

    def apply(self, **kw):
        f = docker_fake(signed=signed_job())
        for k, v in kw.items():
            setattr(f, k, v)
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        return f, rep

    def test_pass_when_oomkilled_and_the_event(self):
        f, rep = self.apply()
        oc = rep["oom_check"]
        self.assertEqual(oc["result"], "pass", oc)
        self.assertEqual((oc["oom_killed"], oc["oom_event"], oc["exit_code"]), (True, True, 137))
        self.assertEqual(oc["image"], "gitea.dooplex.hu/admin/felhom-controller:0.300.0")
        self.assertEqual(sorted(oc), ["detail", "exit_code", "image", "oom_event", "oom_killed", "result"])
        run_argv = f.oom_runs[0]
        name = run_argv[run_argv.index("--name") + 1]
        self.assertTrue(re.match(r"^felhom-oomcheck-[0-9]+-[0-9a-f]{8}$", name), name)
        for flag, val in (("--pull", "never"), ("--network", "none"), ("--memory", "64m"), ("--memory-swap", "64m"),
                          ("--label", "felhom.oomcheck=1"), ("--entrypoint", "sh")):
            self.assertEqual(run_argv[run_argv.index(flag) + 1], val, flag)
        self.assertNotIn("-v", run_argv)
        self.assertNotIn("-p", run_argv)
        self.assertEqual(run_argv[-3:], ["gitea.dooplex.hu/admin/felhom-controller:0.300.0", "-c", osapply.OOMCHECK_SCRIPT])
        self.assertIn(f"container={name}", f.oom_events_argv)
        self.assertIn("event=oom", f.oom_events_argv)
        self.assertEqual(f.oom_removed, [name], "the check container must be removed")
        self.assertTrue(rep["health_after"], "health is read before the check")
        # bounded: every call has a timeout and the clock is read twice — the worst case stays within 90 s
        self.assertLessEqual(sum(osapply.OOMCHECK_TIMEOUTS.values()) + osapply.OOMCHECK_TIMEOUTS["clock"]
                             + osapply.OOMCHECK_SETTLE, 90)

    def test_events_window_ends_after_the_settle_wait(self):
        # Measured (F1/F2): an --until taken right after the run missed the oom event. The window must end after a
        # wait of >= 2 s that comes AFTER the run, at the guest epoch read after that wait, + 1.
        f, rep = self.apply()
        run_i = next(i for i, c in enumerate(f.calls) if c[-1][:2] == ["docker", "run"])
        date_i = [i for i, c in enumerate(f.calls) if c[-1] == ["date", "+%s"]]
        waits = [(i, s) for i, s in getattr(f, "sleeps", []) if i > run_i]
        self.assertTrue(waits and waits[0][1] >= 2, f"no settle wait after the run: {getattr(f, 'sleeps', None)}")
        self.assertTrue(date_i[-1] >= waits[0][0], "the end epoch must be read after the wait")
        ev = f.oom_events_argv
        since, until = int(ev[ev.index("--since") + 1]), int(ev[ev.index("--until") + 1])
        self.assertEqual(until, f.oom_epochs[-1] + 1, "until = the guest epoch read after the wait, + 1")
        self.assertGreater(until, f.oom_epochs[0] + 1)
        self.assertLess(since, f.oom_epochs[0] + 1)

    def test_oomkilled_false_is_fail(self):
        f, rep = self.apply(oom_state="false 0")
        oc = rep["oom_check"]
        self.assertEqual(oc["result"], "fail", oc)
        self.assertIn("OOMKilled=true", oc["detail"])
        self.assertTrue(oc["oom_event"])
        self.assertEqual(len(f.oom_removed), 1)

    def test_no_event_is_fail(self):
        f, rep = self.apply(oom_event=False)
        oc = rep["oom_check"]
        self.assertEqual(oc["result"], "fail", oc)
        self.assertIn("the oom event", oc["detail"])
        self.assertTrue(oc["oom_killed"])

    def test_image_unreadable_is_error(self):
        f, rep = self.apply(oom_image=None)
        oc = rep["oom_check"]
        self.assertEqual(oc["result"], "error", oc)
        self.assertIsNone(oc["image"])
        self.assertIn("image could not be read", oc["detail"])
        self.assertFalse(hasattr(f, "oom_runs"), "no container is started without an image")

    def test_container_removed_even_when_inspect_raises(self):
        f, rep = self.apply(oom_inspect_raises=True)
        oc = rep["oom_check"]
        self.assertEqual(oc["result"], "error", oc)
        self.assertEqual(len(f.oom_removed), 1, "the container must be removed even when inspect raised")
        self.assertTrue(f.oom_removed[0].startswith(osapply.OOMCHECK_PREFIX))
        self.assertNotIn("failed", rep, "the check never turns the step into a failure")

    def test_never_on_guest_or_host_or_in_health_mode(self):
        g = Fake()
        rc, rep = run(g)
        self.assertEqual(rc, 0, rep)
        h = Fake()
        h.plan["layer"] = "host"
        h.plan["packages"] = [{"name": "bash", "version": "5.2.37-2+b10", "origin": "Debian"}]
        h.installed["bash"] = "5.2.37-2+b9"
        h.live["bash"] = {"5.2.37-2+b10"}
        rc_h, rep_h = run(h)
        self.assertEqual(rc_h, 0, rep_h)
        d = docker_fake(signed=signed_job())
        d.plan["mode"] = "health"
        rc_d, rep_d = run(d)
        self.assertEqual(rc_d, 0, rep_d)
        for f, r in ((g, rep), (h, rep_h), (d, rep_d)):
            self.assertNotIn("oom_check", r)
            self.assertFalse(hasattr(f, "oom_runs"), r.get("layer"))
            self.assertFalse(any(c[-1][:2] == ["docker", "run"] for c in f.calls))

    def test_mode_oom_check_runs_only_the_check(self):
        f = docker_fake()  # no authority: the check changes nothing, so it needs none
        f.plan["mode"], f.plan["packages"] = "oom-check", []
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["oom_check"]["result"], "pass", rep)
        self.assertFalse(any("apt-get" in c[-1] or "dpkg-query" in c[-1] for c in f.calls), f.calls)
        self.assertEqual(len(f.oom_removed), 1)

    def test_mode_oom_check_keeps_the_refusals(self):
        f = docker_fake()
        f.plan["mode"], f.plan["packages"] = "oom-check", []
        f.files["/etc/pve/lxc/9201.conf"] = "arch: amd64\n"
        rc, rep = run(f)
        self.assertEqual((rc, rep["refused"]["code"]), (2, "R10"), rep)
        self.assertFalse(hasattr(f, "oom_runs"))
        g = Fake()
        g.plan["mode"] = "oom-check"
        rc, rep = run(g)
        self.assertEqual((rc, rep["refused"]["code"]), (2, "R11"), rep)


PVE = "Proxmox Debian Repository:stable"
PVE_SET = [{"name": "pve-manager", "version": "9.2.21", "origin": "Proxmox Debian Repository"},
           {"name": "libpve-common-perl", "version": "9.1.9", "origin": "Proxmox Debian Repository"}]


def pve_fake(signed=None, ring0=False):
    f = Fake()
    f.installed.update({"pve-manager": "9.2.2", "libpve-common-perl": "9.1.1", "proxmox-kernel-helper": "9.0.4"})
    f.live["pve-manager"] = {"9.2.21", "9.2.2"}
    f.live["libpve-common-perl"] = {"9.1.9", "9.1.1"}
    f.origins = {"pve-manager": PVE, "libpve-common-perl": PVE, "proxmox-kernel-helper": PVE, "shim-signed": PVE,
                 "proxmox-firewall-data": PVE}
    f.plan = {"release_id": "os-pve-t1", "layer": "pve", "lane": "slow", "vmid": 9201, "mode": "apply",
              "packages": [dict(p) for p in PVE_SET]}
    if ring0:
        f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": True})
    if signed is not None:
        f.plan["signed"] = signed
    return f


class PVELane(unittest.TestCase):
    """R-812 option A (`09` §3 decision 163): the host's Proxmox USERSPACE packages, slow lane, no kernel / boot /
    firmware / microcode (R14), no removal, a new package only from PVE_NEW_ALLOW. Red-proof: audits/day-2026-10-07/B/."""

    def refused(self, f, code):
        rc, rep = run(f)
        self.assertEqual(rc, 2, rep)
        self.assertEqual(rep["refused"]["code"], code, rep)
        self.assertEqual(f.installed["pve-manager"], "9.2.2", "nothing may be installed on a refusal")
        return rep

    # THE RED TEST (design-R-812 §5): before the pve layer existed this plan was refused R12.
    def test_pve_manager_plan_is_installed_on_the_host(self):
        f = pve_fake(signed=signed_job(packages=PVE_SET, op="os_pve_step"))
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(f.installed["pve-manager"], "9.2.21")
        self.assertEqual(rep["authority"], "signed")
        self.assertEqual(rep["pve_manager"], "9.2.21", "pveversion after the step is reported")
        inst = [c for c in f.calls if c[0] == "host" and "install" in c[1] and "-s" not in c[1] and "-f" not in c[1]
                and "--print-uris" not in c[1]]
        self.assertTrue(inst, "the pve layer installs on the HOST")
        self.assertFalse([c for c in f.calls if c[0] == "guest" and "install" in c[2]], "nothing installed in the guest")
        self.assertEqual(sorted(rep["health_after"]["host_services"]), sorted(osapply.HOST_SERVICES))

    def test_kernel_in_a_pve_plan_is_refused(self):
        f = pve_fake(ring0=True)
        f.plan["packages"].append({"name": "proxmox-kernel-7.0", "version": "7.0.14-20", "origin": "Proxmox Debian Repository"})
        self.refused(f, "R14")

    def test_shim_in_a_pve_plan_is_refused(self):
        f = pve_fake(ring0=True)
        f.plan["packages"].append({"name": "shim-signed", "version": "1.47+pmx1", "origin": "Proxmox Debian Repository"})
        self.refused(f, "R14")

    def test_kernel_pulled_in_by_the_simulation_is_refused(self):
        f = pve_fake(ring0=True)
        f.installed["proxmox-kernel-helper"] = "9.0.4"
        f.extra_sim = ["Inst proxmox-kernel-helper [9.0.4] (9.0.6 Proxmox Debian Repository:stable [all])"]
        self.refused(f, "R6")  # not in the plan — refused before the name check; R14 below when it IS in the plan
        g = pve_fake(ring0=True)
        g.live["proxmox-kernel-helper"] = {"9.0.6"}
        g.plan["packages"] = [dict(PVE_SET[0])]
        g.extra_sim = ["Inst proxmox-kernel-helper [9.0.4] (9.0.6 Proxmox Debian Repository:stable [all])"]
        # a kernel-helper the plan did not name is R6; the R14 name check covers a listed one (test above)
        self.refused(g, "R6")

    def test_debian_package_in_a_pve_plan_is_refused(self):
        f = pve_fake(ring0=True)
        f.plan["packages"].append({"name": "libc6", "version": "2.41-12+deb13u4", "origin": "Debian"})
        self.refused(f, "R2")

    def test_debian_origin_in_the_pve_simulation_is_refused(self):
        f = pve_fake(ring0=True)
        f.origins["libpve-common-perl"] = DEB
        self.refused(f, "R2")

    def test_docker_package_in_a_pve_plan_is_refused(self):
        f = pve_fake(ring0=True)
        f.plan["packages"].append({"name": "docker-ce", "version": "5:29.8.2-1~debian.13~trixie", "origin": "Proxmox Debian Repository"})
        self.refused(f, "R2")

    def test_pve_in_the_fast_lane_is_refused(self):
        f = pve_fake(ring0=True)
        f.plan["lane"] = "fast"
        self.refused(f, "R3")

    def test_no_authority_is_refused(self):
        self.refused(pve_fake(), "R3")

    def test_docker_signed_op_does_not_authorize_a_pve_step(self):
        self.refused(pve_fake(signed=signed_job(packages=PVE_SET, op="os_docker_step")), "R3")

    def test_pve_on_a_byo_box_is_refused(self):
        f = pve_fake(ring0=True)
        f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"})
        self.refused(f, "R12")

    def test_unlisted_new_package_is_refused(self):
        f = pve_fake(ring0=True)
        f.extra_sim = ["Inst proxmox-new-thing (1.0 Proxmox Debian Repository:stable [all])"]
        self.refused(f, "R6")

    def test_allow_listed_new_package_is_accepted(self):
        f = pve_fake(ring0=True)
        f.extra_sim = ["Inst proxmox-firewall-data (0.1 Proxmox Debian Repository:stable [all])"]
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(f.installed["pve-manager"], "9.2.21")

    def test_allow_new_is_never_an_open_door_in_the_fast_lane(self):
        f = Fake()
        f.plan["layer"] = "host"
        f.extra_sim = ["Inst proxmox-firewall-data (0.1 Proxmox Debian Repository:stable [all])"]
        rc, rep = run(f)
        self.assertEqual((rc, rep["refused"]["code"]), (2, "R6"), rep)

    def test_undo_is_refused(self):
        f = pve_fake(signed=signed_job(packages=PVE_SET, op="os_pve_step"))
        f.plan["undo"] = True
        self.refused(f, "R5")

    def test_ring0_pending_pve_takes_only_installed_proxmox_userspace(self):
        f = pve_fake(ring0=True)
        f.plan["select"], f.plan["packages"] = "pending-pve", []
        f.installed.update({"linux-image-amd64": "6.12.1", "tailscale": "1.102.2"})
        f.pending_sim = [
            "Inst pve-manager [9.2.2] (9.2.21 Proxmox Debian Repository:stable [amd64])",
            "Inst libpve-common-perl [9.1.1] (9.1.9 Proxmox Debian Repository:stable [all])",
            "Inst proxmox-kernel-helper [9.0.4] (9.0.6 Proxmox Debian Repository:stable [all])",
            "Inst proxmox-kernel-7.0.14-20-pve-signed (7.0.14-20 Proxmox Debian Repository:stable [amd64])",
            "Inst proxmox-firewall-data (0.1 Proxmox Debian Repository:stable [all])",
            "Inst libc6 [2.41-12+deb13u3] (2.41-12+deb13u4 Debian:13.7/stable [amd64])",
            "Inst tailscale [1.102.2] (1.102.5 Tailscale:pkgs.tailscale.com [amd64])",
        ]
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["authority"], "ring0")
        self.assertEqual(sorted(u["name"] for u in rep["upgraded"]), ["libpve-common-perl", "pve-manager"],
                         "pending-pve: installed, Proxmox-origin, never kernel/boot/firmware, never Debian or other origins")
        self.assertEqual(f.installed["libc6"], "2.41-12+deb13u3")

    def test_select_pending_pve_needs_the_pve_layer(self):
        f = Fake()
        f.plan["layer"], f.plan["select"], f.plan["packages"] = "host", "pending-pve", []
        rc, rep = run(f)
        self.assertEqual((rc, rep["refused"]["code"]), (2, "R11"), rep)


# ---------- the kernel lane (R-836, `09` §3 decision 172, `11` §5.11) ----------
U = "1af1fcc6-639c-416b-a7e5-c4470d41a502"
OLD, NEW = "7.0.2-6-pve", "7.0.14-22-pve"
KERNEL_SET = [{"name": "proxmox-kernel-7.0", "version": "7.0.14-22", "origin": "Proxmox Debian Repository"},
              {"name": "proxmox-kernel-7.0.14-22-pve-signed", "version": "7.0.14-22", "origin": "Proxmox Debian Repository"}]


class KFake(Fake):
    """A Proxmox host with GRUB on UEFI: /boot on the root LV, a vfat ESP, the bundle's two generators. update-grub is
    emulated like the real 10_linux: WITHOUT the felhom default file the NEWEST kernel becomes the default (measured,
    R-836 — that is the defect the lane exists for); with it, the kernel it names."""

    def __init__(self):
        super().__init__()
        self.running = OLD
        self.boot = {OLD}
        self.efi, self.esp_fs, self.boot_mount, self.mods, self.snippets, self.pin = True, "vfat", "", True, True, False
        self.env = None          # None = no env block on the ESP; else {"felhom_next": ...}
        self.tree = {}           # files put_file wrote
        self.reboots = []
        self.update_grubs = 0
        self.grub_runs_in_postinst = True
        self.installed.update({"proxmox-kernel-7.0": "7.0.2-6", "proxmox-default-kernel": "2.1.0",
                               "pve-firmware": "3.18-3", "proxmox-kernel-7.0.2-6-pve-signed": "7.0.2-6",
                               "pve-manager": "9.2.21"})
        self.live.update({"proxmox-kernel-7.0": {"7.0.14-22", "7.0.2-6"},
                          "proxmox-kernel-7.0.14-22-pve-signed": {"7.0.14-22"},
                          "proxmox-kernel-7.0.14-23-pve-signed": {"7.0.14-23"},
                          "pve-firmware": {"3.18-7", "3.18-3"}})
        self.origins = {}
        self.kernel_pending = ["Inst pve-firmware [3.18-3] (3.18-7 Proxmox Debian Repository:stable [all])",
                               "Inst proxmox-kernel-7.0.14-22-pve-signed (7.0.14-22 Proxmox Debian Repository:stable [amd64])",
                               "Inst proxmox-kernel-7.0 [7.0.2-6] (7.0.14-22 Proxmox Debian Repository:stable [amd64])",
                               "Inst pve-manager [9.2.21] (9.2.22 Proxmox Debian Repository:stable [amd64])",
                               "Inst libc6 [2.41-12+deb13u3] (2.41-12+deb13u4 Debian:13.7/stable [amd64])"]
        self.plan = {"release_id": "kernel-t1", "layer": "kernel", "lane": "slow", "vmid": 9201, "mode": "apply",
                     "select": "pending-kernel", "packages": []}
        self.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": True})
        self.files[osapply.CRASH_GUARD_STATE] = json.dumps({"armed": True, "tripped": False, "unclean_boots_in_window": 0})
        self.cfg = self.render()

    # -- GRUB --
    def newest(self):
        best = None
        for k in self.boot:
            if best is None or dpkg_cmp(k[:-4], "gt", best[:-4]):
                best = k
        return best

    def render(self):
        d = self.tree.get(osapply.KERNEL_DEFAULT_CFG)
        if d:
            dflt = re.search(r'GRUB_DEFAULT="([^"]+)"', d).group(1)
        else:
            dflt = f"gnulinux-advanced-{U}>gnulinux-{self.newest()}-advanced-{U}"
        cfg = ('if [ "${next_entry}" ] ; then\n   set default="${next_entry}"\nelse\n'
               f'   set default="{dflt}"\nfi\n'
               f"submenu 'Advanced options' $menuentry_id_option 'gnulinux-advanced-{U}' {{\n")
        for k in sorted(self.boot):
            cfg += f"  menuentry 'Proxmox VE, with Linux {k}' $menuentry_id_option 'gnulinux-{k}-advanced-{U}' {{ }}\n"
        cfg += "}\n"
        if self.snippets:
            cfg += "### BEGIN /etc/grub.d/01_felhom_oneshot ###\nload_env felhom_next\n"
            cfg += "".join(f"menuentry 'Felhom one-shot: {k}' --id felhom-oneshot-{k} {{ }}\n" for k in sorted(self.boot))
        return cfg

    def lexists(self, p):
        if p == "/sys/firmware/efi":
            return self.efi
        if p.startswith("/usr/lib/grub/x86_64-efi/"):
            return self.mods
        if p in (osapply.ONESHOT_SNIPPET, osapply.ONESHOT_ENTRIES):
            return self.snippets
        if p == osapply.KERNEL_PIN_FILE:
            return self.pin
        m = re.match(r"^/boot/(vmlinuz|initrd\.img)-(.+)$", p)
        if m:
            return m.group(2) in self.boot
        return p in self.tree

    def put_file(self, path, data, mode):
        self.tree[path] = data.decode()

    def read_file(self, p):
        if p == osapply.GRUB_CFG:
            return self.cfg
        if p in self.tree:
            return self.tree[p]
        return super().read_file(p)

    def host(self, argv, timeout=600, stdin=None):
        if argv[0] == "uname":
            self.calls.append(("host", argv))
            return 0, self.running + "\n", ""
        if argv[0] == "findmnt":
            self.calls.append(("host", argv))
            if argv[-1] == osapply.ESP_MOUNT:
                return (0, self.esp_fs + "\n", "") if self.esp_fs else (1, "", "")
            return (0, self.boot_mount + "\n", "") if self.boot_mount else (1, "", "")
        if argv[0] == "grub-editenv":
            self.calls.append(("host", argv))
            if argv[2] == "list":
                return (1, "", "no such file") if self.env is None else \
                    (0, "".join(f"{k}={v}\n" for k, v in self.env.items()), "")
            if argv[2] == "create":
                self.env = {}
                return 0, "", ""
            if argv[2] == "set":
                k, v = argv[3].split("=", 1)
                self.env[k] = v
                return 0, "", ""
            if argv[2] == "unset":
                self.env.pop(argv[3], None)
                return 0, "", ""
        if argv[0] == "update-grub":
            self.calls.append(("host", argv))
            self.update_grubs += 1
            self.cfg = self.render()
            return 0, "", ""
        if argv[0] == "mkdir":
            self.calls.append(("host", argv))
            return 0, "", ""
        if argv[:2] == ["systemctl", "reboot"]:
            self.calls.append(("host", argv))
            self.reboots.append(self.running)
            return 0, "", ""
        return super().host(argv, timeout, stdin)

    def emulate(self, argv):
        a = [x for x in argv if not re.match(r"^[A-Z_]+=", x) and x != "env"]
        if a[0] == "apt-get" and "install" in a and "-s" not in a and "--print-uris" not in a and "-f" not in a:
            if self.install_rc:
                return self.install_rc, "", "E: boom"
            for x in a:
                if "=" in x and not x.startswith("-") and "::" not in x:
                    n, v = x.split("=", 1)
                    self.installed[n] = v
                    m = re.match(r"^proxmox-kernel-[0-9]+\.[0-9]+$", n)
                    if m:
                        self.installed[f"proxmox-kernel-{v}-pve-signed"] = v
                    if m or osapply.KERNEL_IMAGE_RE.match(n):
                        self.boot.add(v + "-pve")
            if self.grub_runs_in_postinst:
                self.cfg = self.render()   # the kernel's postinst runs update-grub (zz-update-grub)
            return 0, "Setting up proxmox-kernel ...\n", ""
        return super().emulate(argv)

    def sim(self, a):
        if "--print-uris" in a:
            return 0, "", ""
        if "dist-upgrade" in a:
            return 0, "\n".join(self.kernel_pending) + "\n", ""
        out, named = "", set()
        for x in a:
            if "=" in x and not x.startswith("-") and "::" not in x:
                named.add(x.split("=", 1)[0])
        for x in a:
            if "=" in x and not x.startswith("-") and "::" not in x:
                n, v = x.split("=", 1)
                if v not in self.avail(n):
                    return 100, "", f"E: Version '{v}' for '{n}' was not found"
                o = self.origins.get(n, PVE)
                out += f"Inst {n} [{self.installed[n]}] ({v} {o} [amd64])\n" if n in self.installed else f"Inst {n} ({v} {o} [amd64])\n"
                if re.match(r"^proxmox-kernel-[0-9]+\.[0-9]+$", n):
                    img = f"proxmox-kernel-{v}-pve-signed"
                    if img not in self.installed and img not in named:
                        out += f"Inst {img} ({v} {PVE} [amd64])\n"
        out += "".join(l + "\n" for l in self.extra_sim)
        return 0, out, ""


def kfake(**kw):
    f = KFake()
    for k, v in kw.items():
        setattr(f, k, v)
    return f


def kmode(f, mode, **extra):
    f.plan = {"release_id": "kernel-t1", "layer": "kernel", "lane": "slow", "vmid": 9201, "mode": mode, "packages": []}
    f.plan.update(extra)
    return run(f)


def staged(f=None):
    f = f or kfake()
    rc, rep = run(f)
    assert rc == 0, rep
    return f


class KernelLane(unittest.TestCase):
    """R-836 / `09` §3 decision 172: stage a kernel once through the ESP flag; the default moves only after a healthy
    one-shot boot. Red-proof: audits/kernel-lane-2026-10-07/A/redproof.txt."""

    def refused(self, f, code, mode=None, **extra):
        rc, rep = kmode(f, mode, **extra) if mode else run(f)
        self.assertEqual(rc, 2, rep)
        self.assertEqual(rep["refused"]["code"], code, rep)
        return rep

    # --- the four red tests the brief names (Part A 3) ---
    def test_installing_a_kernel_does_not_change_the_grub_default(self):
        f = kfake()
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertIn(NEW, f.boot, "the new kernel was installed")
        self.assertEqual(osapply.Kernel.default_kver(f.cfg), OLD,
                         "the default must stay the kernel the box runs (R-836: an install made the new one default)")
        self.assertIn(f"gnulinux-{OLD}-advanced-{U}", f.tree[osapply.KERNEL_DEFAULT_CFG])
        self.assertEqual(f.env, {"felhom_next": NEW}, "the ONLY way to the new kernel is the one-shot flag")
        self.assertEqual(f.reboots, [], "staging never reboots")
        st = json.loads(f.tree[osapply.KERNEL_STATE])
        self.assertEqual((st["phase"], st["from"], st["to"]), ("staged", OLD, NEW))
        self.assertEqual(rep["kernel"]["default"], OLD)

    def test_a_box_without_the_esp_flag_is_refused(self):
        for attr, val, frag in (("esp_fs", "ext4", "vfat"), ("efi", False, "UEFI"), ("snippets", False, "bundle"),
                                ("mods", False, "module"), ("boot_mount", "/boot", "separate"), ("pin", True, "pinned")):
            f = kfake(**{attr: val})
            rep = self.refused(f, "R20")
            self.assertIn(frag, rep["refused"]["reason"], attr)
            self.assertNotIn(NEW, f.boot, f"{attr}: nothing may be installed on a refusal")
            self.assertIsNone(f.env, f"{attr}: no flag on a refusal")

    def test_a_reboot_without_the_flag_is_refused(self):
        f = staged()
        f.env = {"felhom_next": ""}
        self.refused(f, "R22", mode="kernel-reboot")
        self.assertEqual(f.reboots, [])

    def test_a_kernel_outside_the_approved_set_is_refused(self):
        # a signed set that names only the meta-package: the image the sources pull in was never signed for
        f = kfake()
        f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False})
        f.plan.update(select="listed", packages=[dict(KERNEL_SET[0])],
                      signed=signed_job(packages=[KERNEL_SET[0]], op="os_kernel_step"))
        self.refused(f, "R23")
        self.assertNotIn(NEW, f.boot)
        # a signed set names 7.0.14-22; the sources would ALSO bring 7.0.14-23
        f2 = kfake()
        f2.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False})
        f2.plan.update(select="listed", packages=[dict(p) for p in KERNEL_SET],
                       signed=signed_job(packages=KERNEL_SET, op="os_kernel_step"))
        f2.extra_sim = ["Inst proxmox-kernel-7.0.14-23-pve-signed (7.0.14-23 Proxmox Debian Repository:stable [amd64])"]
        self.refused(f2, "R23")
        # a name that is not a kernel package at all
        g = kfake()
        g.plan.update(select="listed", packages=[{"name": "pve-manager", "version": "9.2.22", "origin": "Proxmox Debian Repository"}])
        self.refused(g, "R23")
        # the household was told about another kernel
        h = kfake()
        h.plan["expect_kver"] = "7.0.14-23-pve"
        self.refused(h, "R23")
        self.assertNotIn(NEW, h.boot)

    def test_the_snippet_clears_the_flag_on_use(self):
        """01_felhom_oneshot: the flag is copied, CLEARED and SAVED before any `set default`, all inside the one branch
        that runs only when the flag is set; a default is set only for an installed kernel's own entry."""
        text = (HERE / "felhom-grub-oneshot.sh").read_text()
        body = text[text.index("cat <<EOF"):]
        i_copy = body.index('set felhom_boot="\\${felhom_next}"')
        i_clear = body.index("set felhom_next=\n")
        i_save = body.index("save_env -f (\\$felhom_esp)/EFI/felhom/oneshot.env felhom_next")
        i_default = body.index('set default="felhom-oneshot-%s"')
        self.assertLess(i_copy, i_clear)
        self.assertLess(i_clear, i_save)
        self.assertLess(i_save, i_default, "the flag must be cleared on disk BEFORE GRUB boots anything")
        self.assertIn('if [ "${felhom_boot}" = "%s" ]', body, "a default only for a kernel that is installed")
        self.assertIn(osapply.ONESHOT_ENV[len(osapply.ESP_MOUNT):], text, "the wrapper and GRUB name the same env file")

    # --- the rest of the stage ---
    def test_option_c_options_are_on_the_one_shot_entry_only(self):
        text = (HERE / "felhom-grub-oneshot-entries.sh").read_text()
        self.assertIn(osapply.ONESHOT_ARGS, text)
        self.assertIn("--id felhom-oneshot-$k", text)
        self.assertNotIn("set default", text, "the entries file never sets the default")

    def test_both_generators_ride_the_bundle(self):
        self.assertEqual(osapply.BUNDLE_DESTS[osapply.ONESHOT_SNIPPET][1:4], ("felhom-grub-oneshot.sh", 0o755, "sh"))
        self.assertEqual(osapply.BUNDLE_DESTS[osapply.ONESHOT_ENTRIES][1:4], ("felhom-grub-oneshot-entries.sh", 0o755, "sh"))

    def test_ring0_pending_kernel_takes_only_the_kernel_set(self):
        f = kfake()
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["authority"], "ring0")
        self.assertEqual(sorted(u["name"] for u in rep["upgraded"]), ["proxmox-kernel-7.0", "pve-firmware"])
        self.assertEqual(f.installed["pve-manager"], "9.2.21", "a Proxmox userspace package is the pve lane's")
        self.assertEqual(f.installed["libc6"], "2.41-12+deb13u3", "a Debian package is the fast lane's")

    def test_signed_listed_set_is_installed(self):
        f = kfake()
        f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False})
        f.plan.update(select="listed", packages=[dict(p) for p in KERNEL_SET],
                      signed=signed_job(packages=KERNEL_SET, op="os_kernel_step"))
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["authority"], "signed")
        self.assertEqual(f.env, {"felhom_next": NEW})

    def test_no_authority_and_the_wrong_op_are_refused(self):
        f = kfake()
        f.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False})
        self.refused(f, "R3")
        g = kfake()
        g.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False})
        g.plan.update(select="listed", packages=[dict(p) for p in KERNEL_SET],
                      signed=signed_job(packages=KERNEL_SET, op="os_pve_step"))
        self.refused(g, "R3")

    def test_fast_lane_byo_and_wrong_select_are_refused(self):
        f = kfake()
        f.plan["lane"] = "fast"
        self.refused(f, "R3")
        g = kfake()
        g.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"})
        self.refused(g, "R12")
        h = kfake()
        h.plan["select"] = "pending-pve"
        self.refused(h, "R11")
        k = kfake()
        k.plan.update(layer="host", lane="fast", mode="kernel-status")
        self.refused(k, "R11")

    def test_the_crash_guard_must_be_armed_and_quiet(self):
        f = kfake()
        f.files[osapply.CRASH_GUARD_STATE] = json.dumps({"armed": False, "tripped": True, "unclean_boots_in_window": 2})
        self.refused(f, "R21")
        g = kfake()
        g.files[osapply.CRASH_GUARD_STATE] = json.dumps({"armed": True, "tripped": False, "unclean_boots_in_window": 1})
        self.refused(g, "R21")
        h = kfake()
        del h.files[osapply.CRASH_GUARD_STATE]
        self.refused(h, "R21")
        self.assertNotIn(NEW, h.boot)

    def test_never_two_steps_in_one_night(self):
        f = staged()
        self.refused(f, "R22")  # still staged
        g = staged()
        st = json.loads(g.tree[osapply.KERNEL_STATE])
        st["phase"] = "good"
        g.tree[osapply.KERNEL_STATE] = json.dumps(st)
        g.clock += 3600
        self.refused(g, "R22")  # done, but within 20 h

    def test_removal_new_non_kernel_two_kernels_and_older_are_refused(self):
        f = kfake(extra_sim=["Remv pve-firmware [3.18-3]"])
        self.refused(f, "R4")
        g = kfake(extra_sim=["Inst proxmox-new-thing (1.0 Proxmox Debian Repository:stable [all])"])
        self.refused(g, "R6")
        h = kfake(extra_sim=["Inst proxmox-kernel-7.0.14-23-pve-signed (7.0.14-23 Proxmox Debian Repository:stable [amd64])"])
        self.refused(h, "R23")
        k = kfake(running="7.0.14-23-pve")
        k.boot = {"7.0.14-23-pve"}
        k.cfg = k.render()
        self.refused(k, "R23")
        m = kfake()
        m.origins = {"proxmox-kernel-7.0": DEB}
        self.refused(m, "R2")

    # R-898: ring 0 stages EXACTLY the told kernel (select listed, the set derived from it) — even when the sources
    # offer a newer one by night; a told version that can no longer be installed is refused BEFORE any change (R7).
    def test_ring0_listed_installs_the_told_kernel_not_the_newest(self):
        f = kfake()
        f.live["proxmox-kernel-7.0"] = {"7.0.14-20", "7.0.14-22", "7.0.2-6"}
        f.live["proxmox-kernel-7.0.14-20-pve-signed"] = {"7.0.14-20"}
        told = [{"name": "proxmox-kernel-7.0", "version": "7.0.14-20", "origin": "Proxmox Debian Repository"},
                {"name": "proxmox-kernel-7.0.14-20-pve-signed", "version": "7.0.14-20", "origin": "Proxmox Debian Repository"}]
        f.plan.update(select="listed", packages=told, expect_kver="7.0.14-20-pve")
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["authority"], "ring0")
        self.assertEqual(f.env, {"felhom_next": "7.0.14-20-pve"})
        self.assertEqual(f.installed["proxmox-kernel-7.0"], "7.0.14-20")
        self.assertNotIn("7.0.14-22-pve", f.boot)

    def test_ring0_told_kernel_gone_is_refused_before_any_change(self):
        f = kfake()
        f.live["proxmox-kernel-7.0"] = {"7.0.14-22"}   # 7.0.14-20 is no longer in the archive
        told = [{"name": "proxmox-kernel-7.0", "version": "7.0.14-20", "origin": "Proxmox Debian Repository"},
                {"name": "proxmox-kernel-7.0.14-20-pve-signed", "version": "7.0.14-20", "origin": "Proxmox Debian Repository"}]
        f.plan.update(select="listed", packages=told, expect_kver="7.0.14-20-pve")
        rep = self.refused(f, "R7")
        self.assertIsNone(f.env)
        self.assertNotIn(osapply.KERNEL_DEFAULT_CFG, f.tree, "refused before the default was even pinned")
        self.assertEqual(f.installed["proxmox-kernel-7.0"], "7.0.2-6")

    def test_nothing_pending_changes_nothing(self):
        f = kfake(kernel_pending=[])
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertEqual(rep["upgraded"], [])
        self.assertIsNone(f.env)
        self.assertNotIn(osapply.KERNEL_STATE, f.tree)

    def test_a_failed_install_writes_no_flag(self):
        f = kfake(install_rc=100)
        rc, rep = run(f)
        self.assertEqual(rc, 3, rep)
        self.assertIsNone(f.env)
        self.assertNotIn(osapply.KERNEL_STATE, f.tree)
        self.assertEqual(osapply.Kernel.default_kver(f.cfg), OLD)

    def test_a_postinst_without_update_grub_is_regenerated_and_proved(self):
        f = kfake(grub_runs_in_postinst=False)
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        self.assertIn(f"felhom-oneshot-{NEW}", f.cfg)
        self.assertEqual(osapply.Kernel.default_kver(f.cfg), OLD)

    # --- the reboot, the boot, good / revert ---
    def test_reboot_of_a_staged_step(self):
        f = staged()
        rc, rep = kmode(f, "kernel-reboot")
        self.assertEqual(rc, 0, rep)
        self.assertEqual(f.reboots, [OLD])
        st = json.loads(f.tree[osapply.KERNEL_STATE])
        self.assertEqual(st["phase"], "oneshot")
        self.assertIn("host_services", st["health_before"])

    def test_reboot_needs_a_staged_step(self):
        self.refused(kfake(), "R22", mode="kernel-reboot")

    def boot_into(self, f, kver):
        f.running = kver
        if f.env and f.env.get("felhom_next"):
            f.env["felhom_next"] = ""   # GRUB cleared it (01_felhom_oneshot)
        return kmode(f, "kernel-boot")

    def test_healthy_one_shot_becomes_the_default(self):
        f = staged()
        kmode(f, "kernel-reboot")
        rc, rep = self.boot_into(f, NEW)
        self.assertEqual((rc, rep["kernel_event"], rep["kernel"]["phase"]), (0, "judging", "judging"), rep)
        self.assertEqual(osapply.Kernel.default_kver(f.cfg), OLD, "judging does not move the default")
        rc, rep = kmode(f, "kernel-good")
        self.assertEqual(rc, 0, rep)
        self.assertEqual(osapply.Kernel.default_kver(f.cfg), NEW)
        self.assertEqual(rep["kernel"]["phase"], "good")

    def test_a_panic_falls_back_and_is_recorded(self):
        f = staged()
        kmode(f, "kernel-reboot")
        rc, rep = self.boot_into(f, OLD)
        self.assertEqual((rc, rep["kernel_event"]), (0, "fell_back"), rep)
        self.assertEqual(osapply.Kernel.default_kver(f.cfg), OLD)
        self.refused(f, "R22", mode="kernel-good")

    def test_one_self_revert_then_never_again(self):
        f = staged()
        kmode(f, "kernel-reboot")
        self.boot_into(f, NEW)
        rc, rep = kmode(f, "kernel-revert", reason="the customer guest is not running")
        self.assertEqual(rc, 0, rep)
        self.assertEqual(f.reboots, [OLD, NEW])
        self.assertEqual(osapply.Kernel.default_kver(f.cfg), OLD, "the self-revert boots the default, the old kernel")
        rc, rep = self.boot_into(f, OLD)
        self.assertEqual(rep["kernel_event"], "self_reverted")
        # the same step can never revert again
        st = json.loads(f.tree[osapply.KERNEL_STATE])
        st["phase"] = "judging"
        f.tree[osapply.KERNEL_STATE] = json.dumps(st)
        f.running = NEW
        self.refused(f, "R22", mode="kernel-revert")
        self.assertEqual(len(f.reboots), 2)

    def test_a_revert_that_comes_back_new_is_never_retried(self):
        f = staged()
        kmode(f, "kernel-reboot")
        self.boot_into(f, NEW)
        kmode(f, "kernel-revert")
        rc, rep = self.boot_into(f, NEW)
        self.assertEqual(rep["kernel_event"], "revert_failed")
        self.refused(f, "R22", mode="kernel-revert")

    def test_revert_refuses_an_unknown_default(self):
        f = staged()
        kmode(f, "kernel-reboot")
        self.boot_into(f, NEW)
        f.tree[osapply.KERNEL_DEFAULT_CFG] = f'GRUB_DEFAULT="gnulinux-advanced-{U}>gnulinux-9.9.9-1-pve-advanced-{U}"\n'
        f.cfg = f.render()
        self.refused(f, "R20", mode="kernel-revert")
        self.assertEqual(len(f.reboots), 1)

    def test_cancel_clears_the_flag(self):
        f = staged()
        rc, rep = kmode(f, "kernel-cancel")
        self.assertEqual(rc, 0, rep)
        self.assertFalse(f.env.get("felhom_next"), "the flag is gone")
        self.assertEqual(rep["kernel"]["phase"], "cancelled")

    def test_status_is_read_only_and_names_setup_problems(self):
        f = kfake(esp_fs="ext4")
        rc, rep = kmode(f, "kernel-status")
        self.assertEqual(rc, 0, rep)
        self.assertTrue(rep["kernel"]["setup_problems"])
        self.assertEqual(f.tree, {})
        self.assertFalse([c for c in f.calls if c[1][0] in ("update-grub", "systemctl", "apt-get")])

    def test_facts_carry_the_kernel_lane(self):
        f = staged()
        f.plan = {"release_id": "facts", "layer": "host", "lane": "fast", "vmid": 9201, "mode": "facts", "packages": []}
        rc, rep = run(f)
        self.assertEqual(rc, 0, rep)
        h = rep["facts"]["host"]
        self.assertEqual(h["kernel_lane"]["phase"], "staged")
        self.assertEqual(h["kernel_next_boot"], NEW)
        self.assertIn("one-shot", h["kernel_next_boot_source"])


if __name__ == "__main__":
    unittest.main()