Improve Test test_create_qemu_instance_online change sig to async

Change sig of enable_and_start to async

Adapt Firecracker instance test
- Rename to test_create_firecracker_instance
- Use mocker to patch() settings so it doesn't contamine other tests
- Ensure it ping properly to confirm it is working

Author: olethanh

src/aleph/vm/models.py

@@ -336,10 +336,10 @@ async def start(self):
             await self.vm.start_guest_api()
 
             # Start VM and snapshots automatically
-            # If the execution is confidential, don't start it because we need to wait for the session certificate
-            # files, use the endpoint /control/machine/{ref}/confidential/initialize to get session files and start the VM
+            # If the execution is a confidential instance, it is start later in the process when the session certificate
+            # files are received from the client via the endpoint /control/machine/{ref}/confidential/initialize endpoint
             if self.persistent and not self.is_confidential and self.systemd_manager:
-                self.systemd_manager.enable_and_start(self.controller_service)
+                await self.systemd_manager.enable_and_start(self.controller_service)
 
                 if self.is_program:
                     await self.wait_for_init()
@@ -354,6 +354,7 @@ async def start(self):
             self.ready_event.set()
             await self.save()
         except Exception:
+            logger.exception("%s error during start, tearing down", self)
             await self.vm.teardown()
             await self.vm.stop_guest_api()
             raise
@@ -392,11 +393,13 @@ async def non_blocking_wait_for_boot(self):
             await self.wait_for_persistent_boot()
             logger.info("%s responded to ping. Marking it as started.", self)
             self.times.started_at = datetime.now(tz=timezone.utc)
+            return True
             # await self.save()
         except Exception as e:
             logger.warning("%s failed to responded to ping or is not running, stopping it.: %s ", self, e)
             assert self.vm
             await self.stop()
+            return False
 
     async def wait_for_init(self):
         assert self.vm, "The VM attribute has to be set before calling wait_for_init()"

src/aleph/vm/orchestrator/views/operator.py

