"""Behavioral tests for the UUID-pinned M9 resume-only procedure.""" from __future__ import annotations import importlib.util import hashlib import json import os from pathlib import Path import subprocess import sys import tempfile import unittest from unittest import mock HERE = Path(__file__).parent sys.path.insert(0, str(HERE)) SPEC = importlib.util.spec_from_file_location("vm_resume", HERE / "resume.py") assert SPEC and SPEC.loader resume = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(resume) class FakeVBox: def __init__(self, root: Path, iso: Path, *, mutate: dict[str, str] | None = None): self.root = root self.iso = iso self.target_uuid = resume.TARGET_UUID self.description = "" self.machine_mutations = dict(mutate or {}) self.calls: list[tuple[str, ...]] = [] self.mutations: list[tuple[str, ...]] = [] self.inventory = f'"{resume.VM_NAME}" {{{self.target_uuid}}}\n' self.fail_command: str | None = None def __call__(self, argv): args = tuple(argv) self.calls.append(args) command = args[1] if command not in {"list", "showvminfo"}: self.mutations.append(args) if command == self.fail_command: return subprocess.CompletedProcess(args, 1, "", "synthetic failure detail") if args[1:3] == ("list", "vms"): return subprocess.CompletedProcess(args, 0, self.inventory, "") if command == "modifyvm" and "--description" in args: self.description = args[args.index("--description") + 1] if command == "modifyvm" and "--memory" in args: self.machine_mutations.update({ "memory": "1024", "cpus": "1", "nic1": "nat", **{f"nic{slot}": "none" for slot in range(2, 9)}, }) if command == "createmedium": Path(args[args.index("--filename") + 1]).write_bytes(b"synthetic VDI") if command == "storagectl": self.machine_mutations["storagecontrollername0"] = args[args.index("--name") + 1] if command == "storageattach": port = args[args.index("--port") + 1] medium = args[args.index("--medium") + 1] self.machine_mutations[f"SATA-{port}-0"] = medium if command == "showvminfo": info = { "name": resume.VM_NAME, "UUID": self.target_uuid, "description": self.description, "CfgFile": str(self.root / resume.VM_NAME / f"{resume.VM_NAME}.vbox"), "VMState": "poweroff", "memory": "1024", "cpus": "1", "nic1": "nat", **{f"nic{slot}": "none" for slot in range(2, 9)}, **self.machine_mutations, } output = "\n".join(f"{key}={json.dumps(value)}" for key, value in info.items()) + "\n" return subprocess.CompletedProcess(args, 0, output, "") return subprocess.CompletedProcess(args, 0, "", "") class ResumeTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) / "lab" self.root.mkdir(mode=0o700) os.chmod(self.root, 0o700) self.vm_dir = self.root / resume.VM_NAME self.vm_dir.mkdir() (self.vm_dir / f"{resume.VM_NAME}.vbox").write_text("synthetic", encoding="ascii") self.iso = self.root / "debian-13.7.0-amd64-netinst.iso" self.iso.write_bytes(b"synthetic ISO") self.digest = hashlib.sha256(self.iso.read_bytes()).hexdigest() self.vbox = FakeVBox(self.root, self.iso) self.constants = mock.patch.multiple( resume, LAB_ROOT=self.root, ISO_PATH=self.iso, PINNED_ISO_SHA256=self.digest, ) self.constants.start() self.guard_digest = mock.patch.object(resume.guard, "EXPECTED_ISO_SHA256", self.digest) self.guard_digest.start() self.addCleanup(self.constants.stop) self.addCleanup(self.guard_digest.stop) def tearDown(self): self.temp.cleanup() def run_resume(self, runner=None): return resume.resume_target( lab_root=self.root, iso_path=self.iso, runner=runner or self.vbox, ) def test_resume_configures_only_pinned_partial_vm_and_finishes_guard(self): selected = self.run_resume() self.assertEqual(selected, resume.TARGET_UUID) self.assertEqual(self.vbox.calls[0][1:3], ("list", "vms")) self.assertEqual(self.vbox.calls[1][1:4], ( "showvminfo", resume.TARGET_UUID, "--machinereadable")) modify = next(call for call in self.vbox.mutations if call[1] == "modifyvm" and "--memory" in call) self.assertEqual(modify[modify.index("--nic1") + 1], "nat") for slot in range(2, 9): self.assertEqual(modify[modify.index(f"--nic{slot}") + 1], "none") disk_call = next(call for call in self.vbox.mutations if call[1] == "createmedium") self.assertEqual(disk_call[disk_call.index("--size") + 1], "8192") self.assertTrue((self.vm_dir / f"{resume.VM_NAME}.vdi").is_file()) manifest_path = self.root / resume.guard.IDENTITY_FILE manifest = json.loads(manifest_path.read_text(encoding="utf-8")) self.assertEqual(manifest["uuid"], resume.TARGET_UUID) self.assertEqual(manifest["host_marker"], f"{resume.VM_NAME}:{resume.TARGET_UUID}") self.assertEqual(manifest_path.stat().st_mode & 0o777, 0o600) self.assertFalse(any(call[1] in { "createvm", "startvm", "unregistervm", "deletevm", "closevm", } for call in self.vbox.calls)) self.assertEqual(self.vbox.calls[-2][1:], ("list", "vms")) self.assertEqual(self.vbox.calls[-1][1], "showvminfo") def test_preflight_failures_make_no_mutating_calls(self): cases = ( ("digest", "ISO SHA-256"), ("inventory", "inventory does not match"), ("manifest", "manifest already exists"), ("disk", "VDI already exists"), ("running", "powered off"), ("network", "NIC state"), ("primary_nic", "NIC state"), ("forwarding", "NAT forwarding"), ("forwarding_record", "NAT forwarding"), ("shared", "shared folders"), ("storage", "attached storage"), ("controller", "storage controller"), ("description", "host marker"), ("memory", "default memory or CPU"), ) for case, message in cases: with self.subTest(case=case): for stale in ( self.root / resume.guard.IDENTITY_FILE, self.vm_dir / f"{resume.VM_NAME}.vdi", ): if stale.exists(): stale.unlink() fake = FakeVBox(self.root, self.iso) if case == "inventory": fake.inventory = '"other" {00000000-0000-0000-0000-000000000001}\n' elif case == "manifest": (self.root / resume.guard.IDENTITY_FILE).write_text("occupied", encoding="ascii") elif case == "disk": (self.vm_dir / f"{resume.VM_NAME}.vdi").write_bytes(b"occupied") elif case == "running": fake.machine_mutations["VMState"] = "running" elif case == "network": fake.machine_mutations["nic2"] = "hostonly" elif case == "primary_nic": fake.machine_mutations["nic1"] = "none" elif case == "forwarding": fake.machine_mutations["natpf1"] = "ssh,tcp,,2222,,22" elif case == "forwarding_record": fake.machine_mutations["Forwarding(0)"] = "ssh,tcp,,2222,,22" elif case == "shared": fake.machine_mutations["SharedFolderPathMachineMapping1"] = "/" elif case == "storage": fake.machine_mutations["SATA-0-0"] = str(self.root / "foreign.vdi") elif case == "controller": fake.machine_mutations["storagecontrollername1"] = "SATA" elif case == "description": fake.description = "unreviewed marker" elif case == "memory": fake.machine_mutations["memory"] = "2048" with self.subTest(case=case), self.assertRaisesRegex(resume.ResumeError, message): if case == "digest": with (mock.patch.object(resume, "PINNED_ISO_SHA256", "0" * 64), mock.patch.object(resume.guard, "EXPECTED_ISO_SHA256", "0" * 64)): self.run_resume(fake) else: self.run_resume(fake) self.assertEqual(fake.mutations, []) def test_mutation_failure_reports_exact_partial_uuid_and_does_not_clean_up(self): fake = FakeVBox(self.root, self.iso) fake.fail_command = "storagectl" with self.assertRaisesRegex(resume.ResumeError, resume.TARGET_UUID): self.run_resume(fake) self.assertFalse((self.root / resume.guard.IDENTITY_FILE).exists()) self.assertFalse(any(call[1] in {"unregistervm", "deletevm", "closevm"} for call in fake.calls)) self.assertNotIn("synthetic failure detail", str(fake.calls)) if __name__ == "__main__": unittest.main()