admin
98f3cdd76a
gates / gates (push) Successful in 25s
- felhom-mkfs-guarded: the 0.158.0 exemption for a sibling part mounted at /mnt/<name> + under /mnt/felhom-drives is removed — the agent can make that corroborating mount itself. Every mount of every node outside /mnt/felhom-drives refuses (kernel mount table); an unlistable disk refuses. CandidatePartition.Formattable is false for every part of a disk with a mounted node, so it is never offered for formatting (attach still is). - felhom-internal-place: a failed lvs listing, a listing without the pool, or a blkid exit other than 0/2 refuses (they read as "nothing promised" / "no signature" before). Every new test was seen failing against the 0.158.0 versions. Claude-Session: https://claude.ai/code/session_012qRErfCoiTkvDK9N5XHbzb
256 lines
11 KiB
Python
256 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for configs/felhom-internal-place (felhom.eu architecture/16 §5, §7 — sizes ruled 2026-10-16).
|
|
|
|
The helper is run for real under bash; `lvs`, `lvcreate`, `blkid`, `mkfs.ext4` and `stat` are PATH stubs that read and
|
|
write one JSON state file, so nothing on this machine is touched. Every refusal has a positive control: the same setup
|
|
without the one condition goes through — so a test cannot pass because the helper refuses everything.
|
|
Run: python3 configs/test_felhom_internal_place.py (also run by the Go suite, internal/internalplace)
|
|
"""
|
|
import json, os, pathlib, stat, subprocess, tempfile, unittest
|
|
|
|
HELPER = pathlib.Path(__file__).resolve().parent / "felhom-internal-place"
|
|
GIB = 1 << 30
|
|
|
|
STUB_LVS = r'''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
st = json.load(open(os.environ["IP_STATE"]))
|
|
if st.get("lvs_fail_listing") and "lv_name,pool_lv,lv_size" in sys.argv: sys.exit(5)
|
|
args = [a for a in sys.argv[1:] if not a.startswith("-") and a not in ("b",)]
|
|
fields = sys.argv[sys.argv.index("-o") + 1].split(",")
|
|
target = args[-1]
|
|
listing = target == st["vg"]
|
|
for lv in st["lvs"]:
|
|
full = st["vg"] + "/" + lv["name"]
|
|
if listing and st.get("hide_pool_in_listing") and lv.get("pool", "") == "":
|
|
continue # a listing that silently lost the pool row (the pool's own size stays readable)
|
|
if listing or target == full:
|
|
row = {"lv_name": lv["name"], "pool_lv": lv.get("pool", ""), "lv_size": str(lv["size"])}
|
|
print(" " + " ".join(row[f] for f in fields))
|
|
'''
|
|
STUB_LVCREATE = r'''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
p = os.environ["IP_STATE"]; st = json.load(open(p))
|
|
a = sys.argv[1:]
|
|
size = int(a[a.index("-V") + 1].rstrip("G")) * (1 << 30)
|
|
vg, pool = a[a.index("-T") + 1].split("/")
|
|
st["lvs"].append({"name": a[a.index("-n") + 1], "pool": pool, "size": size, "active": True})
|
|
st.setdefault("calls", []).append("lvcreate " + " ".join(a))
|
|
if st.get("dirty_new"): st["fs"]["felhom-belso"] = {"TYPE": "", "RAW": "junk"}
|
|
json.dump(st, open(p, "w"))
|
|
'''
|
|
STUB_BLKID = r'''#!/usr/bin/env python3
|
|
# like blkid -p: exit 0 when something is found, 2 when nothing is; st["blkid_rc"] forces an error exit
|
|
import json, os, sys
|
|
st = json.load(open(os.environ["IP_STATE"]))
|
|
if st.get("blkid_rc"): sys.exit(st["blkid_rc"])
|
|
fs = st["fs"].get(os.path.basename(sys.argv[-1]), {})
|
|
if "-s" in sys.argv:
|
|
v = fs.get(sys.argv[sys.argv.index("-s") + 1], "")
|
|
if v: print(v)
|
|
sys.exit(0 if v else 2)
|
|
vals = [fs.get(k, "") for k in ("TYPE", "LABEL", "UUID", "RAW") if fs.get(k)]
|
|
if vals: print("\n".join(vals))
|
|
sys.exit(0 if vals else 2)
|
|
'''
|
|
STUB_MKFS = r'''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
p = os.environ["IP_STATE"]; st = json.load(open(p))
|
|
dev = os.path.basename(sys.argv[-1])
|
|
st["fs"][dev] = {"TYPE": "ext4", "LABEL": sys.argv[sys.argv.index("-L") + 1], "UUID": "0b6e4c5a-1d2e-4f3a-9b8c-7d6e5f4a3b2c"}
|
|
st.setdefault("calls", []).append("mkfs.ext4 " + " ".join(sys.argv[1:]))
|
|
json.dump(st, open(p, "w"))
|
|
'''
|
|
STUB_STAT = r'''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
st = json.load(open(os.environ["IP_STATE"]))
|
|
name = os.path.basename(sys.argv[-1])
|
|
if any(lv["name"] == name and lv.get("active", True) for lv in st["lvs"]): print("block special file")
|
|
else: sys.exit(1)
|
|
'''
|
|
|
|
CFG = "dir: local\n\tpath /var/lib/vz\n\tcontent iso,backup\n\nlvmthin: local-lvm\n\tthinpool data\n\tvgname pve\n\tcontent rootdir,images\n"
|
|
|
|
|
|
class Box:
|
|
def __init__(self, pool_gib, lvs=(), fs=None, cfg=CFG, **extra):
|
|
self.t = tempfile.TemporaryDirectory()
|
|
r = pathlib.Path(self.t.name)
|
|
self.bin = r / "bin"; self.bin.mkdir()
|
|
for name, body in (("lvs", STUB_LVS), ("lvcreate", STUB_LVCREATE), ("blkid", STUB_BLKID),
|
|
("mkfs.ext4", STUB_MKFS), ("stat", STUB_STAT)):
|
|
f = self.bin / name; f.write_text(body); f.chmod(0o755)
|
|
self.cfg = r / "storage.cfg"; self.cfg.write_text(cfg)
|
|
self.state = r / "state.json"
|
|
st = {"vg": "pve", "lvs": [{"name": "data", "pool": "", "size": pool_gib * GIB}] + [dict(l) for l in lvs],
|
|
"fs": fs or {}}
|
|
st.update(extra)
|
|
self.state.write_text(json.dumps(st))
|
|
|
|
def run(self, *args):
|
|
env = {"PATH": f"{self.bin}:/usr/bin:/bin", "IP_STATE": str(self.state), "FELHOM_STORAGE_CFG": str(self.cfg)}
|
|
p = subprocess.run(["bash", str(HELPER), *args], env=env, capture_output=True, text=True)
|
|
return p.returncode, p.stdout.strip(), p.stderr.strip()
|
|
|
|
def st(self):
|
|
return json.loads(self.state.read_text())
|
|
|
|
def calls(self):
|
|
return self.st().get("calls", [])
|
|
|
|
def close(self):
|
|
self.t.cleanup()
|
|
|
|
|
|
def thin(name, gib, pool="data"):
|
|
return {"name": name, "pool": pool, "size": gib * GIB}
|
|
|
|
|
|
class Sizes(unittest.TestCase):
|
|
"""The operator's table (SPACE-CHECK §4, ruled 2026-10-16), as the helper computes it on a fresh box."""
|
|
|
|
def plan(self, pool, planned, lvs=()):
|
|
b = Box(pool, lvs)
|
|
try:
|
|
rc, out, err = b.run("plan", str(planned))
|
|
self.assertEqual(rc, 0, err)
|
|
self.assertEqual(b.calls(), [], "plan must change nothing")
|
|
return dict(kv.split("=") for kv in out.split()[1:])
|
|
finally:
|
|
b.close()
|
|
|
|
def test_the_ruled_table(self):
|
|
# disk → pool (GiB), data volume (planned = 32 + mp0), expected internal
|
|
for pool, mp0, reserve, internal in ((140, 50, 16, 42), (349, 100, 34, 183), (800, 130, 48, 590), (1720, 130, 48, 1510)):
|
|
got = self.plan(pool, 32 + mp0)
|
|
self.assertEqual((int(got["reserve"]), int(got["internal"])), (reserve, internal), (pool, got))
|
|
|
|
def test_promised_volumes_count_but_vzdump_snapshots_and_other_pools_do_not(self):
|
|
got = self.plan(349, 0, [thin("vm-9201-disk-0", 32), thin("vm-9201-disk-1", 250),
|
|
thin("snap_vm-9201-disk-1_vzdump", 250), thin("vm-500-disk-0", 40, pool="other")])
|
|
self.assertEqual((int(got["promised"]), int(got["internal"])), (282, 33), got)
|
|
|
|
|
|
class Create(unittest.TestCase):
|
|
def test_a_fresh_box_gets_the_place(self):
|
|
b = Box(349)
|
|
try:
|
|
rc, out, err = b.run("create", "132")
|
|
self.assertEqual(rc, 0, err)
|
|
self.assertRegex(out, r"^created uuid=[0-9a-f-]{36} gib=183$")
|
|
self.assertEqual(len([c for c in b.calls() if c.startswith("lvcreate")]), 1)
|
|
self.assertIn("-V 183G -T pve/data -n felhom-belso", b.calls()[0])
|
|
self.assertIn("-L felhom-belso", b.calls()[1])
|
|
finally:
|
|
b.close()
|
|
|
|
def test_too_little_space_makes_nothing(self):
|
|
b = Box(53) # demo-hp's 128 GB disk
|
|
try:
|
|
rc, out, err = b.run("create", str(32 + 50))
|
|
self.assertEqual(rc, 0, err)
|
|
self.assertTrue(out.startswith("skipped "), out)
|
|
self.assertEqual(b.calls(), [])
|
|
finally:
|
|
b.close()
|
|
|
|
def test_an_existing_place_is_found_never_formatted(self):
|
|
fs = {"felhom-belso": {"TYPE": "ext4", "LABEL": "felhom-belso", "UUID": "11111111-2222-3333-4444-555555555555"}}
|
|
b = Box(349, [thin("felhom-belso", 183)], fs)
|
|
try:
|
|
rc, out, err = b.run("create", "0")
|
|
self.assertEqual(rc, 0, err)
|
|
self.assertEqual(out, "existing uuid=11111111-2222-3333-4444-555555555555 gib=183")
|
|
self.assertEqual(b.calls(), [], "an existing place must never be created or formatted again")
|
|
finally:
|
|
b.close()
|
|
|
|
def test_a_foreign_filesystem_on_the_name_is_refused(self):
|
|
# positive control: test_an_existing_place_is_found_never_formatted (same LV, Felhom's label) goes through
|
|
for fs in ({"TYPE": "xfs", "LABEL": "felhom-belso", "UUID": "x"}, {"TYPE": "ext4", "LABEL": "films", "UUID": "y"}):
|
|
b = Box(349, [thin("felhom-belso", 183)], {"felhom-belso": fs})
|
|
try:
|
|
rc, out, err = b.run("create", "0")
|
|
self.assertEqual(rc, 1, (fs, out))
|
|
self.assertIn("REFUSED", err)
|
|
self.assertEqual(b.calls(), [], fs)
|
|
finally:
|
|
b.close()
|
|
|
|
def test_a_crash_between_lvcreate_and_mkfs_is_finished(self):
|
|
b = Box(349, [thin("felhom-belso", 183)], {})
|
|
try:
|
|
rc, out, err = b.run("create", "0")
|
|
self.assertEqual(rc, 0, err)
|
|
self.assertRegex(out, r"^formatted uuid=")
|
|
self.assertEqual([c.split()[0] for c in b.calls()], ["mkfs.ext4"])
|
|
finally:
|
|
b.close()
|
|
|
|
def test_an_unnamed_signature_is_refused(self):
|
|
b = Box(349, [thin("felhom-belso", 183)], {"felhom-belso": {"RAW": "something"}})
|
|
try:
|
|
rc, out, err = b.run("create", "0")
|
|
self.assertEqual(rc, 1, out)
|
|
self.assertEqual(b.calls(), [])
|
|
finally:
|
|
b.close()
|
|
|
|
def test_a_new_volume_that_is_not_empty_is_not_formatted(self):
|
|
b = Box(349, dirty_new=True)
|
|
try:
|
|
rc, out, err = b.run("create", "132")
|
|
self.assertEqual(rc, 1, out)
|
|
self.assertNotIn("mkfs", " ".join(b.calls()))
|
|
finally:
|
|
b.close()
|
|
|
|
def test_bad_arguments_and_no_local_lvm_are_refused(self):
|
|
b = Box(349, cfg="dir: local\n\tpath /var/lib/vz\n")
|
|
try:
|
|
self.assertEqual(b.run("create", "0")[0], 1)
|
|
self.assertEqual(b.calls(), [])
|
|
finally:
|
|
b.close()
|
|
b = Box(349)
|
|
try:
|
|
for args in (("create", "-1"), ("create", "1x"), ("wipe", "0"), ("create",)):
|
|
self.assertEqual(b.run(*args)[0], 1, args)
|
|
self.assertEqual(b.calls(), [])
|
|
finally:
|
|
b.close()
|
|
|
|
|
|
class FailsClosed(unittest.TestCase):
|
|
"""A read that fails must never read as „empty" or „nothing promised". Positive control: Create (same setups)."""
|
|
|
|
def test_a_failed_volume_listing_creates_nothing(self):
|
|
b = Box(349, lvs_fail_listing=True)
|
|
try:
|
|
rc, out, err = b.run("create", "132")
|
|
self.assertEqual(rc, 1, out)
|
|
self.assertEqual(b.calls(), [])
|
|
finally:
|
|
b.close()
|
|
|
|
def test_a_listing_without_the_pool_creates_nothing(self):
|
|
b = Box(349, [thin("vm-9201-disk-1", 250)], hide_pool_in_listing=True)
|
|
try:
|
|
self.assertEqual(b.run("create", "132")[0], 1)
|
|
self.assertEqual(b.calls(), [])
|
|
finally:
|
|
b.close()
|
|
|
|
def test_a_blkid_error_never_reads_as_empty(self):
|
|
for setup in ({"lvs": [thin("felhom-belso", 183)], "fs": {}}, {}):
|
|
b = Box(349, **({"lvs": setup["lvs"], "fs": setup["fs"]} if setup else {}), blkid_rc=4)
|
|
try:
|
|
rc, out, err = b.run("create", "0" if setup else "132")
|
|
self.assertEqual(rc, 1, (setup, out))
|
|
self.assertNotIn("mkfs", " ".join(b.calls()), setup)
|
|
finally:
|
|
b.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|