felhom-agent/configs/test_felhom_thin_check.py

#!/usr/bin/env python3
"""Tests for configs/felhom-thin-check (R-1012): it lists the thin pool's device ids next to LVM's and NEVER removes
anything. Fake `lvs`, `dmsetup` and `thin_ls` record every call; nothing real is touched. Each refusal has a positive
control (the healthy read), so a test cannot pass because the helper refuses everything.
Run: python3 configs/test_felhom_thin_check.py   (also run by internal/thincheck's Go test)
"""
import os, pathlib, re, subprocess, tempfile, unittest

HELPER = pathlib.Path(__file__).resolve().parent / "felhom-thin-check"


class Box:
    def __init__(self, pool_ids=(1, 2), lvm=(("vm-9201-disk-0", 1), ("vm-9201-disk-1", 2)), attr="twi-aotz--",
                 held="-", reserve_rc=0, thin_ls_rc=0):
        self.t = tempfile.TemporaryDirectory()
        r = pathlib.Path(self.t.name)
        self.log = r / "calls"
        self.log.write_text("")
        lvm_lines = "\\n".join(f"  {n},{i}" for n, i in lvm)
        pool_lines = "\\n".join(str(i) for i in pool_ids)
        self.bins = {}
        for name, body in {
            "lvs": f'''case "$*" in *lv_attr*) echo "  {attr}";; *thin_id*) printf "{lvm_lines}\\n";; esac''',
            "dmsetup": f'''case "$1" in
  status) echo "0 112975872 thin-pool 585 6860/262144 517080/882624 {held} rw discard_passdown queue_if_no_space - 1024";;
  message) [ "$4" = reserve_metadata_snap ] && exit {reserve_rc}; exit 0;;
esac''',
            "thin_ls": f'''[ {thin_ls_rc} -ne 0 ] && exit {thin_ls_rc}; printf "{pool_lines}\\n"''',
        }.items():
            p = r / name
            p.write_text(f'#!/bin/bash\necho "{name} $*" >> "{self.log}"\n{body}\n')
            p.chmod(0o755)
            self.bins[name] = p
        self.lock = r / "lock"

    def run(self, *args):
        script = (f'source "{HELPER}"; LVS="{self.bins["lvs"]}"; DMSETUP="{self.bins["dmsetup"]}"; '
                  f'THIN_LS="{self.bins["thin_ls"]}"; LOCK="{self.lock}"; main "$@"')
        r = subprocess.run(["bash", "-c", script, "x", *args], capture_output=True, text=True)
        return r.returncode, r.stdout, r.stderr

    def calls(self):
        return self.log.read_text().splitlines()


def messages(calls):
    return [c.split()[-1] for c in calls if c.startswith("dmsetup message")]


class Reads(unittest.TestCase):
    def test_healthy_pool_lists_both_sides_and_releases(self):
        b = Box()
        rc, out, err = b.run("pve", "data")
        self.assertEqual(rc, 0, err)
        self.assertEqual(out.splitlines(), ["lvm-before 1 vm-9201-disk-0", "lvm-before 2 vm-9201-disk-1",
                                            "pool 1", "pool 2",
                                            "lvm-after 1 vm-9201-disk-0", "lvm-after 2 vm-9201-disk-1", "ok"])
        self.assertEqual(messages(b.calls()), ["reserve_metadata_snap", "release_metadata_snap"])
        self.assertIn("thin_ls --metadata-snap --no-headers -o DEV /dev/mapper/pve-data_tmeta", b.calls())

    def test_the_2026_10_14_orphan_is_listed(self):
        rc, out, _ = Box(pool_ids=(1, 2, 4)).run("pve", "data")
        self.assertEqual(rc, 0)
        self.assertIn("pool 4", out.splitlines())

    def test_dashed_names_map_to_the_device_mapper_name(self):
        b = Box()
        rc, _, err = b.run("felhom-scratch", "thin-pool")
        self.assertEqual(rc, 0, err)
        self.assertIn("dmsetup status felhom--scratch-thin--pool-tpool", b.calls())


class Refuses(unittest.TestCase):
    def refused(self, box, rc_want, *args):
        rc, out, err = box.run(*args)
        self.assertEqual(rc, rc_want, (out, err))
        self.assertNotIn("ok", out.splitlines())
        return box.calls()

    def test_arguments(self):
        for args in (("pve",), ("pve", "data", "x"), ("pve;id", "data"), ("-pve", "data"), ("pve", "../data"),
                     ("pve", "da ta")):
            calls = self.refused(Box(), 1, *args)
            self.assertEqual(calls, [], args)

    def test_not_a_thin_pool(self):
        calls = self.refused(Box(attr="Vwi-aotz--"), 1, "pve", "vm-9201-disk-1")
        self.assertEqual(messages(calls), [])

    def test_a_snapshot_someone_else_holds_is_never_touched(self):
        b = Box(held="1234")
        rc, _, err = b.run("pve", "data")
        self.assertEqual(rc, 3)
        self.assertEqual(messages(b.calls()), [])  # neither reserve nor release
        # the reason line ends with the reason, not the exit code (found live on demo-hp 2026-10-17)
        self.assertTrue(err.strip().endswith("not touched"), err)

    def test_reserve_failure_does_not_release(self):
        calls = self.refused(Box(reserve_rc=1), 3, "pve", "data")
        self.assertEqual(messages(calls), ["reserve_metadata_snap"])

    def test_thin_ls_failure_still_releases(self):
        calls = self.refused(Box(thin_ls_rc=1), 4, "pve", "data")
        self.assertEqual(messages(calls), ["reserve_metadata_snap", "release_metadata_snap"])


class NeverDeletes(unittest.TestCase):
    def test_the_only_pool_messages_are_reserve_and_release(self):
        src = HELPER.read_text()
        sent = re.findall(r'"\$DMSETUP" message "\$TPOOL" 0 (\S+)', src)
        self.assertEqual(sorted(set(sent)), ["release_metadata_snap", "reserve_metadata_snap"])
        for word in ("delete", "lvremove", "thin_repair", "create_thin", "create_snap", " rm "):
            code = "\n".join(l for l in src.splitlines() if not l.lstrip().startswith("#"))
            self.assertNotIn(word, code, word)


if __name__ == "__main__":
    unittest.main(verbosity=2)