@@ -260,7 +260,7 @@ async def operate_confidential_initialize(request: web.Request, authenticated_se
         godh_file_path = vm_session_path / "vm_godh.b64"
         godh_file_path.write_bytes(godh_file_content.file.read())
 
-        pool.systemd_manager.enable_and_start(execution.controller_service)
+        await pool.systemd_manager.enable_and_start(execution.controller_service)
 
         return web.Response(status=200, body=f"Started VM with ref {vm_hash}")
 

src/aleph/vm/systemd.py

@@ -72,7 +72,7 @@ def is_service_active(self, service: str) -> bool:
             logger.error(error)
             return False
 
-    def enable_and_start(self, service: str) -> None:
+    async def enable_and_start(self, service: str) -> None:
         if not self.is_service_enabled(service):
             self.enable(service)
         if not self.is_service_active(service):

tests/supervisor/test_instance.py

@@ -23,7 +23,9 @@ class MockSystemDManager(SystemDManager):
     execution: MicroVM | None = None
     process: Process | None = None
 
-    async def enable_and_start(self, vm_hash: str):
+    async def enable_and_start(self, service: str) -> tuple[MicroVM | None, Process | None]:
+        vm_hash = service.split("@", maxsplit=1)[1].split(".", maxsplit=1)[0]
+
         config_path = Path(f"{settings.EXECUTION_ROOT}/{vm_hash}-controller.json")
         config = configuration_from_file(config_path)
         self.execution, self.process = await execute_persistent_vm(config)
@@ -35,7 +37,7 @@ def is_service_enabled(self, service: str):
     def is_service_active(self, service: str):
         return self.process is not None
 
-    async def stop_and_disable(self, vm_hash: str):
+    async def stop_and_disable(self, service: str):
         if self.execution:
             await self.execution.shutdown()
             await self.execution.stop()
@@ -45,25 +47,31 @@ async def stop_and_disable(self, vm_hash: str):
 
 
 @pytest.mark.asyncio
-async def test_create_instance():
-    """
-    Create a fake instance locally and check that it start / init / stop properly.
+async def test_create_firecracker_instance(mocker):
+    """Create a fake instance locally and check that it start / init / stop properly.
+
+    NOTE: If Firecracker VM fail to boot because the disk is broken try:
+     ```bash
+     sudo dmsetup remove decadecadecadecadecadecadecadecadecadecadecadecadecadecadecadeca_rootfs
+     sudo dmsetup remove decadecadecadecadecadecadecadecadecadecadecadecadecadecadecadeca_base
+     sudo losetup -l | grep 'persistent' | grep deleted | awk  '{print $1}' | sudo xargs -I{} losetup -d {}
+     sudo rm -rf /var/lib/aleph/vm/volumes/persistent/decadecadecadecadecadecadecadecadecadecadecadecadecadecadecadeca/rootfs.btrfs
+     ```
     """
+    mocker.patch.object(settings, "ALLOW_VM_NETWORKING", True)
+    mocker.patch.object(settings, "USE_FAKE_INSTANCE_BASE", True)
+    mocker.patch.object(settings, "FAKE_DATA_PROGRAM", settings.BENCHMARK_FAKE_DATA_PROGRAM)
+    mocker.patch.object(settings, "USE_JAILER", True)
 
-    settings.USE_FAKE_INSTANCE_BASE = True
-    settings.FAKE_DATA_PROGRAM = settings.BENCHMARK_FAKE_DATA_PROGRAM
-    # settings.FAKE_INSTANCE_MESSAGE
-    settings.ALLOW_VM_NETWORKING = True
-    settings.USE_JAILER = True
-
-    logging.basicConfig(level=logging.DEBUG)
-    settings.PRINT_SYSTEM_LOGS = True
+    # logging.basicConfig(level=logging.DEBUG)
 
     # Ensure that the settings are correct and required files present.
     settings.setup()
     settings.check()
     if not settings.FAKE_INSTANCE_BASE.exists():
-        pytest.xfail("Test Runtime not setup. run `cd runtimes/instance-rootfs && sudo ./create-debian-12-disk.sh`")
+        pytest.xfail(
+            f"Test Runtime not setup. {settings.FAKE_INSTANCE_BASE}. run `cd runtimes/instance-rootfs && sudo ./create-debian-12-disk.sh`"
+        )
 
     # The database is required for the metrics and is currently not optional.
     engine = metrics.setup_engine()
@@ -93,7 +101,7 @@ async def test_create_instance():
         message=message.content,
         original=message.content,
         snapshot_manager=None,
-        systemd_manager=None,
+        systemd_manager=mock_systemd_manager,
         persistent=True,
     )
 
@@ -114,13 +122,17 @@ async def test_create_instance():
     assert vm.enable_networking
 
     await execution.start()
-    firecracker_execution, process = await mock_systemd_manager.enable_and_start(execution.vm_hash)
+    # firecracker_execution, process = await mock_systemd_manager.enable_and_start(execution.vm_hash)
+    firecracker_execution = mock_systemd_manager.execution
     assert isinstance(firecracker_execution, MicroVM)
     assert firecracker_execution.proc is not None
-    await execution.wait_for_init()
 
-    # This sleep is to leave the instance to boot up and prevent disk corruption
+    await execution.init_task
+    assert execution.init_task.result() is True, "VM failed to start"
+
+    # This sleep is to leave the instance to boo
+    # up and prevent disk corruption
     await asyncio.sleep(60)
-    firecracker_execution, process = await mock_systemd_manager.stop_and_disable(execution.vm_hash)
+    firecracker_execution, process = await mock_systemd_manager.stop_and_disable(execution.controller_service)
     await execution.stop()
     assert firecracker_execution is None

tests/supervisor/test_qemu_instance.py

@@ -23,7 +23,13 @@ class MockSystemDManager(SystemDManager):
     execution: QemuVM | None = None
     process: Process | None = None
 
-    async def enable_and_start(self, vm_hash: str):
+    async def enable_and_start(self, service: str) -> tuple[QemuVM | None, Process | None]:
+        # aleph-vm-controller@decadecadecadecadecadecadecadecadecadecadecadecadecadecadecadeca.service-controller.json
+        if '@' in service:
+            vm_hash = service.split('@', maxsplit=1)[1].split('.', maxsplit=1)[0]
+        else:
+            vm_hash = service
+
         config_path = Path(f"{settings.EXECUTION_ROOT}/{vm_hash}-controller.json")
         config = configuration_from_file(config_path)
         self.execution, self.process = await execute_persistent_vm(config)
@@ -33,7 +39,7 @@ def is_service_enabled(self, service: str):
         return self.process is not None
 
     def is_service_active(self, service: str):
-        return self.process is not None
+        return self.process is not None and not self.process.returncode
 
     async def stop_and_disable(self, vm_hash: str):
         if self.process:
@@ -92,34 +98,39 @@ async def test_create_qemu_instance():
     assert vm.vm_id == vm_id
 
     await execution.start()
-    qemu_execution, process = await mock_systemd_manager.enable_and_start(execution.vm_hash)
+    qemu_execution, process = await mock_systemd_manager.enable_and_start(execution.controller_service)
     assert isinstance(qemu_execution, QemuVM)
     assert qemu_execution.qemu_process is not None
-    qemu_execution, process = await mock_systemd_manager.stop_and_disable(execution.vm_hash)
+    await mock_systemd_manager.stop_and_disable(execution.vm_hash)
+    await qemu_execution.qemu_process.wait()
+    assert qemu_execution.qemu_process.returncode is not None
     await execution.stop()
-    assert qemu_execution is None
 
 
 @pytest.mark.asyncio
-async def test_create_qemu_instance_online():
+async def test_create_qemu_instance_online(mocker):
     """
     Create an instance and check that it start / init / stop properly.
+    With network, wait for ping
     """
-
-    settings.USE_FAKE_INSTANCE_BASE = True
-    settings.FAKE_INSTANCE_MESSAGE = settings.FAKE_INSTANCE_QEMU_MESSAGE
-    settings.FAKE_INSTANCE_BASE = settings.FAKE_QEMU_INSTANCE_BASE
-    settings.ENABLE_CONFIDENTIAL_COMPUTING = False
-    settings.ALLOW_VM_NETWORKING = True
-    settings.USE_JAILER = False
+    mocker.patch.object(settings, "ALLOW_VM_NETWORKING", True)
+    mocker.patch.object(settings, "USE_FAKE_INSTANCE_BASE", True)
+    mocker.patch.object(settings, "FAKE_INSTANCE_MESSAGE", settings.FAKE_INSTANCE_QEMU_MESSAGE)
+    mocker.patch.object(settings, "FAKE_INSTANCE_BASE", settings.FAKE_INSTANCE_QEMU_MESSAGE)
+    mocker.patch.object(settings, "ENABLE_CONFIDENTIAL_COMPUTING", False)
+    mocker.patch.object(settings, "USE_JAILER", False)
 
     logging.basicConfig(level=logging.DEBUG)
 
     # Ensure that the settings are correct and required files present.
     settings.setup()
     settings.check()
     if not settings.FAKE_INSTANCE_BASE.exists():
-        pytest.xfail("Test Runtime not setup. run `cd runtimes/instance-rootfs && sudo ./create-debian-12-disk.sh`")
+        pytest.xfail(
+            "Test instance disk {} not setup. run `cd runtimes/instance-rootfs && sudo ./create-debian-12-disk.sh` ".format(
+                settings.FAKE_QEMU_INSTANCE_BASE
+            )
+        )
 
     # The database is required for the metrics and is currently not optional.
     engine = metrics.setup_engine()
@@ -130,29 +141,26 @@ async def test_create_qemu_instance_online():
 
     mock_systemd_manager = MockSystemDManager()
 
-    network = (
-        Network(
-            vm_ipv4_address_pool_range=settings.IPV4_ADDRESS_POOL,
-            vm_network_size=settings.IPV4_NETWORK_PREFIX_LENGTH,
-            external_interface=settings.NETWORK_INTERFACE,
-            ipv6_allocator=make_ipv6_allocator(
-                allocation_policy=settings.IPV6_ALLOCATION_POLICY,
-                address_pool=settings.IPV6_ADDRESS_POOL,
-                subnet_prefix=settings.IPV6_SUBNET_PREFIX,
-            ),
-            use_ndp_proxy=False,
-            ipv6_forwarding_enabled=False,
-        )
-        if settings.ALLOW_VM_NETWORKING
-        else None
+    network = Network(
+        vm_ipv4_address_pool_range=settings.IPV4_ADDRESS_POOL,
+        vm_network_size=settings.IPV4_NETWORK_PREFIX_LENGTH,
+        external_interface=settings.NETWORK_INTERFACE,
+        ipv6_allocator=make_ipv6_allocator(
+            allocation_policy=settings.IPV6_ALLOCATION_POLICY,
+            address_pool=settings.IPV6_ADDRESS_POOL,
+            subnet_prefix=settings.IPV6_SUBNET_PREFIX,
+        ),
+        use_ndp_proxy=False,
+        ipv6_forwarding_enabled=False,
     )
+    network.setup()
 
     execution = VmExecution(
         vm_hash=vm_hash,
         message=message.content,
         original=message.content,
         snapshot_manager=None,
-        systemd_manager=None,
+        systemd_manager=mock_systemd_manager,
         persistent=True,
     )
 
@@ -170,10 +178,11 @@ async def test_create_qemu_instance_online():
     assert vm.vm_id == vm_id
 
     await execution.start()
-    qemu_execution, process = await mock_systemd_manager.enable_and_start(execution.vm_hash)
+    qemu_execution = mock_systemd_manager.execution
     assert isinstance(qemu_execution, QemuVM)
     assert qemu_execution.qemu_process is not None
-    await execution.wait_for_init()
+    await execution.init_task
+    assert execution.init_task.result() is True, "VM failed to start"
     qemu_execution, process = await mock_systemd_manager.stop_and_disable(execution.vm_hash)
     await execution.stop()
     assert qemu_execution is None

tests/supervisor/views/test_operator.py

@@ -219,7 +219,7 @@ async def test_operator_confidential_initialize_not_confidential(aiohttp_client,
 
 
 @pytest.mark.asyncio
-async def test_operator_confidential_initialize(aiohttp_client):
+async def test_operator_confidential_initialize(aiohttp_client, mocker):
     """Test that the certificates system endpoint responds. No auth needed"""
 
     settings.ENABLE_QEMU_SUPPORT = True
@@ -236,7 +236,7 @@ class FakeExecution:
         controller_service: str = ""
 
     class MockSystemDManager:
-        enable_and_start = MagicMock(return_value=True)
+        enable_and_start = mocker.AsyncMock(return_value=True)
 
     class FakeVmPool:
         executions: dict[ItemHash, FakeExecution] = {}