from __future__ import annotations from dataclasses import replace from pathlib import Path from tests.test_lifecycle import LifecycleTestCase from uvm.config import MIN_STORAGE_MIB from uvm.domain import PortRule, VmResources, VmSpec from uvm.errors import ConflictError, FirecrackerError, UvmError, ValidationError class LifecycleResizeTests(LifecycleTestCase): def test_creation_provisions_the_default_or_requested_storage(self) -> None: for capacity in (MIN_STORAGE_MIB, 80 * 1024): with self.subTest(capacity=capacity): vm = self.create_vm(storage_mib=capacity) self.copy_disk.assert_called_with( self.settings.rootfs_image, Path(vm.disk), storage_mib=capacity, ) self.filesystem_resize.assert_called_with(Path(vm.disk), capacity) self.assertEqual(vm.storage_mib, capacity) self.assertEqual(self.store.get(vm.id).storage_mib, capacity) self.assertEqual(VmSpec(cpu=1, ram_mib=512).storage_mib, MIN_STORAGE_MIB) def test_stop_resize_start_preserves_disk_data_credentials_and_network_identity(self) -> None: vm = self.create_vm(name="database", username="admin", password="same-secret", ports=(PortRule(22), PortRule(5432))) sibling = self.create_vm(name="sibling") sibling_before = self.store.get(sibling.id).to_dict() original = replace(vm) disk = Path(vm.disk) contents = b"guest data\nunchanged password hash\noriginal SSH host keys\n" disk.write_bytes(contents) for cpu, ram_mib, storage_mib in ((2, 4096, 50 * 1024), (0.5, 256, MIN_STORAGE_MIB)): with self.subTest(cpu=cpu, storage=storage_mib): previous_pid = self.store.get(vm.id).pid locks = self.store.operation_locks self.host.events.clear() resized = self.service.resize(vm.name, cpu=cpu, ram_mib=ram_mib, storage_mib=storage_mib, timeout=0.5) self.assertEqual(self.store.operation_locks - locks, 1) self.assertEqual((resized.cpu, resized.ram_mib, resized.storage_mib), (cpu, ram_mib, storage_mib)) self.assertEqual(resized.status, "running") self.assertIsNone(resized.pending_resize) self.assertIsNone(resized.operation) self.assertNotEqual(resized.pid, previous_pid) for field in ("id", "name", "guest_ip", "gateway", "network", "bridge", "mac", "tap", "disk", "username", "password", "ports", "created_at"): self.assertEqual(getattr(resized, field), getattr(original, field), field) self.assertEqual(disk.read_bytes(), contents) actions = self.host.actions(vm.id) self.assertLess(actions.index("process.gone"), actions.index("images.resize_disk")) self.assertLess(actions.index("images.resize_disk"), actions.index("process.start")) stopped = self.host.records["process.terminate", vm.id] self.assertEqual(stopped.pending_resize, VmResources(cpu, ram_mib, storage_mib)) self.assertEqual(stopped.operation, "resize") config = self.clients[vm.id].config self.assertEqual(config["machine-config"], {"vcpu_count": max(1, int(cpu)), "mem_size_mib": ram_mib}) self.assertEqual(config["drives"][0]["path_on_host"], vm.disk) self.assertEqual(self.store.get(vm.id).to_dict(), resized.to_dict()) self.assertEqual(self.store.get(sibling.id).to_dict(), sibling_before) self.assertEqual(self.copy_disk.call_count, 2) self.assertEqual(self.provision.call_count, 2) self.assertEqual(self.resize_disk.call_count, 2) def test_cpu_or_memory_only_resize_keeps_storage_and_boots_an_initially_stopped_vm(self) -> None: vm = self.create_vm(storage_mib=50 * 1024) self.service.stop(vm.id) first = self.service.resize(vm.guest_ip, cpu=2) self.assertEqual((first.cpu, first.ram_mib, first.storage_mib), (2, 512, 50 * 1024)) second = self.service.resize(vm.id, ram_mib=4096) self.assertEqual((second.cpu, second.ram_mib, second.storage_mib), (2, 4096, 50 * 1024)) self.assertEqual(second.status, "running") self.resize_disk.assert_not_called() self.assertEqual(self.provision.call_count, 1) def test_invalid_resize_requests_do_not_change_state_or_stop_the_machine(self) -> None: vm = self.create_vm() before = self.settings.state_path.read_bytes() self.host.events.clear() for changes in ({}, {"cpu": 0}, {"cpu": float("nan")}, {"cpu": True}, {"ram_mib": 127}, {"ram_mib": 512.0}, {"storage_mib": 29 * 1024}, {"storage_mib": True}, {"storage_mib": 2**63}, {"cpu": 2, "timeout": -1}): with self.subTest(changes=changes), self.assertRaises(ValidationError): self.service.resize(vm.id, **changes) self.assertEqual(self.settings.state_path.read_bytes(), before) self.assertEqual(self.host.events, []) self.resize_disk.assert_not_called() def test_resize_rejects_unsafe_or_unbootable_vm_before_stopping(self) -> None: for problem in ("unknown", "delete", "legacy", "unprovisioned", "missing", "topology", "kernel"): with self.subTest(problem=problem): vm = self.create_vm() if problem == "unknown": self.process.unknown.add(vm.id) elif problem == "delete": self.persist(vm, operation="delete") elif problem == "legacy": self.persist(vm, ports=None) elif problem == "unprovisioned": self.persist(vm, disk_ready=False) elif problem == "missing": Path(vm.disk).unlink() elif problem == "topology": self.persist(vm, bridge="different") else: self.settings.kernel_image.unlink() before = self.settings.state_path.read_bytes() self.host.events.clear() with self.assertRaises(UvmError): self.service.resize(vm.id, cpu=2) self.assertEqual(self.settings.state_path.read_bytes(), before) self.assertNotIn("process.terminate", self.host.actions(vm.id)) self.resize_disk.assert_not_called() if problem == "topology": self.persist(vm, bridge="uvm0") if problem == "kernel": self.settings.kernel_image.write_bytes(b"kernel") def test_failed_stop_keeps_old_resources_and_persists_retryable_target(self) -> None: vm = self.create_vm() self.host.failures["process.terminate", vm.id] = FirecrackerError("exit unconfirmed") with self.assertRaisesRegex(UvmError, "failed to resize.*exit unconfirmed"): self.service.resize(vm.id, cpu=2, ram_mib=4096, storage_mib=50 * 1024) retained = self.store.get(vm.id) self.assertEqual((retained.cpu, retained.ram_mib, retained.storage_mib), (1, 512, MIN_STORAGE_MIB)) self.assertEqual(retained.pending_resize, VmResources(2, 4096, 50 * 1024)) self.assertEqual(retained.pid, vm.pid) self.resize_disk.assert_not_called() self.assertEqual(len(self.process.started), 1) with self.assertRaisesRegex(ConflictError, "resize is pending"): self.service.start(vm.id) retried = self.service.resize(vm.id, cpu=2) self.assertEqual((retried.cpu, retried.ram_mib, retried.storage_mib), (2, 4096, 50 * 1024)) def test_failed_disk_resize_is_stopped_and_start_retries_persisted_target(self) -> None: vm = self.create_vm(storage_mib=50 * 1024) self.host.failures["images.resize_disk", vm.id] = UvmError("filesystem does not fit") with self.assertRaisesRegex(UvmError, "filesystem does not fit"): self.service.resize(vm.id, cpu=2, storage_mib=MIN_STORAGE_MIB) retained = self.store.get(vm.id) self.assertEqual((retained.cpu, retained.storage_mib), (1, 50 * 1024)) self.assertEqual(retained.pending_resize, VmResources(2, 512, MIN_STORAGE_MIB)) self.assertIsNone(retained.pid) self.assertEqual(len(self.process.started), 1) self.assertEqual(Path(vm.disk).read_bytes(), b"rootfs") restarted = self.service.start(vm.id) self.assertEqual((restarted.cpu, restarted.storage_mib), (2, MIN_STORAGE_MIB)) self.assertIsNone(restarted.pending_resize) self.assertEqual(self.provision.call_count, 1) def test_failed_boot_retains_applied_resources_for_a_later_start(self) -> None: vm = self.create_vm() self.host.failures["client.configure", vm.id] = FirecrackerError("boot failed") with self.assertRaisesRegex(UvmError, "boot failed"): self.service.resize(vm.id, cpu=2, storage_mib=50 * 1024) retained = self.store.get(vm.id) self.assertEqual((retained.cpu, retained.storage_mib), (2, 50 * 1024)) self.assertIsNone(retained.pending_resize) self.assertTrue(retained.disk_ready) restarted = self.service.start(vm.id) self.assertEqual(restarted.status, "running") self.resize_disk.assert_called_once_with(Path(vm.disk), 50 * 1024) def test_legacy_disk_gets_minimum_capacity_on_its_first_resource_resize(self) -> None: vm = self.create_vm() self.persist(vm, storage_mib=None) resized = self.service.resize(vm.id, cpu=2) self.assertEqual(resized.storage_mib, MIN_STORAGE_MIB) self.resize_disk.assert_called_once_with(Path(vm.disk), MIN_STORAGE_MIB) def test_retry_can_change_target_after_disk_publication_before_metadata_save(self) -> None: for recorded_gib, published_gib in ((30, 50), (50, 30)): with self.subTest(recorded=recorded_gib, published=published_gib): vm = self.create_vm(storage_mib=recorded_gib * 1024) self.service.stop(vm.id) self.persist(vm, operation="resize", pending_resize=VmResources(2, 4096, published_gib * 1024)) self.disk_sizes[Path(vm.disk)] = published_gib * 1024**3 self.resize_disk.reset_mock() result = self.service.resize(vm.id, storage_mib=recorded_gib * 1024) self.resize_disk.assert_called_once_with(Path(vm.disk), recorded_gib * 1024) self.assertEqual(result.storage_mib * 1024 * 1024, self.disk_sizes[Path(vm.disk)]) self.assertIsNone(result.pending_resize) def test_interrupted_resize_can_be_stopped_then_resumed_or_deleted(self) -> None: vm = self.create_vm() self.host.failures["images.resize_disk", vm.id] = KeyboardInterrupt() with self.assertRaises(KeyboardInterrupt): self.service.resize(vm.id, storage_mib=50 * 1024) stopped = self.service.stop(vm.id) self.assertEqual(stopped.pending_resize.storage_mib, 50 * 1024) self.assertEqual(stopped.operation, "resize") self.service.destroy(vm.id) self.assertNotIn(vm.id, self.store.load().vms) self.assertFalse(Path(vm.disk).exists())