"""Behavioral tests for the M9 create-only VirtualBox procedure.""" from __future__ import annotations import importlib.util import json import os from pathlib import Path import subprocess import sys import tempfile import unittest from unittest import mock import uuid HERE = Path(__file__).parent sys.path.insert(0, str(HERE)) SPEC = importlib.util.spec_from_file_location("vm_create", HERE / "create.py") assert SPEC and SPEC.loader create = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(create) class FakeVBox: def __init__(self, root: Path, iso: Path, *, inventory: str = ""): self.root = root self.iso = iso self.inventory = inventory self.target_uuid = "1fd553db-3b0e-43e8-b407-244ea5f8c5c8" self.description = "" self.storage_controllers: dict[str, str] = {} self.calls: list[tuple[str, ...]] = [] self.mutations: list[tuple[str, ...]] = [] 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 args[1:3] == ("list", "vms"): return subprocess.CompletedProcess(args, 0, self.inventory, "") if command == "createvm": folder = self.root / create.VM_NAME folder.mkdir() (folder / f"{create.VM_NAME}.vbox").write_text("synthetic", encoding="ascii") self.inventory = f'"{create.VM_NAME}" {{{self.target_uuid}}}\n' return subprocess.CompletedProcess(args, 0, f"VM created.\nUUID: {self.target_uuid}\n", "") if command == "createmedium": Path(args[args.index("--filename") + 1]).write_bytes(b"synthetic disk") return subprocess.CompletedProcess(args, 0, "", "") if command == "storagectl": self.storage_controllers["storagecontrollername0"] = args[args.index("--name") + 1] return subprocess.CompletedProcess(args, 0, "", "") if command == "modifyvm" and "--description" in args: self.description = args[args.index("--description") + 1] return subprocess.CompletedProcess(args, 0, "", "") if command == "showvminfo": info = { "name": create.VM_NAME, "UUID": self.target_uuid, "description": self.description, "CfgFile": str(self.root / create.VM_NAME / f"{create.VM_NAME}.vbox"), "nic1": "nat", **{f"nic{slot}": "none" for slot in range(2, 9)}, "natpf1": "", "SATA-0-0": str(self.root / create.VM_NAME / f"{create.VM_NAME}.vdi"), "SATA-1-0": str(self.iso), **self.storage_controllers, } 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 CreateTests(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.iso = self.root / create.guard.EXPECTED_ISO self.iso.write_bytes(b"synthetic ISO") self.digest = __import__("hashlib").sha256(self.iso.read_bytes()).hexdigest() self.vbox = FakeVBox(self.root, self.iso) def tearDown(self): self.temp.cleanup() def run_create(self, runner=None): with mock.patch.object(create.guard, "EXPECTED_ISO_SHA256", self.digest): return create.create_target( lab_root=self.root, iso_path=self.iso, runner=runner or self.vbox, ) def test_create_is_scoped_and_guarded_with_mode_0600_manifest(self): target_uuid = self.run_create() self.assertEqual(target_uuid, self.vbox.target_uuid) self.assertEqual(self.vbox.calls[2][1:4], ("showvminfo", target_uuid, "--machinereadable")) manifest_path = self.root / create.guard.IDENTITY_FILE manifest = json.loads(manifest_path.read_text(encoding="utf-8")) self.assertEqual(manifest["name"], create.VM_NAME) self.assertEqual(manifest["uuid"], target_uuid) self.assertEqual(manifest["host_marker"], f"{create.VM_NAME}:{target_uuid}") self.assertEqual(manifest_path.stat().st_mode & 0o777, 0o600) createvm = self.vbox.mutations[0] self.assertEqual(createvm[:2], ("VBoxManage", "createvm")) self.assertEqual(createvm[createvm.index("--name") + 1], create.VM_NAME) self.assertEqual(createvm[createvm.index("--ostype") + 1], "Debian13_64") self.assertEqual(createvm[createvm.index("--basefolder") + 1], str(self.root)) modify = next(call for call in self.vbox.mutations if call[1] == "modifyvm" and "--memory" in call) self.assertEqual(modify[modify.index("--memory") + 1], "1024") self.assertEqual(modify[modify.index("--cpus") + 1], "1") self.assertEqual(modify[modify.index("--nic1") + 1], "nat") for slot in range(2, 9): self.assertEqual(modify[modify.index(f"--nic{slot}") + 1], "none") self.assertEqual(modify[modify.index("--audio") + 1], "none") self.assertEqual(modify[modify.index("--usb") + 1], "off") self.assertEqual(modify[modify.index("--boot1") + 1], "dvd") self.assertEqual(modify[modify.index("--boot2") + 1], "disk") 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.assertFalse(any(call[1] in {"startvm", "unregistervm", "snapshot"} 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 = ( ("bad-digest", "ISO SHA-256"), ("registered", "already registered"), ("malformed", "malformed VBoxManage inventory"), ("manifest", "manifest already exists"), ) for case, message in cases: with self.subTest(case=case): fake = FakeVBox(self.root, self.iso) iso_path = self.iso if case == "bad-digest": with mock.patch.object(create.guard, "EXPECTED_ISO_SHA256", "0" * 64): with self.assertRaisesRegex(create.CreateError, message): create.create_target(lab_root=self.root, iso_path=iso_path, runner=fake) self.assertEqual(fake.calls, []) continue if case == "registered": fake.inventory = f'"{create.VM_NAME}" {{{fake.target_uuid}}}\n' elif case == "malformed": fake.inventory = "not a machine-readable row\n" else: (self.root / create.guard.IDENTITY_FILE).write_text("exists", encoding="ascii") with mock.patch.object(create.guard, "EXPECTED_ISO_SHA256", self.digest): with self.assertRaisesRegex(create.CreateError, message): create.create_target(lab_root=self.root, iso_path=iso_path, runner=fake) self.assertEqual(fake.mutations, []) def test_occupied_manifest_and_target_paths_make_no_vbox_calls(self): cases = ( ("manifest-file", "manifest already exists", "manifest", False), ("manifest-dangling-symlink", "manifest already exists", "manifest", True), ("target-directory", "exact target VM path already exists", "target", False), ("target-dangling-symlink", "exact target VM path already exists", "target", True), ) for case, message, occupied_kind, dangling in cases: with self.subTest(case=case): fake = FakeVBox(self.root, self.iso) occupied_path = ( self.root / create.guard.IDENTITY_FILE if occupied_kind == "manifest" else self.root / create.VM_NAME ) if dangling: occupied_path.symlink_to(self.root / "missing-target") elif occupied_kind == "manifest": occupied_path.write_text("occupied", encoding="ascii") else: occupied_path.mkdir() with mock.patch.object(create.guard, "EXPECTED_ISO_SHA256", self.digest): with self.assertRaisesRegex(create.CreateError, message): create.create_target(lab_root=self.root, iso_path=self.iso, runner=fake) self.assertEqual(fake.calls, []) if dangling or occupied_kind == "manifest": occupied_path.unlink() else: occupied_path.rmdir() def test_failed_configuration_retains_partial_identity_and_does_not_cleanup(self): class FailingVBox(FakeVBox): def __call__(inner, argv): if len(argv) > 1 and argv[1] == "storagectl": args = tuple(argv) inner.calls.append(args) inner.mutations.append(args) return subprocess.CompletedProcess(args, 1, "", "sensitive detail omitted") return super(FailingVBox, inner).__call__(argv) fake = FailingVBox(self.root, self.iso) with mock.patch.object(create.guard, "EXPECTED_ISO_SHA256", self.digest): with self.assertRaisesRegex(create.CreateError, self.vbox.target_uuid): create.create_target(lab_root=self.root, iso_path=self.iso, runner=fake) self.assertTrue((self.root / create.VM_NAME / f"{create.VM_NAME}.vbox").exists()) self.assertFalse(any(call[1] in {"unregistervm", "closevm", "deletevm"} for call in fake.calls)) self.assertNotIn("sensitive detail omitted", str(fake.calls)) if __name__ == "__main__": unittest.main()