#!/usr/bin/env python3 """Tests for felhom-ep0-copy-gc (R-901). No PBS is reached: `proxmox-backup-client` is a fake on PATH that answers `namespace list` from FAKE_EP0_NS and records every `namespace delete`. Each test asserts the CONSEQUENCE: which namespace was deleted from the copy, and that nothing is deleted when ep0's answer cannot be trusted. Run: python3 scripts/ep0-copy-gc/test_ep0_copy_gc.py""" import datetime as dt import importlib.machinery import importlib.util import json import os import stat import subprocess import tempfile import unittest HERE = os.path.dirname(os.path.abspath(__file__)) SCRIPT = os.path.join(HERE, "felhom-ep0-copy-gc") FAKE = r'''#!/usr/bin/env python3 import json, os, sys a = sys.argv[1:] with open(os.environ["FAKE_LOG"], "a") as f: f.write(" ".join(a) + " pw=" + ("set" if os.environ.get("PBS_PASSWORD") else "missing") + "\n") if a[:2] == ["namespace", "list"]: if os.environ.get("FAKE_EP0_FAIL"): sys.stderr.write("connection refused\n"); sys.exit(255) print(json.dumps([{"ns": n} for n in json.loads(os.environ["FAKE_EP0_NS"])])) sys.exit(0) if a[:2] == ["namespace", "delete"]: sys.exit(0) sys.exit(9) ''' def load(): loader = importlib.machinery.SourceFileLoader("gc", SCRIPT) spec = importlib.util.spec_from_loader("gc", loader) m = importlib.util.module_from_spec(spec) loader.exec_module(m) return m class Decide(unittest.TestCase): def test_rule(self): m = load() d = dt.date(2026, 10, 8) state = {"old-cust": "2026-09-30", "fresh": "2026-10-07", "back": "2026-09-01"} dele, new = m.decide({"old-cust", "fresh", "new-gone", "live", "operator", "back"}, {"live", "back"}, state, d) self.assertEqual(dele, ["old-cust"]) # absent 8 days → delete self.assertEqual(new.get("fresh"), "2026-10-07") # absent 1 day → kept, remembered self.assertEqual(new.get("new-gone"), "2026-10-08") # first seen absent today self.assertNotIn("back", new) # reappeared on ep0 → forgotten self.assertNotIn("operator", new) # KEEP is never a candidate self.assertNotIn("live", new) class EndToEnd(unittest.TestCase): def setUp(self): self.t = tempfile.mkdtemp() bindir = os.path.join(self.t, "bin"); os.makedirs(bindir) p = os.path.join(bindir, "proxmox-backup-client") open(p, "w").write(FAKE); os.chmod(p, os.stat(p).st_mode | stat.S_IEXEC) self.ns = os.path.join(self.t, "ns") for n in ("demo-hp", "gone-cust", "operator"): os.makedirs(os.path.join(self.ns, n)) for f in ("ep0.secret", "local.secret", "fp"): open(os.path.join(self.t, f), "w").write("x") self.state = os.path.join(self.t, "state", "absent.json") self.log = os.path.join(self.t, "calls.log") self.env = dict(os.environ, PATH=bindir + ":" + os.environ["PATH"], EP0_COPY_NS_DIR=self.ns, EP0_COPY_GC_STATE=self.state, EP0_TOKEN_FILE=os.path.join(self.t, "ep0.secret"), EP0_FINGERPRINT_FILE=os.path.join(self.t, "fp"), LOCAL_TOKEN_FILE=os.path.join(self.t, "local.secret"), FAKE_LOG=self.log, FAKE_EP0_NS=json.dumps(["demo-hp", "operator"])) def run_gc(self, *args, **env): e = dict(self.env, **env) return subprocess.run([SCRIPT] + list(args), env=e, capture_output=True, text=True) def deletes(self): if not os.path.exists(self.log): return [] return [l.split()[2] for l in open(self.log) if l.startswith("namespace delete")] def seed(self, ns, first): os.makedirs(os.path.dirname(self.state), exist_ok=True) json.dump({ns: first}, open(self.state, "w")) def test_apply_deletes_after_grace(self): self.seed("gone-cust", (dt.date.today() - dt.timedelta(days=8)).isoformat()) r = self.run_gc("--apply") self.assertEqual(r.returncode, 0, r.stdout + r.stderr) self.assertEqual(self.deletes(), ["gone-cust"]) self.assertIn("pw=set", open(self.log).read()) self.assertEqual(json.load(open(self.state)), {}) def test_inside_grace_nothing_deleted(self): r = self.run_gc("--apply") self.assertEqual(r.returncode, 0) self.assertEqual(self.deletes(), []) self.assertIn("gone-cust", json.load(open(self.state))) def test_dry_run_never_deletes(self): self.seed("gone-cust", "2026-01-01") r = self.run_gc() self.assertEqual(self.deletes(), []) self.assertIn("DRY RUN", r.stdout) def test_ep0_empty_list_aborts(self): self.seed("gone-cust", "2026-01-01") r = self.run_gc("--apply", FAKE_EP0_NS="[]") self.assertEqual(r.returncode, 2) self.assertEqual(self.deletes(), []) def test_ep0_unreachable_aborts(self): self.seed("gone-cust", "2026-01-01") r = self.run_gc("--apply", FAKE_EP0_FAIL="1") self.assertEqual(r.returncode, 2) self.assertEqual(self.deletes(), []) self.assertEqual(json.load(open(self.state)), {"gone-cust": "2026-01-01"}) # state untouched def test_mass_absence_aborts(self): # ep0 rebuilt / half-restored: it lists only demo-hp; three copy namespaces look „deleted" → abort, delete none. for n in ("c2", "c3"): os.makedirs(os.path.join(self.ns, n)) self.seed("gone-cust", "2026-01-01") r = self.run_gc("--apply") self.assertEqual(r.returncode, 2, r.stdout) self.assertEqual(self.deletes(), []) self.assertEqual(json.load(open(self.state)), {"gone-cust": "2026-01-01"}) # state untouched def test_one_deletion_per_run(self): os.makedirs(os.path.join(self.ns, "c2")) os.makedirs(os.path.dirname(self.state), exist_ok=True) json.dump({"gone-cust": "2026-01-01", "c2": "2026-01-01"}, open(self.state, "w")) r = self.run_gc("--apply") self.assertEqual(r.returncode, 0, r.stdout) self.assertEqual(len(self.deletes()), 1) def test_hold_file_stops_everything(self): hold = os.path.join(self.t, "HOLD"); open(hold, "w").write("") self.seed("gone-cust", "2026-01-01") r = self.run_gc("--apply", EP0_COPY_GC_HOLD=hold) self.assertEqual(r.returncode, 0) self.assertEqual(self.deletes(), []) self.assertFalse(os.path.exists(self.log)) # not even ep0 was asked def test_secret_never_printed(self): open(os.path.join(self.t, "ep0.secret"), "w").write("SEKRIT-VALUE") self.seed("gone-cust", "2026-01-01") r = self.run_gc("--apply") self.assertNotIn("SEKRIT", r.stdout + r.stderr) if __name__ == "__main__": unittest.main()