#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Behaviour tests for felhom-host-install.sh — the paths that need no Proxmox host. HOW IT RUNS ANYWHERE. The installer runs as root on a Proxmox host; the CI runner is Alpine + BusyBox + bash + python3 + git. So each test lifts the functions it needs out of felhom-host-install.sh VERBATIM (by name, `^name() {` to the first `^}`), runs them under bash in a temp directory, and replaces the host commands (`systemctl`, `chown`, …) with PATH stubs that record what they were asked. Paths the functions act on are variables the test points into the temp directory. Nothing touches the real host. Where behaviour cannot be isolated, a STATIC test checks the wiring (the function is CALLED, from the right place) — a helper defined and never called is the seam-built-never-wired shape. Rows: R-275, R-276, R-881 (uninstall residue); R-306 (--preflight-only writes no state); R-130 (lvm minimum wording); R-180 (archive storage outside the ACL set); R-179 (NAS units); R-310 (golden refusal wording, the uninstall's terminal); R-274 (local golden reuse). Run: python3 scripts/test_hostinstall.py """ import os import re import shutil import stat import subprocess import sys import tempfile HERE = os.path.dirname(os.path.abspath(__file__)) SCRIPT = os.path.join(HERE, "felhom-host-install.sh") AGENT_OS_APPLY = os.path.join(os.path.dirname(os.path.dirname(HERE)), "felhom-agent", "configs", "felhom-os-apply") # The root-owned files the agent's config bundle installs (felhom-agent configs/felhom-os-apply # BUNDLE_FILES, agent v0.147.0). Frozen here so CI (which has no sibling checkout) still checks it; # where the sibling IS present, test_bundle_list_is_current fails when the bundle gains a file. BUNDLE_DESTS = [ "/usr/local/sbin/felhom-mkfs-guarded", "/usr/local/sbin/felhom-selfupdate-guarded", "/usr/local/sbin/felhom-pbs-apply", "/usr/local/sbin/felhom-backup-target-apply", "/usr/local/sbin/felhom-os-apply", "/usr/local/sbin/felhom-crash-guard", "/usr/local/sbin/felhom-priv-apply", "/var/lib/vz/snippets/felhom-guest-hook.sh", "/usr/local/sbin/felhom-shared-parent.sh", "/etc/systemd/system/felhom-shared-parent.service", "/etc/systemd/system/felhom-crash-guard.service", "/etc/systemd/system/felhom-crash-guard-check.service", "/etc/systemd/system/felhom-crash-guard-check.timer", "/etc/felhom/crash-guard.conf", "/etc/systemd/system/felhom-agent.service", "/etc/systemd/system/felhom-agent-rollback.service", "/etc/systemd/system/felhom-agent.service.d/felhom-agent-limits.conf", "/usr/local/sbin/felhom-mgmt-watchdog", "/etc/tmpfiles.d/felhom-privsep.conf", "/etc/systemd/system/felhom-mgmt-watchdog.service", "/etc/systemd/system/felhom-mgmt-watchdog.timer", "/etc/systemd/system/felhom-sshd.service", "/etc/felhom-oob.nft", "/etc/systemd/system/felhom-oob-nft.service", "/etc/sudoers.d/felhom-op", "/etc/sudoers.d/felhom-agent", ] SRC = open(SCRIPT, encoding="utf-8").read() def func(name): m = re.search(r"^%s\(\) \{[^\n]*\n.*?^\}\n" % re.escape(name), SRC, re.S | re.M) if not m: raise AssertionError("%s() not found in felhom-host-install.sh" % name) return m.group(0) def one_liners(): """The log_* helpers and die (one-line definitions).""" out = [l for l in SRC.splitlines() if re.match(r"^(log_\w+|die)\(\)\s+\{.*\}\s*$", l)] if len(out) < 8: raise AssertionError("log helpers not found (%d)" % len(out)) return "RED=; GREEN=; YELLOW=; BLUE=; CYAN=; NC=\n" + "\n".join(out) + "\n" def section(start_re, end_re): s = re.search(start_re, SRC, re.M) e = re.search(end_re, SRC, re.M) if not s or not e or e.start() <= s.start(): raise AssertionError("section %r..%r not found" % (start_re, end_re)) return SRC[s.start():e.start()] class Sandbox: """A temp dir with a stub bin/ first on PATH. Each stub appends its argv to calls.log.""" def __init__(self): self.root = tempfile.mkdtemp(prefix="hostinstall-test-") self.bin = os.path.join(self.root, "bin") os.mkdir(self.bin) self.calls = os.path.join(self.root, "calls.log") open(self.calls, "w").close() def stub(self, name, body="exit 0"): p = os.path.join(self.bin, name) with open(p, "w") as f: f.write('#!/bin/sh\nprintf "%%s\\n" "%s $*" >> "%s"\n%s\n' % (name, self.calls, body)) os.chmod(p, 0o755) def path(self, *parts): return os.path.join(self.root, *parts) def logged(self): return open(self.calls).read() def run(self, script): env = dict(os.environ) env["PATH"] = self.bin + os.pathsep + env.get("PATH", "") env["SB"] = self.root p = subprocess.run(["bash", "-c", script], capture_output=True, text=True, env=env) return p.returncode, p.stdout + p.stderr def close(self): shutil.rmtree(self.root, ignore_errors=True) def prelude(dry=False): return one_liners() + ("DRY_RUN=%s\n" % ("true" if dry else "false")) + func("run") # ── R-275: the agent config and every copy of it ───────────────────────────────────────────────── # The names measured on demo-hp 2026-08-09; none but the last matched the old `.bak*` glob's intent, # and the old glob missed even that one's siblings. DEMO_HP_COPIES = ["agent.json.campaign8-before", "agent.json.campaign9-before", "agent.json.campaign9-prev", "agent.json.pre-e-target-move", "agent.json.pre-prunegate.bak"] def test_purge_default_dir_removes_every_copy(): sb = Sandbox() try: d = sb.path("etc-felhom-agent") os.mkdir(d) for n in ["agent.json", ".hidden-copy"] + DEMO_HP_COPIES: open(os.path.join(d, n), "w").write("secret") rc, out = sb.run(prelude() + func("_purge_agent_config") + 'AGENT_CFG_DIR_DEFAULT="$SB/etc-felhom-agent"\n_purge_agent_config "$SB/etc-felhom-agent/agent.json"\n') assert rc == 0, out left = os.listdir(d) if os.path.exists(d) else [] assert not os.path.exists(d), "agent config dir survived the uninstall with: %s" % sorted(left) finally: sb.close() def test_purge_custom_path_keeps_foreign_files(): sb = Sandbox() try: d = sb.path("shared") os.mkdir(d) for n in ["agent.json", "other.conf"] + DEMO_HP_COPIES: open(os.path.join(d, n), "w").write("x") rc, out = sb.run(prelude() + func("_purge_agent_config") + 'AGENT_CFG_DIR_DEFAULT="/etc/felhom-agent"\n_purge_agent_config "$SB/shared/agent.json"\n') assert rc == 0, out left = sorted(os.listdir(d)) assert left == ["other.conf"], "custom-path purge left/removed the wrong files: %s" % left finally: sb.close() def test_purge_dry_run_touches_nothing(): sb = Sandbox() try: d = sb.path("etc-felhom-agent") os.mkdir(d) open(os.path.join(d, "agent.json"), "w").write("x") rc, out = sb.run(prelude(dry=True) + func("_purge_agent_config") + 'AGENT_CFG_DIR_DEFAULT="$SB/etc-felhom-agent"\n_purge_agent_config "$SB/etc-felhom-agent/agent.json"\n') assert rc == 0, out assert os.path.exists(os.path.join(d, "agent.json")), "dry-run removed the config" assert "rm -rf" in out, "dry-run did not print the removal: %s" % out finally: sb.close() def test_seal_makes_old_copies_root_only(): sb = Sandbox() try: sb.stub("chown") d = sb.path("etc-felhom-agent") os.mkdir(d) for n in DEMO_HP_COPIES: p = os.path.join(d, n) open(p, "w").write("x") os.chmod(p, 0o644) rc, out = sb.run(prelude() + func("_seal_old_agent_config") + '_seal_old_agent_config "$SB/etc-felhom-agent"\n') assert rc == 0, out log = sb.logged() for n in DEMO_HP_COPIES: p = os.path.join(d, n) assert "chown root:root %s" % p in log, "%s not given to root: %s" % (n, log) assert stat.S_IMODE(os.stat(p).st_mode) == 0o600, "%s mode %o" % (n, stat.S_IMODE(os.stat(p).st_mode)) assert os.path.exists(p), "seal DELETED %s (it may be an operator's backup)" % n assert "now root-only" in out finally: sb.close() def test_seal_is_called_when_the_user_is_created(): body = func("step_agent_install") m = re.search(r'useradd --system[^\n]*"\$AGENT_USER"\n(.*?)\n\s*fi\n', body, re.S) assert m and '_seal_old_agent_config "$AGENT_CFG_DIR_DEFAULT"' in m.group(1), \ "_seal_old_agent_config is not called right after the service user is created" def test_uninstall_wiring(): body = func("run_uninstall") stop = body.find("run systemctl stop felhom-agent") purge = body.find('_purge_agent_config "$agent_cfg"') wg = body.find("_teardown_wg_tunnel") assert purge > 0, "run_uninstall does not call _purge_agent_config" assert wg > 0, "run_uninstall does not call _teardown_wg_tunnel (R-276)" assert stop > 0 and stop < wg, "the WireGuard teardown must run after the agent is stopped" assert '"${agent_cfg}".bak*' not in body, "the old .bak* glob is back" assert re.search(r'for bak in "\$\{AGENT_SUDOERS\}"\.\*', body), "sudoers copies are not removed" # ── R-276: the WireGuard tunnel ────────────────────────────────────────────────────────────────── def _wg_sandbox(active, enabled, conf, sticky=False): sb = Sandbox() # systemctl stub: is-active/is-enabled read flag files; disable --now clears them (unless sticky). sb.stub("systemctl", r''' case "$1" in is-active) [ -e "$SB/wg.active" ]; exit $? ;; is-enabled) [ -e "$SB/wg.enabled" ]; exit $? ;; disable) rm -f "$SB/wg.enabled"; %s exit 0 ;; esac exit 0''' % ("" if sticky else 'rm -f "$SB/wg.active";')) if active: open(sb.path("wg.active"), "w").close() if enabled: open(sb.path("wg.enabled"), "w").close() if conf: open(sb.path("wg-felhom.conf"), "w").write("[Interface]\nPrivateKey = x\n") return sb WG_RUN = ('WG_UNIT="wg-quick@wg-felhom"; WG_CONF="$SB/wg-felhom.conf"; _WG_TEARDOWN_NOTE=""; _WG_PRESENT=false\n' '_teardown_wg_tunnel\necho "PRESENT=$_WG_PRESENT NOTE=$_WG_TEARDOWN_NOTE"\n') def test_wg_teardown_brings_the_tunnel_down(): sb = _wg_sandbox(active=True, enabled=True, conf=True) try: rc, out = sb.run(prelude() + func("_teardown_wg_tunnel") + WG_RUN) assert rc == 0, out assert "systemctl disable --now wg-quick@wg-felhom" in sb.logged(), sb.logged() assert not os.path.exists(sb.path("wg.active")), "tunnel still active" assert not os.path.exists(sb.path("wg-felhom.conf")), "wg-felhom.conf survived" assert "PRESENT=true NOTE=\n" in out + "\n", out assert "is down" in out finally: sb.close() def test_wg_still_active_is_reported_not_claimed_down(): sb = _wg_sandbox(active=True, enabled=True, conf=True, sticky=True) try: rc, out = sb.run(prelude() + func("_teardown_wg_tunnel") + WG_RUN) assert rc == 0, out assert "STILL active" in out and "is down" not in out, out assert "NOTE=wg-quick@wg-felhom is STILL active" in out, out finally: sb.close() def test_wg_absent_is_a_noop(): sb = _wg_sandbox(active=False, enabled=False, conf=False) try: rc, out = sb.run(prelude() + func("_teardown_wg_tunnel") + WG_RUN) assert rc == 0, out assert "disable" not in sb.logged(), sb.logged() assert "PRESENT=false" in out, out finally: sb.close() def test_wg_dry_run_changes_nothing(): sb = _wg_sandbox(active=True, enabled=True, conf=True) try: rc, out = sb.run(prelude(dry=True) + func("_teardown_wg_tunnel") + WG_RUN) assert rc == 0, out assert os.path.exists(sb.path("wg-felhom.conf")) and os.path.exists(sb.path("wg.active")) assert "disable --now wg-quick@wg-felhom" in out, out finally: sb.close() def test_statement_names_the_tunnel_and_the_peer(): sb = Sandbox() try: sb.stub("pvesm", "exit 1") rc, out = sb.run(prelude() + func("_uninstall_statement") + 'vmid=9201; pool_removed=false; _busy_mounts=(); _had_break_glass=false; REMOVE_GOLDEN=false\n' 'PVE_POOL=felhom; ISLAND_BRIDGE=vmbr9; WG_UNIT="wg-quick@wg-felhom"; WG_CONF=/etc/wireguard/wg-felhom.conf\n' '_WG_PRESENT=true; _WG_TEARDOWN_NOTE=""\n_uninstall_statement full\n') assert rc == 0, out wiped, kept = out.split("KEPT", 1) assert "WireGuard tunnel to the Felhom off-site endpoint" in wiped, out assert "WireGuard PEER" in kept, out assert "priv-apply" in wiped, out finally: sb.close() # ── R-881: every file the config bundle installs is removed by the uninstall ───────────────────── def test_uninstall_removes_every_bundle_file(): usect = section(r"^_guest_drive_note\(\)", r"^# run_adopt_pool") # the uninstall names some paths through these variables — expand them before searching. for var, val in [("AGENT_UNIT", "/etc/systemd/system/felhom-agent.service"), ("AGENT_SUDOERS", "/etc/sudoers.d/felhom-agent"), ("AGENT_BIN", "/usr/local/bin/felhom-agent")]: usect = usect.replace("${%s}" % var, val).replace("$%s" % var, val) missing = [p for p in BUNDLE_DESTS if p not in usect] assert not missing, "the uninstall does not remove bundle file(s): %s" % missing def test_bundle_list_is_current(): if not os.path.exists(AGENT_OS_APPLY): print(" note: %s absent (CI) — the frozen list is checked, not its currency" % AGENT_OS_APPLY) return text = open(AGENT_OS_APPLY, encoding="utf-8").read() m = re.search(r"^BUNDLE_FILES = \[(.*?)^\]", text, re.S | re.M) assert m, "BUNDLE_FILES not found in felhom-os-apply" dests = re.findall(r'^\s*\("(/[^"]+)"', m.group(1), re.M) assert len(dests) >= 20, "parsed only %d bundle entries" % len(dests) new = sorted(set(dests) - set(BUNDLE_DESTS)) assert not new, "the agent bundle installs file(s) this test (and maybe the uninstall) does not know: %s" % new # ── R-306: --preflight-only writes no state ────────────────────────────────────────────────────── def _state_run(sb, preflight_only, dry=False): return sb.run(prelude(dry=dry) + func("_state_mark") + func("_state_put") + func("_state_get") + 'PREFLIGHT_ONLY=%s\nSTATE_DIR="$SB/state"; STATE_FILE="$SB/state/state.json"\n' '_state_put dnsmasq_preexisting yes\n_state_mark preflight\necho "GOT=$(_state_get dnsmasq_preexisting)"\n' % ("true" if preflight_only else "false")) def test_preflight_only_writes_no_state(): sb = Sandbox() try: rc, out = _state_run(sb, preflight_only=True) assert rc == 0, out assert not os.path.exists(sb.path("state", "state.json")), \ "--preflight-only wrote state.json: %s" % open(sb.path("state", "state.json")).read() assert "GOT=\n" in out + "\n", out finally: sb.close() def test_install_preflight_does_write_state(): # the control: the same helpers DO write on a real install, or the test above proves nothing. sb = Sandbox() try: rc, out = _state_run(sb, preflight_only=False) assert rc == 0, out assert "GOT=yes" in out, out assert '"preflight"' in open(sb.path("state", "state.json")).read() finally: sb.close() def test_state_json_has_no_other_writer(): # every write must go through the two guarded helpers; a direct write would bypass the guard. bad = [l.strip() for l in SRC.splitlines() if re.search(r'>\s*"?\$STATE_FILE|json\.dump\(d,open\(f', l) and not re.match(r"\s*STATE_FILE=\"\$STATE_FILE\" python3 -c", l)] assert not bad, "state.json written outside _state_mark/_state_put: %s" % bad body = func("step_preflight") assert "_state_put dnsmasq_preexisting" in body, "the ownership record no longer goes through _state_put" # ── R-130: the local-lvm minimum says what it does (it warns; the install continues) ──────────── def test_lvm_minimum_is_named_as_what_it_does(): assert not re.search(r"HARD_MIN_LVM|hard min", SRC), "a 'hard min' that only warns is back" body = func("step_preflight") m = re.search(r'^.*RECOMMENDED_MIN_LVM_GIB.*$', body, re.M) assert m and "log_warn" in m.group(0) and "die" not in m.group(0), "the lvm check is not a warning: %s" % (m and m.group(0)) assert "recommended" in m.group(0) and "continues" in m.group(0), m.group(0) # ── R-180: the archive storage must be one the agent's token is granted on ───────────────────── def _granted(storages, archive, target="felhom-backup"): sb = Sandbox() try: rc, out = sb.run(func("_archive_storage_granted") + 'PVE_STORAGES=(%s); ARCHIVE_STORAGE="%s"; BACKUP_TARGET_ID="%s"\n' 'if _archive_storage_granted; then echo GRANTED; else echo REFUSED; fi\n' % (storages, archive, target)) assert rc == 0, out return out.strip() finally: sb.close() def test_archive_storage_in_the_acl_set_is_granted(): assert _granted("local local-lvm felhom-pbs", "local") == "GRANTED", "archive storage case ('local local-lvm felhom-pbs', 'local') -> %s, want GRANTED" % _granted("local local-lvm felhom-pbs", "local") assert _granted("local nvme-scratch", "nvme-scratch") == "GRANTED", "archive storage case ('local nvme-scratch', 'nvme-scratch') -> %s, want GRANTED" % _granted("local nvme-scratch", "nvme-scratch") # --acl-storages adds it def test_archive_storage_outside_the_acl_set_is_refused(): # the demo-hp 2026-08-03 shape with a storage that is not the backup target assert _granted("local local-lvm felhom-pbs", "nvme-scratch") == "REFUSED", "archive storage case ('local local-lvm felhom-pbs', 'nvme-scratch') -> %s, want REFUSED" % _granted("local local-lvm felhom-pbs", "nvme-scratch") assert _granted("local local-lvm felhom-pbs", "local-lvm2") == "REFUSED", "archive storage case ('local local-lvm felhom-pbs', 'local-lvm2') -> %s, want REFUSED" % _granted("local local-lvm felhom-pbs", "local-lvm2") # no prefix match assert _granted("local local-lvm felhom-pbs", "felhom-backup", target="") == "REFUSED", "archive storage case ('local local-lvm felhom-pbs', 'felhom-backup', target='') -> %s, want REFUSED" % _granted("local local-lvm felhom-pbs", "felhom-backup", target="") def test_archive_storage_on_the_backup_target_is_granted_in_step_6(): assert _granted("local local-lvm felhom-pbs", "felhom-backup") == "GRANTED", "archive storage case ('local local-lvm felhom-pbs', 'felhom-backup') -> %s, want GRANTED" % _granted("local local-lvm felhom-pbs", "felhom-backup") def test_archive_storage_check_is_wired_into_preflight(): body = func("step_preflight") m = re.search(r"archive storage '\$ARCHIVE_STORAGE' present.*?_archive_storage_granted[^\n]*\n[^\n]*\|\| die", body, re.S) assert m, "step_preflight does not refuse when _archive_storage_granted fails" # and before anything is minted: the call sits in step_preflight, which runs before step_token. pre = [m.start() for m in re.finditer(r"^\s*step_preflight\s*$", SRC, re.M)] tok = re.search(r"^step_token\s*$", SRC, re.M) assert pre and tok and max(pre) < tok.start(), "preflight no longer runs before the token step" # ── R-179: the NAS network-storage units ───────────────────────────────────────────────────────── AGENT_NETMOUNT = os.path.join(os.path.dirname(os.path.dirname(HERE)), "felhom-agent", "internal", "storage", "netmount.go") NET_MARK = "# Managed by felhom-agent (network storage) — do not edit by hand.\n" SHARE = r"mnt-felhom\x2ddrives-Felhom\x2dShare" NET_SYSTEMCTL = """ for u in "$@"; do last="$u"; done case "$1" in is-active) [ -e "$SB/active/$last" ]; exit $? ;; disable) [ -e "$SB/sticky/$last" ] || rm -f "$SB/active/$last"; exit 0 ;; esac exit 0""" def _net_sandbox(sticky=()): sb = Sandbox() os.mkdir(sb.path("units")); os.mkdir(sb.path("active")); os.mkdir(sb.path("sticky")) sb.stub("systemctl", NET_SYSTEMCTL) units = {SHARE + ".automount": NET_MARK + "[Automount]\n", SHARE + ".mount": NET_MARK + "[Mount]\n", r"mnt-felhom\x2ddrives-disk1.mount": "# Managed by felhom-agent — do not edit by hand.\n[Mount]\n", "mnt-other.mount": "[Mount]\nWhere=/mnt/other\n"} for n, c in units.items(): open(sb.path("units", n), "w").write(c) open(sb.path("active", SHARE + ".mount"), "w").close() for n in sticky: open(sb.path("sticky", n), "w").close() return sb NET_RUN = ('NET_UNIT_DIR="$SB/units"; NET_UNIT_MARKER="Managed by felhom-agent (network storage)"\n' '_NET_UNITS_REMOVED=(); _NET_UNITS_BUSY=()\n_remove_network_storage_units\n' 'echo "REMOVED=${#_NET_UNITS_REMOVED[@]} BUSY=${_NET_UNITS_BUSY[*]}"\n') def test_net_units_removed_and_only_ours(): sb = _net_sandbox() try: rc, out = sb.run(prelude() + func("_remove_network_storage_units") + NET_RUN) assert rc == 0, out left = sorted(os.listdir(sb.path("units"))) assert left == sorted([r"mnt-felhom\x2ddrives-disk1.mount", "mnt-other.mount"]), \ "wrong units left after the uninstall: %s" % left log = sb.logged() a = log.find("disable --now -- %s.automount" % SHARE) m = log.find("disable --now -- %s.mount" % SHARE) assert a >= 0 and m >= 0 and a < m, "automount must be stopped before its mount:\n%s" % log assert "disk1" not in log and "mnt-other" not in log, "touched a unit that is not a network share:\n%s" % log assert "REMOVED=2 BUSY=" in out, out finally: sb.close() def test_net_units_busy_share_is_not_forced(): sb = _net_sandbox(sticky=(SHARE + ".mount",)) try: rc, out = sb.run(prelude() + func("_remove_network_storage_units") + NET_RUN) assert rc == 0, out assert os.path.exists(sb.path("units", SHARE + ".mount")), "a busy share's unit was removed" assert not os.path.exists(sb.path("units", SHARE + ".automount")) assert "BUSY=%s.mount" % SHARE in out and "NOT forcing" in out, out assert "umount" not in sb.logged(), sb.logged() finally: sb.close() def test_net_units_dry_run_changes_nothing(): sb = _net_sandbox() try: rc, out = sb.run(prelude(dry=True) + func("_remove_network_storage_units") + NET_RUN) assert rc == 0, out assert len(os.listdir(sb.path("units"))) == 4, os.listdir(sb.path("units")) assert "disable" not in sb.logged(), sb.logged() finally: sb.close() def test_net_units_wired_before_the_umount_loop(): body = func("run_uninstall") call = body.find("_remove_network_storage_units") loop = body.find("findmnt -rn -o TARGET") assert call > 0, "run_uninstall does not call _remove_network_storage_units (R-179)" assert call < loop, "the share units must be stopped before the drive umount loop" def test_net_unit_marker_matches_the_agent(): m = re.search(r'^NET_UNIT_MARKER="([^"]+)"', SRC, re.M) assert m, "NET_UNIT_MARKER not found" if not os.path.exists(AGENT_NETMOUNT): print(" note: %s absent (CI) — marker currency not checked" % AGENT_NETMOUNT) return a = re.search(r'netUnitMarker\s*=\s*"([^"]+)"', open(AGENT_NETMOUNT, encoding="utf-8").read()) assert a and a.group(1) == m.group(1), "installer marker %r != agent netUnitMarker %r" % (m.group(1), a and a.group(1)) # ── R-310: the golden refusal names the vouched version once; no terminal is a refusal, in words ── GOLDEN_RUN = """ golden_local_matches_manifest() { GOLDEN_CHECK_WHY="%s"; return 1; } resolve_artifacts() { :; } _state_mark() { :; } GOLDEN_VOLID="local:backup/vzdump-lxc-9100-2026_08_03-07_33_00.tar.zst"; GOLDEN_VOLID_EXPLICIT=true FORCE_GITEA_GOLDEN=false; ART_GOLDEN_VER="0.213.0" step_golden echo REACHED """ def _golden_refusal(why): sb = Sandbox() try: rc, out = sb.run(prelude() + func("step_golden") + GOLDEN_RUN % why) assert rc != 0 and "REACHED" not in out and "refusing the golden you named" in out, out return out finally: sb.close() def test_golden_refusal_names_the_vouched_version_once(): out = _golden_refusal("it is controller 0.192.0, but the vouched golden is 0.213.0") assert out.count("0.213.0") == 1, "the vouched version is stated %d times:\n%s" % (out.count("0.213.0"), out) def test_golden_refusal_still_names_the_version_when_the_reason_does_not(): out = _golden_refusal("the archive could not be resolved to a file on disk") assert "The vouched golden is 0.213.0." in out, out def _tty_run(with_tty): import fcntl import termios script = prelude() + func("_require_tty") + '_require_tty "the typed vmid confirmation"\necho PASSED\n' if not with_tty: p = subprocess.run(["bash", "-c", script], capture_output=True, text=True, stdin=subprocess.DEVNULL, start_new_session=True) return p.returncode, p.stdout + p.stderr import pty master, slave = pty.openpty() def ctty(): os.setsid() fcntl.ioctl(slave, termios.TIOCSCTTY, 0) try: p = subprocess.run(["bash", "-c", script], capture_output=True, text=True, stdin=slave, preexec_fn=ctty) return p.returncode, p.stdout + p.stderr finally: os.close(master); os.close(slave) def test_require_tty_refuses_in_words_without_a_terminal(): rc, out = _tty_run(False) assert rc != 0 and "PASSED" not in out, out assert "needs a terminal" in out and "nothing was destroyed" in out, out assert "No such device" not in out, out def test_require_tty_passes_with_a_terminal(): # the control: with a controlling terminal the guard lets the prompt happen. rc, out = _tty_run(True) assert rc == 0 and "PASSED" in out, out def test_require_tty_guards_the_uninstall_prompt(): body = func("run_uninstall") g = body.find('_require_tty "the typed vmid confirmation"') r = body.find('read -rp "Type the vmid') assert g > 0 and g < r, "the vmid confirmation is not guarded by _require_tty" # ── R-274: a local golden is checked against the manifest, and the disclosure says it is reused ── def test_local_golden_is_checked_before_use(): body = func("step_golden") chk = body.find('golden_local_matches_manifest "$GOLDEN_VOLID"') use = body.find('log_skip " using local golden') assert 0 < chk < use, "step_golden adopts a local golden without checking it against the manifest" g = func("golden_local_matches_manifest") assert "ART_GOLDEN_SHA" in g and "ART_GOLDEN_VER" in g, "the check no longer compares sha256 and version" def test_byo_disclosure_names_the_golden_reuse(): body = func("_byo_disclosure_ack") assert re.search(r"reuse:.*golden.*vouched", body, re.S), "the BYO disclosure does not say a local golden is reused (R-274)" def main(): tests = [(n, f) for n, f in sorted(globals().items()) if n.startswith("test_") and callable(f)] fails = 0 for name, f in tests: try: f() print("PASS", name) except AssertionError as e: fails += 1 print("FAIL", name, "--", e) print("%d passed, %d failed" % (len(tests) - fails, fails)) return 1 if fails else 0 if __name__ == "__main__": sys.exit(main())