#!/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) # 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 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 "-s" in a: return self.sim(a) 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": return 0, "felhom-controller\trunning\tUp 1 hour (healthy)\napp\trunning\tUp 1 hour (healthy)\n", "" if cmd == "getent": return 0, "1.2.3.4 deb.debian.org\n", "" if cmd == "sh": 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: return 0, "'http://x/libc6.deb' libc6.deb 4000000 SHA256:x\n", "" 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 = 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]] 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) self.assertNotIn("restart_needed", rep, "the restart scan runs only after an install (R-845)") 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]] 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") 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_host_scans_every_pass_guest_only_after_install(self): # A host pass that installs nothing still scans, so a reboot clears the flag (the hub alarm reads it). f = Fake() f.plan["layer"] = "host" f.plan["mode"] = "inventory" f.restart_out = "2101 lxc-start\n" rc, rep = run(f) self.assertEqual(rc, 0, rep) self.assertTrue(rep["reboot_scanned"], rep) self.assertTrue(rep["reboot_needed"], rep) f = Fake() f.plan["layer"] = "host" f.plan["mode"] = "inventory" f.restart_out = "" # after the reboot: nothing maps a deleted file rc, rep = run(f) self.assertTrue(rep["reboot_scanned"] and rep["reboot_needed"] is False, rep) f = Fake() f.plan["mode"] = "inventory" rc, rep = run(f) self.assertFalse(rep["reboot_scanned"], "the guest scans only after an install (R-845)") 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")) if __name__ == "__main__": unittest.main()