from __future__ import annotations from pathlib import Path from unittest.mock import Mock from tests.test_lifecycle import LifecycleTestCase from uvm.errors import ConflictError, GuestAgentUnavailableError, GuestExecutionError, NotFoundError, UvmError from uvm.guest_exec import ExecResult class GuestExecLifecycleTests(LifecycleTestCase): def test_create_provisions_agent_before_the_disk_is_ready_or_the_guest_is_started(self) -> None: def provision(disk, vm_id, username): self.host.call("images.provision_agent", vm_id) vm = self.store.get(vm_id) self.assertFalse(vm.disk_ready) self.assertEqual(self.process.observe(vm), "gone") self.assertEqual(disk, Path(vm.disk)) self.assertEqual(username, "worker") self.provision_agent.side_effect = provision vm = self.create_vm(username="worker") self.assert_call_order(vm, "images.provision_agent", "process.start") def test_stopped_boots_automatically_provision_agent_before_starting_the_vmm(self) -> None: vm = self.create_vm(username="worker", password="retained-password") identity = (vm.id, vm.guest_ip, vm.mac, vm.username, vm.password) def provision(disk, vm_id, username): self.host.call("images.provision_agent", vm_id) record = self.store.get(vm_id) self.assertTrue(self.store.operation_active) self.assertEqual(self.process.observe(record), "gone") self.assertEqual((record.id, record.guest_ip, record.mac, record.username, record.password), identity) self.assertEqual((disk, username), (Path(vm.disk), "worker")) self.provision_agent.side_effect = provision for action in ("start", "restart", "resize"): with self.subTest(action=action): if action == "start": self.service.stop(vm.id) self.host.events.clear() self.provision_agent.reset_mock() self.provision.reset_mock() if action == "resize": started = self.service.resize(vm.id, cpu=2) else: started = getattr(self.service, action)(vm.id) self.assertEqual(started.status, "running") self.provision_agent.assert_called_once_with(Path(vm.disk), vm.id, "worker") self.provision.assert_not_called() self.assert_call_order(vm, "images.provision_agent", "process.start") def test_live_or_ambiguous_vms_never_run_offline_agent_provisioning(self) -> None: vm = self.create_vm() self.provision_agent.reset_mock() self.service.start(vm.id) self.process.unknown.add(vm.id) with self.assertRaises(ConflictError): self.service.start(vm.id) self.provision_agent.assert_not_called() def test_default_agent_setup_failure_blocks_boot_and_keeps_the_retained_disk(self) -> None: vm = self.create_vm() self.service.stop(vm.id) before = Path(vm.disk).read_bytes() count = len(self.process.started) self.provision_agent.side_effect = UvmError("agent setup failed") with self.assertRaisesRegex(UvmError, "agent setup failed"): self.service.start(vm.id) self.assertEqual(len(self.process.started), count) self.assertEqual(Path(vm.disk).read_bytes(), before) self.assertEqual(self.store.get(vm.id).status, "failed") self.provision_agent.side_effect = None self.assertEqual(self.service.start(vm.id).status, "running") def test_default_agent_setup_failure_rolls_back_a_new_vm_before_boot(self) -> None: self.provision_agent.side_effect = UvmError("unsupported guest image") with self.assertRaisesRegex(UvmError, "unsupported guest image"): self.create_vm() self.assertEqual(self.store.load().vms, {}) self.assertEqual(self.process.started, []) def test_execution_pins_the_exact_vm_under_lock_and_waits_without_holding_it(self) -> None: vm = self.create_vm(name="worker", ports=()) result = ExecResult(7, "output", "problem", False, False) session = Mock() def connect(path, vm_id, username, pid): self.assertTrue(self.store.operation_active) self.assertEqual((path, vm_id, username, pid), (self.settings.guest_socket_path(vm.id), vm.id, vm.username, vm.pid)) return session def execute(command, **options): self.assertFalse(self.store.operation_active) self.assertFalse(self.store.transaction_active) self.service.stop(vm.id) return result self.service._guest_connector = Mock(side_effect=connect) session.execute.side_effect = execute self.assertEqual(self.service.execute(vm.id, "exit 7", sudo=True, timeout_seconds=2), result) session.execute.assert_called_once_with("exit 7", sudo=True, timeout_seconds=2) session.close.assert_called_once() def test_execution_rejects_aliases_stopped_unknown_and_pending_vms_before_connecting(self) -> None: vm = self.create_vm(name="worker") self.service._guest_connector = Mock() for alias in (vm.name, vm.guest_ip): with self.assertRaises(NotFoundError): self.service.execute(alias, "id") for changes in ({"status": "stopped"}, {"operation": "policy"}, {"operation": "delete"}): self.persist(vm, **changes) with self.assertRaises(ConflictError): self.service.execute(vm.id, "id") self.persist(vm, status="running", operation=None) self.process.unknown.add(vm.id) with self.assertRaises(ConflictError): self.service.execute(vm.id, "id") self.service._guest_connector.assert_not_called() def test_unavailable_or_lost_agent_does_not_retry_or_mutate_vm_state(self) -> None: vm = self.create_vm() before = self.settings.state_path.read_bytes() self.service._guest_connector = Mock(side_effect=GuestAgentUnavailableError("not ready")) with self.assertRaises(GuestAgentUnavailableError): self.service.execute(vm.id, "id") self.service._guest_connector.assert_called_once() session = Mock() session.execute.side_effect = GuestExecutionError("connection lost", outcome_unknown=True) self.service._guest_connector = Mock(return_value=session) with self.assertRaises(GuestExecutionError): self.service.execute(vm.id, "id") session.execute.assert_called_once() session.close.assert_called_once() self.assertEqual(self.settings.state_path.read_bytes(), before) def test_agent_upgrade_requires_a_stopped_proven_offline_disk_and_retains_identity(self) -> None: vm = self.create_vm(username="worker") self.provision_agent.reset_mock() with self.assertRaises(ConflictError): self.service.install_guest_agent(vm.id) self.provision_agent.assert_not_called() stopped = self.service.stop(vm.id) before = stopped.to_dict() def provision(disk, vm_id, username, *, install): self.assertTrue(self.store.operation_active) self.assertEqual(self.process.observe(self.store.get(vm_id)), "gone") self.assertEqual((disk, username, install), (Path(vm.disk), "worker", True)) self.provision_agent.side_effect = provision self.assertEqual(self.service.install_guest_agent(vm.id).to_dict(), before) self.provision_agent.assert_called_once() self.assertEqual(self.store.get(vm.id).to_dict(), before) self.assertEqual(len(self.process.started), 1)