Merge pull request #249 from horw/feat/qemu-qmp

pytest-embedded-qemu/pyproject.toml

@@ -30,6 +30,7 @@ requires-python = ">=3.7"
 
 dependencies = [
     "pytest-embedded~=1.5.0",
+    "qemu.qmp==0.0.3"
 ]
 
 [project.optional-dependencies]

pytest-embedded-qemu/pytest_embedded_qemu/app.py

@@ -162,6 +162,7 @@ def __init__(
         if self.encrypt:
             self.encrypted_image_path = os.path.join(self.binary_path, ENCRYPTED_IMAGE_FN)
 
+        self.qemu_prog_path = kwargs.get('qemu_prog_path', 'qemu-system-xtensa')
         self.create_image()
 
     @property
@@ -176,7 +177,7 @@ def qemu_version(self) -> Version:
         if self._QEMU_VERSION is not None:
             return self._QEMU_VERSION
 
-        s = subprocess.check_output(['qemu-system-xtensa', '--version'], encoding='utf-8')
+        s = subprocess.check_output([self.qemu_prog_path, '--version'], encoding='utf-8')
         version = self.QEMU_VERSION_REGEX.search(s)
         if version is None:
             raise ValueError(f'Could not parse QEMU version from {s}')

pytest-embedded-qemu/pytest_embedded_qemu/dut.py

@@ -23,3 +23,6 @@ def __init__(
 
     def write(self, s: AnyStr) -> None:
         self.qemu.write(s)
+
+    def hard_reset(self):
+        self._hard_reset_func()

pytest-embedded-qemu/pytest_embedded_qemu/qemu.py

@@ -1,8 +1,10 @@
+import asyncio
 import os
 import shlex
 import typing as t
 
 from pytest_embedded.log import DuplicateStdoutPopen
+from qemu.qmp import QMPClient
 
 from . import DEFAULT_IMAGE_FN
 
@@ -26,6 +28,8 @@ class Qemu(DuplicateStdoutPopen):
     QEMU_STRAP_MODE_FMT = '-global driver=esp32.gpio,property=strap_mode,value={}'
     QEMU_SERIAL_TCP_FMT = '-serial tcp::{},server,nowait'
 
+    QEMU_DEFAULT_QMP_FMT = '-qmp tcp:localhost:{},server,wait=off'
+
     def __init__(
         self,
         qemu_image_path: t.Optional[str] = None,
@@ -49,9 +53,33 @@ def __init__(
             raise ValueError(f'QEMU image path doesn\'t exist: {image_path}')
 
         qemu_prog_path = qemu_prog_path or self.qemu_prog_name
+
+        if qemu_cli_args:
+            qemu_cli_args = qemu_cli_args.strip("\"").strip("\'")
         qemu_cli_args = shlex.split(qemu_cli_args or self.qemu_default_args)
         qemu_extra_args = shlex.split(qemu_extra_args or '')
 
+        self.qmp_addr = None
+        self.qmp_port = None
+        self.qmp = QMPClient()
+
+        dut_index = int(kwargs.pop('dut_index', 0))
+        for i, v in enumerate(qemu_cli_args):
+            if v == '-qmp':
+                d = qemu_cli_args[i + 1]
+                if not d.startswith('tcp'):
+                    raise ValueError('Please use TCP for qmp, example: -qmp tcp:localhost:4488,server,wait=off')
+                cmd = d.split(',')
+                _, self.qmp_addr, self.qmp_port = cmd[0].split(':')
+                self.qmp_port = int(self.qmp_port) + dut_index
+                cmd[0] = f'tcp:{self.qmp_addr}:{self.qmp_port}'
+                qemu_cli_args[i + 1] = ','.join(cmd)
+                break
+        else:
+            self.qmp_addr = 'localhost'
+            self.qmp_port = 4488 + dut_index
+            qemu_cli_args += shlex.split(self.QEMU_DEFAULT_QMP_FMT.format(self.qmp_port))
+
         super().__init__(
             cmd=[qemu_prog_path, *qemu_cli_args, *qemu_extra_args] + ['-drive', f'file={image_path},if=mtd,format=raw'],
             **kwargs,
@@ -77,11 +105,22 @@ def qemu_default_args(self):
 
         return self.QEMU_DEFAULT_ARGS
 
+    def qmp_execute_cmd(self, execute, arguments=None):
+        response = None
+
+        async def h_r():
+            nonlocal response
+            try:
+                await self.qmp.connect((str(self.qmp_addr), int(self.qmp_port)))
+                response = await self.qmp.execute(execute, arguments=arguments)
+            finally:
+                await self.qmp.disconnect()
+
+        asyncio.run(h_r())
+        return response
+
     def _hard_reset(self):
-        """
-        This is a fake hard_reset. Keep this API to keep the consistency.
-        """
-        # TODO: implement with QMP
-        #  https://gitlab.com/qemu-project/python-qemu-qmp/-/issues/6
-        #  for now got so many unexpected exceptions while __del__
-        raise NotImplementedError
+        self.qmp_execute_cmd('system_reset')
+
+    def take_screenshot(self, image_path):
+        self.qmp_execute_cmd('screendump', arguments={'filename': image_path})

pytest-embedded-qemu/tests/test_qemu.py

@@ -33,6 +33,32 @@ def test_pexpect_by_qemu(dut):
     result.assert_outcomes(passed=1)
 
 
+@qemu_bin_required
+def test_pexpect_make_restart_by_qemu_xtensa(testdir):
+    testdir.makepyfile("""
+        import pexpect
+        import pytest
+
+        def test_pexpect_by_qemu(dut):
+            dut.expect('Hello world!')
+            dut.hard_reset()
+            dut.expect('cpu_start')
+            dut.expect('Hello world!')
+            dut.hard_reset()
+            dut.expect('cpu_start')
+            dut.expect('Hello world!')
+    """)
+
+    result = testdir.runpytest(
+        '-s',
+        '--embedded-services', 'idf,qemu',
+        '--app-path', os.path.join(testdir.tmpdir, 'hello_world_esp32'),
+        '--qemu-cli-args="-qmp tcp:localhost:4488,server,wait=off -machine esp32 -nographic"',
+    )
+
+    result.assert_outcomes(passed=1)
+
+
 @qemu_bin_required
 def test_pexpect_by_qemu_riscv(testdir):
     testdir.makepyfile("""

pytest-embedded/pytest_embedded/plugin.py

@@ -1088,6 +1088,7 @@ def _fixture_classes_and_options(
                             'skip_regenerate_image': skip_regenerate_image,
                             'encrypt': encrypt,
                             'keyfile': keyfile,
+                            'qemu_prog_path': qemu_prog_path,
                         }
                     )
                 else:
@@ -1204,6 +1205,7 @@ def _fixture_classes_and_options(
                     'qemu_extra_args': qemu_extra_args,
                     'app': None,
                     'meta': _meta,
+                    'dut_index': dut_index,
                 }
         elif fixture == 'wokwi':
             if 'wokwi' in _services: