qemudriver: Allow machine-specific options and pre-start QMP commands

doc/configuration.rst

@@ -2867,6 +2867,10 @@ The QEMUDriver also requires the specification of:
   specify the build device tree
 - a path key, this is the path to the rootfs
 
+To allow interactions with the prepared, not yet started QEMU instance, the ``on()``
+method can take an optional ``pre_start_hook`` callable which is executed right before
+the CPU(s) are released.
+
 SigrokDriver
 ~~~~~~~~~~~~
 The :any:`SigrokDriver` uses a `SigrokDevice`_ resource to record samples and provides

dockerfiles/README.rst

@@ -22,7 +22,7 @@ Example showing how to build labgrid-client image:
 
 .. code-block:: bash
 
-   $ docker build --target labgrid-client -t docker.io/labgrid/client -f dockerfiles/Dockerfile .
+   $ docker build --build-arg VERSION="$(python -m setuptools_scm)" --target labgrid-client -t docker.io/labgrid/client -f dockerfiles/Dockerfile .
 
 Using `BuildKit <https://docs.docker.com/develop/develop-images/build_enhancements/>`_
 is recommended to reduce build times.

labgrid/driver/qemudriver.py

@@ -8,6 +8,7 @@
 import re
 import tempfile
 import time
+from typing import List, Optional, Dict, Tuple, Any, Callable, Union
 
 import attr
 from pexpect import TIMEOUT
@@ -95,18 +96,18 @@ class QEMUDriver(ConsoleExpectMixin, Driver, PowerProtocol, ConsoleProtocol):
         default=None,
         validator=attr.validators.optional(attr.validators.instance_of(str)))
 
-    def __attrs_post_init__(self):
+    def __attrs_post_init__(self) -> None:
         super().__attrs_post_init__()
-        self.status = 0
-        self.txdelay = None
-        self._child = None
-        self._tempdir = None
-        self._socket = None
-        self._clientsocket = None
-        self._forwarded_ports = {}
+        self.status: int = 0
+        self.txdelay: Optional[float] = None
+        self._child: Optional[subprocess.Popen] = None
+        self._tempdir: Optional[str] = None
+        self._socket: Optional[socket.socket] = None
+        self._clientsocket: Optional[socket.socket] = None
+        self._forwarded_ports: Dict[Tuple[str, str, int], Tuple[str, str, int, str, int]] = {}
         atexit.register(self._atexit)
 
-    def _atexit(self):
+    def _atexit(self) -> None:
         if not self._child:
             return
         self._child.terminate()
@@ -116,7 +117,7 @@ def _atexit(self):
             self._child.kill()
             self._child.communicate(timeout=1)
 
-    def get_qemu_version(self, qemu_bin):
+    def get_qemu_version(self, qemu_bin: str) -> Tuple[int, int, int]:
         p = subprocess.run([qemu_bin, "-version"], stdout=subprocess.PIPE, encoding="utf-8")
         if p.returncode != 0:
             raise ExecutionError(f"Unable to get QEMU version. QEMU exited with: {p.returncode}")
@@ -127,7 +128,7 @@ def get_qemu_version(self, qemu_bin):
 
         return (int(m.group('major')), int(m.group('minor')), int(m.group('micro')))
 
