felhom.eu/scripts/test_felhom_restore_beside.py

#!/usr/bin/env python3
"""Tests for felhom-restore-beside.sh (R-834) — a fake `pct` on PATH holds the guest config in a file.

The restored config is the PRODUCTION one (onboot 1, mp8 on the household's drives, mp9 on the
original's bootstrap, NICs up). The script must end with onboot 0, no host-path bind, every NIC
link_down, and never start the guest. Asserted on the consequence: the fake's final config file and
its call log. Red-proof: drop the `pct set --delete` line from the script and test_strips fails.

Run: python3 scripts/test_felhom_restore_beside.py
"""
import os
import pathlib
import stat
import subprocess
import sys
import tempfile
import unittest

SCRIPT = pathlib.Path(__file__).with_name("felhom-restore-beside.sh")

PROD_CONF = """arch: amd64
hostname: demo-hp
onboot: 1
rootfs: nvme-scratch:9299/vm-9299-disk-0.raw,size=16G
mp0: nvme-scratch:9299/vm-9299-disk-1.raw,mp=/var/lib/felhom,backup=1,size=70G
mp8: /mnt/felhom-drives,mp=/mnt/felhom-drives
mp9: /var/lib/felhom-agent/guests/9201/bootstrap,mp=/etc/felhom-bootstrap,ro=1
net0: name=eth0,bridge=vmbr0,hwaddr=BC:24:11:0F:7E:C5,ip=dhcp,type=veth
net1: name=eth1,bridge=vmbr9,hwaddr=BC:24:11:28:B4:F5,ip=169.254.253.2/30,type=veth
"""

# A minimal pct: status (exists iff the conf file exists), restore (writes PROD_CONF, records whether
# --onboot 0 was passed), config, set (--delete / --netN / --onboot), start (recorded — must never run).
FAKE_PCT = r'''#!/usr/bin/env python3
import os, sys
d = os.environ["FAKE_PCT_DIR"]
conf = os.path.join(d, "conf")
with open(os.path.join(d, "calls"), "a") as f:
    f.write(" ".join(sys.argv[1:]) + "\n")
cmd = sys.argv[1]
def read():
    return [l for l in open(conf).read().splitlines() if l]
def write(lines):
    open(conf, "w").write("\n".join(lines) + "\n")
if cmd == "status":
    sys.exit(0 if os.path.exists(conf) else 2)
if cmd == "restore":
    lines = open(os.path.join(d, "prod")).read().splitlines()
    args = sys.argv[3:]
    if "--onboot" in args:
        v = args[args.index("--onboot") + 1]
        lines = [("onboot: " + v) if l.startswith("onboot:") else l for l in lines]
    write([l for l in lines if l])
    sys.exit(0)
if cmd == "config":
    print("\n".join(read()))
    sys.exit(0)
if cmd == "set":
    lines = read()
    a = sys.argv[3:]
    while a:
        k, v = a[0], a[1]
        a = a[2:]
        if k == "--delete":
            drop = set(v.split(","))
            lines = [l for l in lines if l.split(":")[0] not in drop]
        else:
            key = k.lstrip("-")
            lines = [l for l in lines if l.split(":")[0] != key] + [key + ": " + v]
    write(lines)
    sys.exit(0)
if cmd == "start":
    sys.exit(0)
sys.exit(9)
'''


def run(tmp, *args, script=SCRIPT, exists=False):
    bindir = pathlib.Path(tmp, "bin")
    bindir.mkdir(exist_ok=True)
    pct = bindir / "pct"
    pct.write_text(FAKE_PCT)
    pct.chmod(pct.stat().st_mode | stat.S_IEXEC)
    pathlib.Path(tmp, "prod").write_text(PROD_CONF)
    if exists:
        pathlib.Path(tmp, "conf").write_text(PROD_CONF)
    env = dict(os.environ, PATH=f"{bindir}:{os.environ['PATH']}", FAKE_PCT_DIR=tmp)
    return subprocess.run(["bash", str(script), *args], env=env, capture_output=True, text=True)


class RestoreBeside(unittest.TestCase):
    def test_strips(self):
        with tempfile.TemporaryDirectory() as tmp:
            r = run(tmp, "9299", "tmp:backup/ct/9201/x", "nvme-scratch")
            self.assertEqual(r.returncode, 0, r.stderr + r.stdout)
            conf = pathlib.Path(tmp, "conf").read_text()
            calls = pathlib.Path(tmp, "calls").read_text()
            self.assertIn("onboot: 0", conf)
            self.assertNotIn("onboot: 1", conf)
            for line in conf.splitlines():
                self.assertFalse(line.startswith("mp") and ": /" in line, f"host bind left: {line}")
                if line.startswith("net"):
                    self.assertIn("link_down=1", line)
            self.assertIn("mp0:", conf, "a storage volume must stay")
            self.assertNotIn("\nstart", "\n" + calls, "the script started the guest")
            self.assertIn("--onboot 0", calls.splitlines()[1], "onboot 0 must be set AT restore time")

    def test_refuses_an_existing_vmid(self):
        with tempfile.TemporaryDirectory() as tmp:
            r = run(tmp, "9201", "tmp:backup/ct/9201/x", "nvme-scratch", exists=True)
            self.assertNotEqual(r.returncode, 0)
            self.assertIn("already exists", r.stderr)
            self.assertNotIn("restore", pathlib.Path(tmp, "calls").read_text())

    def test_refuses_the_agent_bands(self):
        for v in ("990003", "9999", "abc"):
            with tempfile.TemporaryDirectory() as tmp:
                r = run(tmp, v, "tmp:backup/ct/9201/x", "s")
                self.assertNotEqual(r.returncode, 0, v)

    def test_readback_catches_a_bind_left_behind(self):
        # Red-proof built in: the same script with the --delete line removed must FAIL its read-back.
        with tempfile.TemporaryDirectory() as tmp:
            broken = pathlib.Path(tmp, "broken.sh")
            broken.write_text(SCRIPT.read_text().replace('\tpct set "$vmid" --delete "$binds"\n', "\t:\n"))
            self.assertNotEqual(broken.read_text(), SCRIPT.read_text(), "mutation did not apply")
            r = run(tmp, "9299", "tmp:backup/ct/9201/x", "s", script=broken)
            self.assertEqual(r.returncode, 2, r.stdout + r.stderr)
            self.assertIn("host-bind-left", r.stderr)


if __name__ == "__main__":
    sys.exit(unittest.main())