#!/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): pass 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": 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 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), "" 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 == "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": 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 = "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//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 (` namespaces="felhom-op-v1" `). 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) if __name__ == "__main__": unittest.main() 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")