-    def get_qemu_base_args(self):
+    def get_qemu_base_args(self) -> List[str]:
         """Returns the base command line used for Qemu without the options
         related to QMP. These options can be used to start an interactive
         Qemu manually for debugging tests
@@ -156,19 +157,20 @@ def get_qemu_base_args(self):
             disk_opts = ""
             if self.disk_opts:
                 disk_opts = f",{self.disk_opts}"
-            if self.machine == "vexpress-a9":
+            machine_base = self.machine.split(',')[0]
+            if machine_base == "vexpress-a9":
                 cmd.append("-drive")
                 cmd.append(
                     f"if=sd,format={disk_format},file={disk_path},id=mmc0{disk_opts}")
                 boot_args.append("root=/dev/mmcblk0p1 rootfstype=ext4 rootwait")
-            elif self.machine in ["pc", "q35", "virt"]:
+            elif machine_base in ["pc", "q35", "virt"]:
                 cmd.append("-drive")
                 cmd.append(
                     f"if=virtio,format={disk_format},file={disk_path}{disk_opts}")
                 boot_args.append("root=/dev/vda rootwait")
             else:
                 raise NotImplementedError(
-                    f"QEMU disk image support not implemented for machine '{self.machine}'"
+                    f"QEMU disk image support not implemented for machine '{machine_base}'"
                 )
         if self.rootfs is not None:
             cmd.append("-fsdev")
@@ -229,7 +231,7 @@ def get_qemu_base_args(self):
 
         return cmd
 
-    def on_activate(self):
+    def on_activate(self) -> None:
         self._tempdir = tempfile.mkdtemp(prefix="labgrid-qemu-tmp-")
         sockpath = f"{self._tempdir}/serialrw"
         self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -247,7 +249,7 @@ def on_activate(self):
         self._cmd.append("-serial")
         self._cmd.append("chardev:serialsocket")
 
-    def on_deactivate(self):
+    def on_deactivate(self) -> None:
         if self.status:
             self.off()
         if self._clientsocket:
@@ -258,9 +260,9 @@ def on_deactivate(self):
         shutil.rmtree(self._tempdir)
 
     @step()
-    def on(self):
-        """Start the QEMU subprocess, accept the unix socket connection and
-        afterwards start the emulator using a QMP Command"""
+    def on(self, pre_start_hook: Optional[Callable[[], None]] = None) -> None:
+        """Start the QEMU subprocess, accept the unix socket connection, run the
+        pre_start_hook if applicable and start the emulator using a QMP Command"""
         if self.status:
             return
         self.logger.debug("Starting with: %s", self._cmd)
@@ -288,10 +290,13 @@ def on(self):
         for v in self._forwarded_ports.values():
             self._add_port_forward(*v)
 
+        if pre_start_hook:
+            pre_start_hook()
+
         self.monitor_command("cont")
 
     @step()
-    def off(self):
+    def off(self) -> None:
         """Stop the emulator using a monitor command and await the exitcode"""
         if not self.status:
             return
@@ -302,37 +307,38 @@ def off(self):
         self._child = None
         self.status = 0
 
-    def cycle(self):
+    def cycle(self) -> None:
         """Cycle the emulator by restarting it"""
         self.off()
         self.on()
 
     @step(result=True, args=['command', 'arguments'])
-    def monitor_command(self, command, arguments={}):
+    def monitor_command(self, command: str, arguments: Dict[str, Any] = {}) -> Any:
         """Execute a monitor_command via the QMP"""
         if not self.status:
             raise ExecutionError(
                 "Can't use monitor command on non-running target")
         return self.qmp.execute(command, arguments)
 
-    def _add_port_forward(self, proto, local_address, local_port, remote_address, remote_port):
+    def _add_port_forward(self, proto: str, local_address: str, local_port: int, remote_address: str, remote_port: int) -> None:
         self.monitor_command(
             "human-monitor-command",
             {"command-line": f"hostfwd_add {proto}:{local_address}:{local_port}-{remote_address}:{remote_port}"},
         )
 
-    def add_port_forward(self, proto, local_address, local_port, remote_address, remote_port):
+    def add_port_forward(self, proto: str, local_address: str, local_port: int, remote_address: str, remote_port: int) -> None:
         self._add_port_forward(proto, local_address, local_port, remote_address, remote_port)
-        self._forwarded_ports[(proto, local_address, local_port)] = (proto, local_address, local_port, remote_address, remote_port)
+        self._forwarded_ports[(proto, local_address, local_port)] = (
+            proto, local_address, local_port, remote_address, remote_port)
 
-    def remove_port_forward(self, proto, local_address, local_port):
+    def remove_port_forward(self, proto: str, local_address: str, local_port: int) -> None:
         del self._forwarded_ports[(proto, local_address, local_port)]
         self.monitor_command(
             "human-monitor-command",
             {"command-line": f"hostfwd_remove {proto}:{local_address}:{local_port}"},
         )
 
-    def _read(self, size=1, timeout=10, max_size=None):
+    def _read(self, size: int = 1, timeout: float = 10, max_size: Optional[int] = None) -> bytes:
         ready, _, _ = select.select([self._clientsocket], [], [], timeout)
         if ready:
             # Collect some more data
@@ -345,8 +351,8 @@ def _read(self, size=1, timeout=10, max_size=None):
             raise TIMEOUT(f"Timeout of {timeout:.2f} seconds exceeded")
         return res
 
-    def _write(self, data):
+    def _write(self, data: bytes) -> int:
         return self._clientsocket.send(data)
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"QemuDriver({self.target.name})"

tests/test_qemudriver.py

@@ -13,7 +13,8 @@ def qemu_env(tmpdir):
             role: foo
         images:
           kernel: "test.zImage"
-          dtb: test.dtb"
+          disk: "test.qcow2"
+          dtb: "test.dtb"
         tools:
           qemu: "qemu-system-arm"
         paths:
@@ -69,6 +70,15 @@ def qemu_version_mock(mocker):
 def test_qemu_instance(qemu_target, qemu_driver):
     assert (isinstance(qemu_driver, QEMUDriver))
 
+def test_qemu_get_qemu_base_args_disk(qemu_target, qemu_driver):
+    qemu_driver.disk = 'disk'
+    supported_machines = ['vexpress-a9', 'pc', 'q35', 'virt']
+    for machine in supported_machines:
+        qemu_driver.machine = machine
+        qemu_driver.get_qemu_base_args()
+        qemu_driver.machine = machine + ',option=value'
+        qemu_driver.get_qemu_base_args()
+
 def test_qemu_activate_deactivate(qemu_target, qemu_driver, qemu_version_mock):
     qemu_target.activate(qemu_driver)
     qemu_target.deactivate(qemu_driver)
@@ -81,6 +91,18 @@ def test_qemu_on_off(qemu_target, qemu_driver, qemu_mock, qemu_version_mock):
 
     qemu_target.deactivate(qemu_driver)
 
+def test_qemu_on_pre_start_hook(qemu_target, qemu_driver, qemu_mock, qemu_version_mock):
+    qemu_target.activate(qemu_driver)
+
+    called = False
+    def _hook():
+        nonlocal called
+        qemu_driver.monitor_command("info")
+        called = True
+
+    qemu_driver.on(pre_start_hook=_hook)
+    assert called
+
 def test_qemu_read_write(qemu_target, qemu_driver, qemu_mock, qemu_version_mock):
     qemu_target.activate(qemu_driver)