test: boot generated VM and wait for ssh port

test/test_smoke.py

@@ -7,6 +7,7 @@
 
 # local test utils
 import testutil
+from vm import VM
 
 
 @pytest.fixture(name="output_path")
@@ -63,5 +64,8 @@ def test_smoke(output_path, config_json):
     assert journal_output != ""
     generated_img = pathlib.Path(output_path) / "qcow2/disk.qcow2"
     assert generated_img.exists(), f"output file missing, dir content: {os.listdir(os.fspath(output_path))}"
-    # TODO: boot and do basic checks, see
-    # https://github.com/osbuild/osbuild-deploy-container/compare/main...mvo5:integration-test?expand=1
+    with VM(generated_img) as test_vm:
+        test_vm.start()
+        # TODO: login once user creation in osbuild-deploy-container is ready
+        ready = test_vm.wait_ssh_ready()
+        assert ready

test/vm.py

@@ -0,0 +1,72 @@
+import pathlib
+import random
+import subprocess
+import sys
+import socket
+import time
+
+
+class VM:
+    MEM = "2000"
+
+    def __init__(self, img, snapshot=True):
+        self._img = pathlib.Path(img)
+        self._qemu_p = None
+        # there is no race free way to get a free port and pass to qemu via CLI
+        self._ssh_port = 10022 + random.randint(1, 1000)
+        self._snapshot = snapshot
+
+    def __del__(self):
+        self.force_stop()
+
+    def start(self):
+        log_path = self._img.with_suffix(".serial-log")
+        qemu_cmdline = [
+            "qemu-system-x86_64", "-enable-kvm",
+            "-m", self.MEM,
+            # get "illegal instruction" inside the VM otherwise
+            "-cpu", "host",
+            "-serial", f"file:{log_path}",
+            "-netdev", f"user,id=net.0,hostfwd=tcp::{self._ssh_port}-:22",
+            "-device", "rtl8139,netdev=net.0",
+        ]
+        if self._snapshot:
+            qemu_cmdline.append("-snapshot")
+        qemu_cmdline.append(self._img)
+        self._log(f"vm starting, log available at {log_path}")
+
+        # XXX: use systemd-run to ensure cleanup?
+        self._qemu_p = subprocess.Popen(
+            qemu_cmdline, stdout=sys.stdout, stderr=sys.stderr)
+        # XXX: also check that qemu is working and did not crash
+        self.wait_ssh_ready()
+        self._log(f"vm ready at port {self._ssh_port}")
+
+    def _log(self, msg):
+        # XXX: use a proper logger
+        sys.stdout.write(msg.rstrip("\n") + "\n")
+
+    def wait_ssh_ready(self, max_wait=120):
+        sleep = 5
+        for i in range(max_wait // sleep):
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                try:
+                    s.connect(("localhost", self._ssh_port))
+                    data = s.recv(256)
+                    if b"OpenSSH" in data:
+                        return True
+                except ConnectionRefusedError:
+                    time.sleep(sleep)
+        raise ConnectionRefusedError("cannot connect to {self._ssh_port} after {maxwait}")
+
+    def force_stop(self):
+        if self._qemu_p:
+            self._qemu_p.kill()
+            self._qemu_p = None
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, type, value, tb):
+        self.force_stop()