#!/usr/bin/env python3 """Tests for configs/felhom-vzdump-partial-clean (R-1006): it removes a DEAD half-written vzdump backup and refuses everything else. Each refusal has a positive control: the same setup without the one condition is removed — so a test cannot pass because the helper refuses everything. A fake storage.cfg and a fake /proc; nothing real is touched. Run: python3 configs/test_felhom_vzdump_partial_clean.py (also run by internal/backup's Go test) """ import os, pathlib, subprocess, tempfile, time, unittest HELPER = pathlib.Path(__file__).resolve().parent / "felhom-vzdump-partial-clean" BOOT = 1_791_967_481 NAME = "vzdump-lxc-9201-2026_10_14-10_42_25" class Box: """A temp tree: a dir storage `local` (path vz), an lvmthin `local-lvm`, a dir storage `other`, and a fake /proc.""" def __init__(self): self.t = tempfile.TemporaryDirectory() r = pathlib.Path(self.t.name) self.vz, self.other, self.nas, self.proc = r / "vz", r / "other", r / "nas", r / "proc" for d in (self.vz / "dump", self.other / "dump", self.nas / "dump", self.proc): d.mkdir(parents=True) (self.proc / "stat").write_text(f"cpu 1 2 3\nbtime {BOOT}\n") self.cfg = r / "storage.cfg" self.cfg.write_text(f"dir: local\n\tpath {self.vz}\n\tcontent iso,backup\n\n" f"lvmthin: local-lvm\n\tthinpool data\n\tvgname pve\n\n" f"dir: other\n\tpath {self.other}\n\tcontent backup\n\n" f"nfs: nas\n\tpath {self.nas}\n\tserver 192.0.2.1\n\texport /x\n\tcontent backup\n") self.pid = 1000 def partial(self, name=NAME + ".tar.dat", mtime=BOOT - 100, where=None, size=4096): p = (where or self.vz / "dump") / name if name.endswith(".tmp"): (p / "etc" / "vzdump").mkdir(parents=True) (p / "etc" / "vzdump" / "pct.conf").write_text("x") for q in [p / "etc" / "vzdump" / "pct.conf", p / "etc" / "vzdump", p / "etc", p]: os.utime(q, (mtime, mtime)) else: p.write_bytes(b"\0" * size) os.utime(p, (mtime, mtime)) return p def process(self, cmdline, fds=(), cwd=None): self.pid += 1 d = self.proc / str(self.pid) (d / "fd").mkdir(parents=True) (d / "cmdline").write_bytes(b"\0".join(a.encode() for a in cmdline) + b"\0") for i, f in enumerate(fds): os.symlink(str(f), d / "fd" / str(i + 3)) os.symlink(str(cwd or "/"), d / "cwd") def run(self, storage, vmid, name, quiet=120): script = (f'source "{HELPER}"; STORAGE_CFG="{self.cfg}"; PROC="{self.proc}"; QUIET_S={quiet}; ' f'main "$@"') r = subprocess.run(["bash", "-c", script, "x", storage, str(vmid), name], capture_output=True, text=True) return r.returncode, r.stdout + r.stderr class Removes(unittest.TestCase): def test_dead_tar_dat_before_boot(self): b = Box(); p = b.partial() rc, out = b.run("local", 9201, p.name) self.assertEqual(rc, 0, out); self.assertFalse(p.exists()); self.assertIn("written before this boot", out) def test_dead_tmp_dir_before_boot(self): b = Box(); p = b.partial(NAME + ".tmp") rc, out = b.run("local", 9201, p.name) self.assertEqual(rc, 0, out); self.assertFalse(p.exists()) def test_quiet_partial_after_boot(self): b = Box(); p = b.partial(mtime=int(time.time()) - 300) rc, out = b.run("local", 9201, p.name) self.assertEqual(rc, 0, out); self.assertFalse(p.exists()); self.assertIn("untouched", out) def test_qemu_vma_dat(self): b = Box(); p = b.partial("vzdump-qemu-341-2026_10_14-10_42_25.vma.dat") rc, out = b.run("local", 341, p.name) self.assertEqual(rc, 0, out); self.assertFalse(p.exists()) class Keeps(unittest.TestCase): """Each case: (setup that must be REFUSED, the same setup that must be REMOVED).""" def pair(self, refuse, control, quiet=120): b = Box(); target, args = refuse(b) rc, out = b.run(*args, quiet=quiet) self.assertNotEqual(rc, 0, f"must refuse: {out}") self.assertIn("REFUSED", out) self.assertTrue(os.path.lexists(target), "a refused entry must still exist") c = Box(); ctarget, cargs = control(c) crc, cout = c.run(*cargs, quiet=quiet) self.assertEqual(crc, 0, f"positive control must be removed: {cout}") self.assertFalse(os.path.lexists(ctarget)) def test_complete_archive_of_same_vmid(self): self.pair(lambda b: (b.partial(NAME + ".tar.zst"), ("local", 9201, NAME + ".tar.zst")), lambda b: (b.partial(), ("local", 9201, NAME + ".tar.dat"))) def test_log_and_notes(self): for suffix in (".log", ".tar.zst.notes"): self.pair(lambda b, s=suffix: (b.partial(NAME + s), ("local", 9201, NAME + s)), lambda b: (b.partial(), ("local", 9201, NAME + ".tar.dat"))) def test_partial_of_another_vmid(self): other = "vzdump-lxc-9202-2026_10_14-10_42_25.tar.dat" self.pair(lambda b: (b.partial(other), ("local", 9201, other)), lambda b: (b.partial(other), ("local", 9202, other))) def test_partial_on_a_non_dir_storage(self): self.pair(lambda b: (b.partial(), ("local-lvm", 9201, NAME + ".tar.dat")), lambda b: (b.partial(), ("local", 9201, NAME + ".tar.dat"))) def test_partial_on_a_network_storage_with_a_path(self): # an nfs storage HAS a path: only the „dir only" rule (both its guards) can refuse it self.pair(lambda b: (b.partial(where=b.nas / "dump"), ("nas", 9201, NAME + ".tar.dat")), lambda b: (b.partial(where=b.other / "dump"), ("other", 9201, NAME + ".tar.dat"))) def test_partial_on_another_storage_than_named(self): self.pair(lambda b: (b.partial(where=b.other / "dump"), ("local", 9201, NAME + ".tar.dat")), lambda b: (b.partial(where=b.other / "dump"), ("other", 9201, NAME + ".tar.dat"))) def test_unknown_storage(self): self.pair(lambda b: (b.partial(), ("nosuch", 9201, NAME + ".tar.dat")), lambda b: (b.partial(), ("local", 9201, NAME + ".tar.dat"))) def test_vzdump_task_in_flight(self): def refuse(b): p = b.partial(); b.process(["task UPID:demo-hp:001:002:6ACF4071:vzdump:9201:felhom-agent@pve!agent:"]); return p, ("local", 9201, p.name) def control(b): p = b.partial(); b.process(["task UPID:demo-hp:001:002:6ACF4071:vzrestore:9202:root@pam:"]); return p, ("local", 9201, p.name) self.pair(refuse, control) def test_any_vzdump_job_even_without_the_vmid(self): # a job for several guests: `vzdump::` — and a vzdump for ANOTHER guest — both refuse for cmd in (["task UPID:demo-hp:001:002:6ACF4071:vzdump::root@pam:"], ["task UPID:demo-hp:001:002:6ACF4071:vzdump:9202:felhom-agent@pve!agent:"]): def refuse(b, c=cmd): p = b.partial(); b.process(c); return p, ("local", 9201, p.name) def control(b): p = b.partial(); b.process(["task UPID:demo-hp:001:002:6ACF4071:vzstart:9202:root@pam:"]); return p, ("local", 9201, p.name) self.pair(refuse, control) def test_vzdump_command_for_the_vmid(self): def refuse(b): p = b.partial(); b.process(["/usr/bin/perl", "/usr/bin/vzdump", "9201", "--storage", "local"]); return p, ("local", 9201, p.name) def control(b): p = b.partial(); b.process(["/usr/bin/perl", "/usr/bin/vzdumpfoo", "9201"]); return p, ("local", 9201, p.name) self.pair(refuse, control) def test_file_held_open_by_an_orphan_writer(self): def refuse(b): p = b.partial(); b.process(["zstd", "--threads=1"], fds=[p]); return p, ("local", 9201, p.name) def control(b): p = b.partial(); b.process(["zstd", "--threads=1"], fds=[b.vz / "dump" / "something-else"]); return p, ("local", 9201, p.name) self.pair(refuse, control) def test_tmp_dir_with_a_process_inside(self): def refuse(b): p = b.partial(NAME + ".tmp"); b.process(["tar", "cpf", "-"], cwd=p / "etc"); return p, ("local", 9201, p.name) def control(b): p = b.partial(NAME + ".tmp"); b.process(["tar", "cpf", "-"], cwd=b.vz); return p, ("local", 9201, p.name) self.pair(refuse, control) def test_recently_written_after_boot(self): self.pair(lambda b: (b.partial(mtime=int(time.time()) - 10), ("local", 9201, NAME + ".tar.dat")), lambda b: (b.partial(mtime=int(time.time()) - 300), ("local", 9201, NAME + ".tar.dat"))) def test_symlink(self): def refuse(b): real = b.partial(where=b.vz) link = b.vz / "dump" / (NAME + ".tar.dat") os.symlink(real, link) return link, ("local", 9201, link.name) # quiet=0: the link was made just now, so only the symlink rule may refuse it self.pair(refuse, lambda b: (b.partial(), ("local", 9201, NAME + ".tar.dat")), quiet=0) def test_near_miss_names(self): for bad in (NAME + ".tar.dat.bak", "vzdump-lxc-92011-2026_10_14-10_42_25.tar.dat", NAME + ".tar.da", "x" + NAME + ".tar.dat", NAME + ".TAR.DAT", "vzdump-lxc-9201-2026_10_14-10_42.tar.dat"): self.pair(lambda b, n=bad: (b.partial(n), ("local", 9201, n)), lambda b: (b.partial(), ("local", 9201, NAME + ".tar.dat"))) def test_path_traversal_and_bad_args(self): b = Box(); p = b.partial() for args in (("local", 9201, "../dump/" + p.name), ("local", 9201, ""), ("loc al", 9201, p.name), ("local", "9201x", p.name), ("local", 12, p.name)): rc, out = b.run(*args) self.assertNotEqual(rc, 0, f"{args}: {out}") self.assertTrue(p.exists()) def test_wrong_kind(self): def refuse(b): # a FILE named .tmp p = b.vz / "dump" / (NAME + ".tmp"); p.write_bytes(b"x"); os.utime(p, (BOOT - 100, BOOT - 100)); return p, ("local", 9201, p.name) self.pair(refuse, lambda b: (b.partial(NAME + ".tmp"), ("local", 9201, NAME + ".tmp"))) if __name__ == "__main__": unittest.main()