210 lines
10 KiB
Python
210 lines
10 KiB
Python
"""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()
|