admin
b43705ac7b
- The R-297 named-golden refusal no longer repeats "The vouched golden is X." when its reason already says it (kept when the reason does not name the version). - --uninstall checks it can open /dev/tty before the typed vmid confirmation and, if not, refuses with a sentence (deliberate; --force does not skip it) instead of "/dev/tty: No such device". scripts/test_hostinstall.py: test_golden_refusal_* (2), test_require_tty_* (3, with a pty control). The runbook line naming the pty requirement is a documentation edit for the lead. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
607 lines
27 KiB
Python
607 lines
27 KiB
Python
#!/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).
|
|
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"
|
|
|
|
|
|
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())
|