from __future__ import annotations import json import shutil import tempfile import unittest from collections.abc import Callable, Iterable, Iterator from contextlib import contextmanager from dataclasses import replace from ipaddress import IPv4Address, IPv4Network from pathlib import Path from unittest.mock import patch from uvm.config import Settings from uvm.domain import State, VmRecord, VmSpec from uvm.errors import ConflictError, FirecrackerError, UvmError, ValidationError from uvm.firecracker.process import ProcessInfo from uvm.images import ImageStore from uvm.lifecycle import LifecycleService from uvm.network import NetworkManager from uvm.state import StateStore class TrackedStateStore(StateStore): def __init__(self, settings: Settings) -> None: super().__init__(settings) self.operation_active = False self.transaction_active = False self.operation_locks = 0 @contextmanager def operation_lock(self) -> Iterator[None]: if self.operation_active or self.transaction_active: raise AssertionError("nested or inverted operation lock") self.operation_locks += 1 with super().operation_lock(): self.operation_active = True try: yield finally: self.operation_active = False @contextmanager def transaction(self) -> Iterator[State]: if self.transaction_active: raise AssertionError("nested state transaction") self.transaction_active = True try: with super().transaction() as state: yield state finally: self.transaction_active = False class FakeHost: """Record boundary calls and inject one-shot failures before their effects.""" def __init__(self, store: TrackedStateStore) -> None: self.store = store self.events: list[tuple[str, str | None]] = [] self.records: dict[tuple[str, str], VmRecord] = {} self.failures: dict[tuple[str, str | None], BaseException] = {} def call(self, action: str, vm_id: str | None = None, *, readonly: bool = False) -> None: if self.store.transaction_active: raise AssertionError(f"state transaction held during {action}") if not readonly and not self.store.operation_active: raise AssertionError(f"operation lock missing during {action}") self.events.append((action, vm_id)) if vm_id is not None: vm = self.store.load().vms.get(vm_id) if vm is not None: self.records[action, vm_id] = vm error = self.failures.pop((action, vm_id), None) if error is None: error = self.failures.pop((action, None), None) if error is not None: raise error def actions(self, vm_id: str) -> list[str]: return [action for action, target in self.events if target == vm_id] class FakeKnownHosts: def __init__(self, host: FakeHost) -> None: self.host = host def remove(self, vm: VmRecord) -> None: self.host.call("known_hosts.remove", vm.id) if Path(vm.socket).parent.exists(): raise AssertionError("known_hosts cleanup must follow VM runtime deletion") class FakeFirewall: def __init__(self, host: FakeHost, network: FakeNetwork) -> None: self.host = host self.network = network self.tables: dict[str, tuple[VmRecord, bool]] = {} def check_available(self) -> None: self.host.call("firewall.check_available", readonly=True) def apply(self, vm: VmRecord, *, quarantine: bool = False) -> None: self.host.call("firewall.quarantine" if quarantine else "firewall.apply", vm.id) if vm.ports is None and not quarantine: raise ValidationError("legacy VM has no managed port policy") self.tables[vm.id] = (replace(vm), quarantine) self.host.call("firewall.readback", vm.id) def status(self, vm: VmRecord) -> str: self.host.call("firewall.status", vm.id, readonly=True) if vm.id not in self.tables: return "unmanaged" if vm.ports is None else "missing" installed, quarantine = self.tables[vm.id] if installed.tap != vm.tap: return "drift" if quarantine: return "quarantined" if vm.ports is None: return "drift" same_identity = installed.guest_ip == vm.guest_ip and installed.mac == vm.mac same_ports = {rule.port for rule in installed.ports} == {rule.port for rule in vm.ports} return "enforced" if same_identity and same_ports else "drift" def cleanup_connections(self, vm: VmRecord) -> None: self.host.call("firewall.cleanup_connections", vm.id) if vm.tap in self.network.taps: raise UvmError("cannot clean connections while TAP exists") def remove(self, vm: VmRecord) -> None: self.host.call("firewall.remove", vm.id) if vm.tap in self.network.taps: raise UvmError("cannot remove guard while TAP exists") self.tables.pop(vm.id, None) class FakeNetwork: def __init__(self, settings: Settings, host: FakeHost) -> None: self._settings = settings self.host = host self.taps: dict[str, str] = {} self.active_taps: set[str] = set() self.created_taps: list[str] = [] self.deleted_taps: list[str] = [] self.firewall = FakeFirewall(host, self) def allocate_ip(self, vms: Iterable[VmRecord], requested: IPv4Address | None) -> IPv4Address: return NetworkManager.allocate_ip(self, vms, requested) # type: ignore[arg-type] mac_for = staticmethod(NetworkManager.mac_for) tap_name = staticmethod(NetworkManager.tap_name) def ensure_bridge(self) -> None: self.host.call("network.ensure_bridge") def prepare_tap(self, vm: VmRecord) -> None: self.host.call("network.prepare_tap", vm.id) if vm.tap in self.taps: raise ConflictError(f"TAP device already exists: {vm.tap}") self.taps[vm.tap] = vm.id self.created_taps.append(vm.tap) vm.tap_owned = True def has_tap(self, vm: VmRecord) -> bool: self.host.call("network.has_tap", vm.id, readonly=True) owner = self.taps.get(vm.tap) if owner is not None and owner != vm.id: raise ConflictError(f"refusing to modify unowned TAP {vm.tap}") return owner is not None def activate_tap(self, vm: VmRecord) -> None: self.host.call("network.activate_tap", vm.id) if not self.has_tap(vm): raise UvmError(f"TAP device missing: {vm.tap}") self.active_taps.add(vm.tap) def deactivate_tap(self, vm: VmRecord) -> None: self.host.call("network.deactivate_tap", vm.id) if self.has_tap(vm): self.active_taps.discard(vm.tap) def delete_vm_tap(self, vm: VmRecord) -> None: self.host.call("network.delete_vm_tap", vm.id) if self.has_tap(vm): del self.taps[vm.tap] self.active_taps.discard(vm.tap) self.deleted_taps.append(vm.tap) vm.tap_owned = False class FakeProcess: def __init__(self, host: FakeHost) -> None: self.host = host self.boot_id = "test-host-boot" self.next_pid = 12345 self.running: dict[str, ProcessInfo] = {} self.unknown: set[str] = set() self.started: list[ProcessInfo] = [] self.terminated: list[tuple[str, int | None, float | None]] = [] def host_boot_id(self) -> str: return self.boot_id def observe(self, vm: VmRecord) -> str: self.host.call("process.observe", vm.id, readonly=True) if vm.pid is not None and vm.pid <= 0: return "unknown" if vm.boot_id and vm.boot_id != self.boot_id: return "gone" if vm.id in self.unknown: return "unknown" if vm.pid is None: if (vm.spawn_pending or vm.operation == "spawning" or vm.process_start_time is not None or Path(vm.socket).exists()): return "unknown" return "gone" info = self.running.get(vm.id) if info is None or info.pid != vm.pid: return "gone" if vm.process_start_time is not None and info.start_time != vm.process_start_time: return "gone" if vm.process_start_time is None or vm.boot_id is None: return "unknown" return "alive" def adopt_legacy_identity(self, vm: VmRecord) -> bool: self.host.call("process.adopt_legacy_identity", vm.id) info = self.running.get(vm.id) if (vm.boot_id is None and info is not None and vm.id not in self.unknown and (vm.pid, vm.process_start_time) == (info.pid, info.start_time)): vm.boot_id = info.boot_id return True return False def start(self, vm: VmRecord, *, on_spawn: Callable[[ProcessInfo], None] | None = None) -> ProcessInfo: self.host.call("process.start", vm.id) if self.observe(vm) != "gone": raise FirecrackerError("refusing to replace an existing or ambiguous process") vm.operation = "spawning" vm.spawn_pending = True vm.boot_id = self.host_boot_id() # A failure here models an interrupt inside Popen, before identity is known. self.host.call("process.spawn", vm.id) info = ProcessInfo(self.next_pid, str(self.next_pid * 10), self.boot_id) self.next_pid += 1 self.running[vm.id] = info self.started.append(info) vm.pid, vm.process_start_time = info.pid, info.start_time if on_spawn is not None: on_spawn(info) else: vm.spawn_pending = False Path(vm.socket).touch() with Path(vm.log).open("ab") as log: log.write(f"boot {info.pid}\n".encode("ascii")) self.host.call("process.ready", vm.id) return info def terminate(self, vm: VmRecord, *, timeout: float | None = None) -> None: self.terminated.append((vm.id, vm.pid, timeout)) self.host.call("process.terminate", vm.id) observation = self.observe(vm) if observation == "unknown": raise FirecrackerError("process identity is ambiguous") if observation == "alive": del self.running[vm.id] Path(vm.socket).unlink(missing_ok=True) self.host.call("process.gone", vm.id) class FakeClient: def __init__(self, host: FakeHost, vm_id: str) -> None: self.host = host self.vm_id = vm_id self.config: dict | None = None def configure_and_start(self, config: dict) -> None: self.host.call("client.configure", self.vm_id) self.config = config class LifecycleTestCase(unittest.TestCase): def setUp(self) -> None: temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) base = Path(temporary.name) / "uvm" self.settings = replace(Settings(), base=base, allow_unverified_downloads=True) self.settings.bin_dir.mkdir(parents=True) self.settings.firecracker_binary.touch() self.settings.images_dir.mkdir(parents=True) self.settings.kernel_image.write_bytes(b"kernel") self.settings.rootfs_image.write_bytes(b"rootfs") self.store = TrackedStateStore(self.settings) self.host = FakeHost(self.store) self.images = ImageStore(self.settings) self.disk_sizes: dict[Path, int] = {} self.enterContext(patch.object(self.images, "_copy_sparse", side_effect=shutil.copy2)) self.filesystem_resize = self.enterContext(patch.object( self.images, "_resize_filesystem", side_effect=lambda disk, capacity: self.disk_sizes.update({disk: capacity * 1024 * 1024}), )) self.enterContext(patch.object( self.images, "disk_size_bytes", side_effect=lambda disk: self.disk_sizes.get(disk, disk.stat().st_size), )) self.resize_disk = self.enterContext(patch.object( self.images, "resize_vm_disk", side_effect=self.resize_vm_disk, )) self.provision = self.enterContext(patch.object(self.images, "provision_credentials")) self.configure_network = self.enterContext(patch.object( self.images, "configure_guest_network", side_effect=self.configure_guest_network, )) self.provision_agent = self.enterContext(patch.object(self.images, "provision_guest_agent")) self.copy_disk = self.enterContext(patch.object( self.images, "create_vm_disk", wraps=self.images.create_vm_disk, )) self.network = FakeNetwork(self.settings, self.host) self.process = FakeProcess(self.host) self.known_hosts = FakeKnownHosts(self.host) self.clients: dict[str, FakeClient] = {} self.service = LifecycleService( self.settings, self.store, self.images, self.network, # type: ignore[arg-type] self.process, # type: ignore[arg-type] known_hosts=self.known_hosts, # type: ignore[arg-type] client_factory=self.make_client, # type: ignore[arg-type] ) self.require_root = self.enterContext(patch("uvm.lifecycle.require_root")) self.check_kvm = self.enterContext(patch("uvm.lifecycle.check_kvm")) for target in ("uvm.system.CommandRunner.run", "subprocess.run", "subprocess.Popen", "os.kill"): self.enterContext(patch(target, side_effect=AssertionError(f"unexpected host operation: {target}"))) def make_client(self, socket: Path, _timeout: float) -> FakeClient: client = FakeClient(self.host, socket.parent.name) self.clients[client.vm_id] = client return client def resize_vm_disk(self, disk: Path, storage_mib: int) -> None: vm_id = disk.parent.name self.host.call("images.resize_disk", vm_id) if self.process.observe(self.store.get(vm_id)) != "gone": raise AssertionError("cannot resize a running or ambiguous VM") self.disk_sizes[disk] = storage_mib * 1024 * 1024 def configure_guest_network(self, disk: Path, guest_ip: str) -> None: vm = self.store.get(disk.parent.name) self.host.call("images.configure_network", vm.id) if self.process.observe(vm) != "gone": raise AssertionError("cannot configure a running or ambiguous VM disk") if guest_ip != vm.guest_ip: raise AssertionError("guest network must use the VM's retained address") def create_vm(self, **kwargs) -> VmRecord: return self.service.create(VmSpec(cpu=1, ram_mib=512, **kwargs)) def persist(self, vm: VmRecord, **changes) -> VmRecord: with self.store.transaction() as state: updated = replace(state.vms[vm.id], **changes) state.vms[vm.id] = updated return updated def assert_call_order(self, vm: VmRecord, *expected: str) -> None: actual = [action for action in self.host.actions(vm.id) if action in expected] self.assertEqual(actual, list(expected)) class LifecycleTests(LifecycleTestCase): def test_create_uses_a_private_disk_and_persists_running_state(self) -> None: vm = self.service.create(VmSpec(cpu=1, ram_mib=512)) self.assertEqual(vm.status, "running") self.assertNotEqual(Path(vm.disk), self.settings.rootfs_image) self.assertEqual(Path(vm.disk).read_bytes(), b"rootfs") self.assertEqual(vm.username, "root") self.assertEqual(vm.password, "root") self.assertEqual(self.store.load().vms[vm.id].pid, 12345) self.assertEqual(self.network.created_taps, [vm.tap]) self.assertEqual(self.clients[vm.id].config["drives"][0]["path_on_host"], vm.disk) self.assertTrue(vm.runtime_owned) self.assertTrue(vm.tap_owned) self.assertTrue(vm.disk_ready) self.assertFalse(vm.spawn_pending) self.assertEqual(self.process.observe(vm), "alive") self.assertEqual(self.network.firewall.status(vm), "enforced") self.provision.assert_called_once_with( Path(vm.disk), "root", "root", ) def test_create_persists_and_provisions_custom_guest_credentials(self) -> None: vm = self.service.create( VmSpec(cpu=1, ram_mib=512, username="admin", password="secret-value") ) persisted = self.store.load().vms[vm.id] self.assertEqual(persisted.username, "admin") self.assertEqual(persisted.password, "secret-value") self.provision.assert_called_once_with( Path(vm.disk), "admin", "secret-value", ) def test_create_rolls_back_when_firecracker_configuration_fails(self) -> None: self.host.failures["client.configure", None] = FirecrackerError("simulated API failure") with self.assertRaisesRegex(UvmError, r"failed to create vm-.*: simulated API failure") as raised: self.service.create(VmSpec(cpu=1, ram_mib=512)) self.assertIsInstance(raised.exception.__cause__, FirecrackerError) self.assertEqual(self.store.load().vms, {}) self.assertEqual(len(self.network.deleted_taps), 1) self.assertEqual([pid for _, pid, _ in self.process.terminated], [12345]) self.assertEqual(self.network.firewall.tables, {}) self.assertEqual(self.process.running, {}) self.assertEqual(list(self.settings.vms_dir.iterdir()), []) def test_create_rolls_back_before_network_setup_when_provisioning_fails(self) -> None: self.provision.side_effect = UvmError("invalid guest image") with self.assertRaisesRegex(UvmError, r"failed to create vm-.*: invalid guest image"): self.service.create(VmSpec(cpu=1, ram_mib=512)) self.assertEqual(self.store.load().vms, {}) self.assertEqual(self.network.created_taps, []) self.assertFalse(self.process.started) self.assertEqual(list(self.settings.vms_dir.iterdir()), []) def test_create_rolls_back_when_interrupted(self) -> None: self.host.failures["network.prepare_tap", None] = KeyboardInterrupt() with self.assertRaises(KeyboardInterrupt): self.service.create(VmSpec(cpu=1, ram_mib=512)) self.assertEqual(self.store.load().vms, {}) self.assertEqual(list(self.settings.vms_dir.iterdir()), []) def test_stop_keeps_disk_and_destroy_releases_state(self) -> None: vm = self.service.create(VmSpec(cpu=1, ram_mib=512)) disk = Path(vm.disk) stopped = self.service.stop(vm.id) self.assertEqual(stopped.status, "stopped") self.assertTrue(disk.exists()) self.assertIsNone(stopped.pid) destroyed = self.service.destroy(vm.id) self.assertEqual(destroyed.id, vm.id) self.assertFalse(self.settings.vm_dir(vm.id).exists()) self.assertEqual(self.store.load().vms, {}) def test_stop_edit_disk_start_preserves_guest_data_credentials_and_identity(self) -> None: vm = self.create_vm(name="database", username="admin", password="original-secret") original = replace(vm) disk = Path(vm.disk) inode = disk.stat().st_ino stopped = self.service.stop(vm.name, timeout=0.25) self.assertEqual(self.process.terminated[-1], (vm.id, vm.pid, 0.25)) self.assertEqual(stopped.status, "stopped") self.assertNotIn(vm.tap, self.network.taps) self.assertNotIn(vm.id, self.network.firewall.tables) disk.write_bytes(b"guest-edited-data\npassword-hash\noriginal-ssh-host-keys\n") edited = disk.read_bytes() started = self.service.start(vm.guest_ip) for field in ("id", "name", "guest_ip", "mac", "tap", "disk", "username", "password", "ports", "created_at"): self.assertEqual(getattr(started, field), getattr(original, field), field) self.assertNotEqual(started.pid, original.pid) self.assertNotEqual(started.process_start_time, original.process_start_time) self.assertEqual(started.boot_id, self.process.host_boot_id()) self.assertEqual(disk.stat().st_ino, inode) self.assertEqual(disk.read_bytes(), edited) self.assertEqual(self.settings.rootfs_image.read_bytes(), b"rootfs") self.provision.assert_called_once_with(disk, "admin", "original-secret") self.copy_disk.assert_called_once() self.assertEqual(self.store.get(vm.id).to_dict(), started.to_dict()) def test_start_uses_private_disk_when_rootfs_template_has_been_removed(self) -> None: vm = self.create_vm() self.service.stop(vm.id) self.settings.rootfs_image.unlink() with patch.object(self.images, "installed_assets", side_effect=AssertionError("start needs only the kernel")): started = self.service.start(vm.id) self.assertEqual(started.status, "running") self.assertEqual(Path(started.disk).read_bytes(), b"rootfs") self.assertEqual(self.provision.call_count, 1) self.assertEqual(self.copy_disk.call_count, 1) def test_boot_persists_intent_and_identity_with_policy_before_tap_and_vmm(self) -> None: vm = self.create_vm() self.assert_call_order( vm, "images.configure_network", "network.prepare_tap", "firewall.apply", "firewall.readback", "network.activate_tap", "process.start", "process.spawn", "process.ready", "client.configure", "firewall.status", ) policy_record = self.host.records["firewall.apply", vm.id] self.assertTrue(policy_record.disk_ready) self.assertTrue(policy_record.tap_owned) pending = self.host.records["process.start", vm.id] self.assertTrue(pending.spawn_pending) self.assertEqual(pending.operation, "spawning") self.assertIsNone(pending.pid) spawned = self.host.records["process.ready", vm.id] self.assertEqual((spawned.pid, spawned.process_start_time, spawned.boot_id), (vm.pid, vm.process_start_time, vm.boot_id)) self.assertFalse(spawned.spawn_pending) self.assertEqual(spawned.operation, "configuring") self.assertEqual(self.host.records["client.configure", vm.id].pid, vm.pid) def test_guest_network_configuration_failure_rolls_back_before_marking_the_disk_ready(self) -> None: self.host.failures["images.configure_network", None] = UvmError("guest network verification failed") with self.assertRaisesRegex(UvmError, "guest network verification failed"): self.create_vm() self.assertEqual(self.store.load().vms, {}) self.assertFalse(self.network.created_taps) self.assertFalse(self.process.started) self.assertEqual(list(self.settings.vms_dir.iterdir()), []) def test_retained_guest_network_is_configured_after_exit_and_before_boot(self) -> None: vm = self.create_vm() self.host.events.clear() self.service.restart(vm.id) self.assert_call_order(vm, "images.configure_network", "process.start", "client.configure") actions = self.host.actions(vm.id) self.assertLess(max(index for index, action in enumerate(actions) if action == "process.gone"), actions.index("images.configure_network")) self.configure_network.assert_called_with(Path(vm.disk), vm.guest_ip) self.assertEqual(self.provision.call_count, 1) def test_live_start_does_not_modify_the_guest_disk(self) -> None: vm = self.create_vm() self.configure_network.reset_mock() self.service.start(vm.id) self.configure_network.assert_not_called() def test_failed_retained_network_configuration_keeps_disk_for_retry(self) -> None: vm = self.create_vm() self.service.stop(vm.id) disk = Path(vm.disk) disk.write_bytes(b"retained data and credentials") self.host.failures["images.configure_network", vm.id] = UvmError("network verification failed") with self.assertRaisesRegex(UvmError, "network verification failed"): self.service.start(vm.id) self.assertEqual(disk.read_bytes(), b"retained data and credentials") self.assertEqual(self.process.observe(self.store.get(vm.id)), "gone") self.assertEqual(len(self.process.started), 1) self.assertEqual(self.service.start(vm.id).status, "running") self.assertEqual(self.provision.call_count, 1) def test_failed_initial_policy_never_activates_tap_or_starts_vmm(self) -> None: self.host.failures["firewall.apply", None] = UvmError("policy unavailable") with self.assertRaisesRegex(UvmError, "policy unavailable"): self.create_vm() self.assertFalse(any(action in {"network.activate_tap", "process.start", "client.configure"} for action, _ in self.host.events)) self.assertEqual(self.store.load().vms, {}) self.assertEqual(self.network.taps, {}) self.assertEqual(self.network.firewall.tables, {}) def test_unavailable_firewall_fails_preflight_before_reserving_or_copying(self) -> None: self.host.failures["firewall.check_available", None] = UvmError("bridge filtering unavailable") with self.assertRaisesRegex(UvmError, "bridge filtering unavailable"): self.create_vm() self.assertEqual(self.store.load().vms, {}) self.assertEqual(self.host.events, [("firewall.check_available", None)]) self.assertEqual(list(self.settings.vms_dir.iterdir()), []) self.provision.assert_not_called() self.copy_disk.assert_not_called() def test_create_rejects_retained_topology_changes_before_allocation_or_bridge_work(self) -> None: vm = self.create_vm() self.copy_disk.reset_mock() self.provision.reset_mock() changes = ( {"gateway": IPv4Address(vm.guest_ip)}, {"gateway": IPv4Address("10.42.0.10")}, {"network": IPv4Network("10.42.0.0/25")}, {"bridge": "uvm-other"}, {"gateway": IPv4Address(vm.guest_ip), "network": IPv4Network("10.42.0.0/25")}, ) for status in ("running", "stopped"): if status == "stopped": self.service.stop(vm.id) before = self.settings.state_path.read_bytes() before_directories = set(self.settings.vms_dir.iterdir()) before_taps = self.network.taps.copy() before_active = self.network.active_taps.copy() before_tables = self.network.firewall.tables.copy() for changed in changes: with self.subTest(status=status, changed=changed): settings = replace(self.settings, **changed) service = LifecycleService( settings, self.store, self.images, self.network, self.process, known_hosts=self.known_hosts, client_factory=self.make_client, ) self.host.events.clear() with ( patch.object(self.network, "_settings", settings), patch.object(self.network, "allocate_ip", wraps=self.network.allocate_ip) as allocate, self.assertRaisesRegex(ConflictError, "topology conflicts with retained VMs"), ): service.create(VmSpec(cpu=1, ram_mib=512)) allocate.assert_not_called() self.assertEqual(self.settings.state_path.read_bytes(), before) self.assertEqual(set(self.settings.vms_dir.iterdir()), before_directories) self.assertEqual(self.network.taps, before_taps) self.assertEqual(self.network.active_taps, before_active) self.assertEqual(self.network.firewall.tables, before_tables) self.assertEqual(self.host.events, [("firewall.check_available", None)]) self.copy_disk.assert_not_called() self.provision.assert_not_called() self.assertEqual(len(self.process.started), 1) self.assertEqual(Path(vm.disk).read_bytes(), b"rootfs") def test_non_rfc1918_network_fails_before_kvm_assets_or_registry_reservation(self) -> None: for network, gateway in ( ("203.0.113.0/24", "203.0.113.1"), ("100.64.0.0/24", "100.64.0.1"), ("8.8.8.0/24", "8.8.8.1"), ("10.0.0.0/7", "10.0.0.1"), ): with self.subTest(network=network): settings = replace(self.settings, network=IPv4Network(network), gateway=IPv4Address(gateway)) service = LifecycleService( settings, self.store, self.images, self.network, self.process, known_hosts=self.known_hosts, client_factory=self.make_client, ) with ( patch.object(self.network, "_settings", settings), patch.object(self.images, "installed_assets", wraps=self.images.installed_assets) as assets, patch.object(self.network, "allocate_ip", wraps=self.network.allocate_ip) as allocate, self.assertRaisesRegex(ValidationError, "RFC 1918"), ): service.create(VmSpec(cpu=1, ram_mib=512)) assets.assert_not_called() allocate.assert_not_called() self.check_kvm.assert_not_called() self.assertFalse(self.settings.state_path.exists()) self.assertEqual(self.store.load().vms, {}) self.assertEqual(list(self.settings.vms_dir.iterdir()), []) self.assertEqual(self.host.events, []) self.provision.assert_not_called() self.copy_disk.assert_not_called() def test_create_cleans_up_if_policy_readback_is_not_enforced_after_boot(self) -> None: with patch.object(self.network.firewall, "status", return_value="drift"): with self.assertRaisesRegex(UvmError, "enforcement could not be confirmed after boot"): self.create_vm() self.assertEqual(self.store.load().vms, {}) self.assertEqual(self.process.running, {}) self.assertEqual(self.network.taps, {}) self.assertEqual(self.network.firewall.tables, {}) self.assertEqual(list(self.settings.vms_dir.iterdir()), []) def test_start_rejects_missing_or_unprovisioned_disk_without_network_mutation(self) -> None: for problem in ("missing", "unprovisioned"): with self.subTest(problem=problem): vm = self.create_vm() self.service.stop(vm.id) if problem == "missing": Path(vm.disk).unlink() else: self.persist(vm, disk_ready=False) before = self.settings.state_path.read_bytes() self.host.events.clear() with self.assertRaisesRegex(UvmError, "disk"): self.service.start(vm.id) self.assertEqual(self.settings.state_path.read_bytes(), before) self.assertFalse(set(self.host.actions(vm.id)) - {"process.observe", "process.adopt_legacy_identity"}) def test_stop_quarantines_and_confirms_exit_before_releasing_network(self) -> None: vm = self.create_vm() self.host.events.clear() self.service.stop(vm.id, timeout=0) self.assert_call_order( vm, "firewall.quarantine", "network.deactivate_tap", "process.terminate", "process.gone", "network.delete_vm_tap", "firewall.cleanup_connections", "firewall.remove", ) pending = self.host.records["process.terminate", vm.id] self.assertEqual((pending.status, pending.operation), ("stopping", "stop")) self.assertEqual(self.process.terminated[-1], (vm.id, vm.pid, 0)) def test_failed_process_cleanup_keeps_disk_tap_guard_identity_and_reservation(self) -> None: for action in ("stop", "destroy", "restart"): with self.subTest(action=action): vm = self.create_vm(name=f"failed-{action}") disk = Path(vm.disk) self.host.events.clear() self.host.failures["process.terminate", vm.id] = FirecrackerError("SIGKILL exit unconfirmed") with self.assertRaisesRegex(UvmError, "process exit unconfirmed.*SIGKILL"): getattr(self.service, action)(vm.id, timeout=0.1) retained = self.store.get(vm.id) self.assertEqual(retained.status, "failed") self.assertEqual((retained.pid, retained.process_start_time), (vm.pid, vm.process_start_time)) self.assertEqual(retained.guest_ip, vm.guest_ip) self.assertEqual(disk.read_bytes(), b"rootfs") self.assertTrue(retained.runtime_owned) self.assertTrue(retained.tap_owned) self.assertEqual(self.network.taps[vm.tap], vm.id) self.assertNotIn(vm.tap, self.network.active_taps) self.assertEqual(self.network.firewall.status(retained), "quarantined") self.assertEqual(self.process.observe(retained), "alive") self.assertFalse(set(self.host.actions(vm.id)) & { "network.delete_vm_tap", "firewall.cleanup_connections", "firewall.remove", "process.start", }) def test_create_retains_resources_when_rollback_cannot_confirm_exit(self) -> None: self.host.failures["client.configure", None] = FirecrackerError("API failed") self.host.failures["process.terminate", None] = FirecrackerError("still alive") with self.assertRaisesRegex(UvmError, "failed to create.*API failed; cleanup incomplete.*still alive"): self.create_vm() vm, = self.store.load().vms.values() self.assertEqual(vm.status, "failed") self.assertEqual(self.process.observe(vm), "alive") self.assertTrue(Path(vm.disk).exists()) self.assertTrue(vm.runtime_owned) self.assertTrue(vm.tap_owned) self.assertEqual(self.network.taps[vm.tap], vm.id) self.assertEqual(self.network.firewall.status(vm), "quarantined") self.assertNotIn("network.delete_vm_tap", self.host.actions(vm.id)) self.assertNotIn("firewall.remove", self.host.actions(vm.id)) def test_uncertain_spawn_interrupt_retains_resources_and_survives_stop_retry(self) -> None: for interrupt in (KeyboardInterrupt(), SystemExit(143)): with self.subTest(interrupt=type(interrupt).__name__): self.host.failures["process.spawn", None] = interrupt previous = set(self.store.load().vms) with self.assertRaises(type(interrupt)): self.create_vm() vm, = [vm for vm in self.store.load().vms.values() if vm.id not in previous] self.assertEqual(self.process.observe(vm), "unknown") self.assertTrue(vm.spawn_pending or vm.operation == "spawning") self.assertIsNone(vm.pid) self.assertEqual(vm.status, "failed") self.assertTrue(Path(vm.disk).exists()) self.assertEqual(self.network.taps[vm.tap], vm.id) self.assertEqual(self.network.firewall.status(vm), "quarantined") self.assertNotIn(vm.tap, self.network.active_taps) for action in ("stop", "destroy"): with self.assertRaisesRegex(UvmError, "process exit unconfirmed"): getattr(self.service, action)(vm.id) retained = self.store.get(vm.id) self.assertEqual(self.process.observe(retained), "unknown") self.assertTrue(Path(vm.disk).exists()) self.assertIn(vm.tap, self.network.taps) self.assertNotIn("network.delete_vm_tap", self.host.actions(vm.id)) self.assertNotIn("firewall.remove", self.host.actions(vm.id)) def test_directory_reservation_collision_never_deletes_existing_contents(self) -> None: directory = self.settings.vm_dir("vm-existing") directory.mkdir(parents=True) disk = directory / "rootfs.ext4" disk.write_bytes(b"not-owned-by-this-create") marker = directory / "operator-data" marker.write_bytes(b"keep") before = {path.name: (path.stat().st_ino, path.read_bytes()) for path in directory.iterdir()} with patch("uvm.lifecycle.new_vm_id", return_value="vm-existing"): with self.assertRaisesRegex(UvmError, "failed to create vm-existing"): self.create_vm() self.assertEqual({path.name: (path.stat().st_ino, path.read_bytes()) for path in directory.iterdir()}, before) self.provision.assert_not_called() self.copy_disk.assert_not_called() self.assertEqual(self.network.created_taps, []) self.assertEqual(self.process.started, []) retained = self.store.get("vm-existing") self.assertFalse(retained.runtime_owned) self.assertFalse(retained.disk_ready) with self.assertRaisesRegex(UvmError, "refusing to remove unowned VM directory"): self.service.destroy(retained.id) self.assertFalse(self.store.get(retained.id).runtime_owned) self.assertEqual({path.name: (path.stat().st_ino, path.read_bytes()) for path in directory.iterdir()}, before) def test_delete_legacy_runtime_without_private_disk_preserves_shared_rootfs(self) -> None: shared_rootfs = self.settings.rootfs_image before = (shared_rootfs.stat().st_ino, shared_rootfs.read_bytes()) for disk_field in ({}, {"disk": ""}): with self.subTest(disk_field=disk_field): directory = self.settings.vm_dir("vm-legacy") directory.mkdir(parents=True, mode=0o700) raw = { "id": "vm-legacy", "cpu": 1, "ram_mib": 512, "guest_ip": "10.42.0.2", "gateway": "10.42.0.1", "tap": self.network.tap_name("vm-legacy"), "mac": self.network.mac_for(1), "socket": str(directory / "firecracker.sock"), "config": str(directory / "config.json"), "log": str(directory / "firecracker.log"), "status": "running", "pid": 12345, "process_start_time": "42", **disk_field, } Path(raw["socket"]).touch() Path(raw["log"]).write_bytes(b"legacy boot log") Path(raw["config"]).write_text(json.dumps({ "drives": [{"path_on_host": str(shared_rootfs), "is_read_only": False}], }), encoding="utf-8") self.settings.state_path.write_text(json.dumps({ "schema_version": 1, "vms": {raw["id"]: raw}, }), encoding="utf-8") self.settings.state_path.chmod(0o600) vm = self.store.get(raw["id"]) self.assertTrue(vm.runtime_owned) self.assertEqual(vm.disk, "") self.assertFalse(vm.disk_ready) self.assertEqual(self.process.observe(vm), "gone") self.service.destroy(vm.id) self.assertEqual(self.store.load().vms, {}) self.assertFalse(directory.exists()) self.assertEqual((shared_rootfs.stat().st_ino, shared_rootfs.read_bytes()), before) self.copy_disk.assert_not_called() self.provision.assert_not_called() self.check_kvm.assert_not_called() def test_stale_stopping_and_deleting_records_can_finish_cleanup(self) -> None: for action, status, operation in (("stop", "stopping", "stop"), ("destroy", "terminating", "delete")): with self.subTest(action=action): vm = self.create_vm() self.persist(vm, status=status, operation=operation) del self.process.running[vm.id] self.host.events.clear() result = getattr(self.service, action)(vm.id) self.assertEqual(result.id, vm.id) self.assertNotIn(vm.tap, self.network.taps) self.assertNotIn(vm.id, self.network.firewall.tables) self.assertFalse(Path(vm.socket).exists()) self.assertNotIn("process.start", self.host.actions(vm.id)) if action == "stop": retained = self.store.get(vm.id) self.assertEqual((retained.status, retained.operation), ("stopped", None)) self.assertTrue(Path(vm.disk).exists()) else: self.assertNotIn(vm.id, self.store.load().vms) self.assertFalse(self.settings.vm_dir(vm.id).exists()) def test_failed_delete_intent_cannot_be_resurrected_by_stop_or_start(self) -> None: vm = self.create_vm() self.host.failures["firewall.cleanup_connections", vm.id] = UvmError("conntrack remains") with self.assertRaisesRegex(UvmError, "failed to delete.*conntrack remains"): self.service.destroy(vm.id) retained = self.store.get(vm.id) self.assertEqual((retained.status, retained.operation), ("failed", "delete")) self.assertTrue(Path(vm.disk).exists()) self.assertIn(vm.id, self.network.firewall.tables) for action in ("start", "restart", "sync_ports"): with self.subTest(action=action), self.assertRaisesRegex(ConflictError, "deletion is pending"): getattr(self.service, action)(vm.id) with self.assertRaisesRegex(ConflictError, "deletion is pending"): self.service.set_ports(vm.id, ()) stopped = self.service.stop(vm.id) self.assertEqual((stopped.status, stopped.operation), ("stopped", "delete")) with self.assertRaisesRegex(ConflictError, "deletion is pending"): self.service.start(vm.id) self.service.destroy(vm.id) self.assertNotIn(vm.id, self.store.load().vms) self.assertFalse(self.settings.vm_dir(vm.id).exists()) self.assertEqual(len(self.process.started), 1) def test_failed_start_retains_edited_disk_and_retries_without_provisioning(self) -> None: vm = self.create_vm() self.service.stop(vm.id) Path(vm.disk).write_bytes(b"retained edits and host keys") self.host.failures["client.configure", vm.id] = FirecrackerError("API retry needed") with self.assertRaisesRegex(UvmError, "failed to start.*API retry needed"): self.service.start(vm.id) failed = self.store.get(vm.id) self.assertEqual(failed.status, "failed") self.assertIsNone(failed.pid) self.assertTrue(failed.disk_ready) self.assertTrue(failed.runtime_owned) self.assertNotIn(vm.tap, self.network.taps) started = self.service.start(vm.id) self.assertEqual(started.status, "running") self.assertEqual(Path(vm.disk).read_bytes(), b"retained edits and host keys") self.assertEqual(len({info.pid for info in self.process.started}), 3) self.assertEqual(self.provision.call_count, 1) self.assertEqual(self.copy_disk.call_count, 1) def test_restart_uses_one_lock_and_replaces_only_target_process(self) -> None: vm = self.create_vm() other = self.create_vm() other_record = self.store.get(other.id).to_dict() before_locks = self.store.operation_locks self.host.events.clear() restarted = self.service.restart(vm.id, timeout=0.75) self.assertEqual(self.store.operation_locks - before_locks, 1) self.assertNotEqual(restarted.pid, vm.pid) self.assertEqual(self.process.terminated[0], (vm.id, vm.pid, 0.75)) self.assertEqual(self.store.get(other.id).to_dict(), other_record) self.assertEqual(self.process.observe(other), "alive") self.assertEqual(self.network.firewall.status(other), "enforced") self.assertEqual(self.provision.call_count, 2)