felhom.eu/scripts/ep0-copy-gc/test_ep0_copy_gc.py

#!/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_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()