Add QEMU support for NuttX

Author: fdcavalcanti

pytest-embedded-nuttx/pytest_embedded_nuttx/__init__.py

@@ -4,20 +4,23 @@
 
 from pytest_embedded.utils import lazy_load
 
-from .dut import NuttxDut, NuttxEspDut
+from .app import NuttxApp
+from .dut import NuttxDut, NuttxEspDut, NuttxQemuDut, NuttxSerialDut
 
 __getattr__ = lazy_load(
     importlib.import_module(__name__),
     {
         'NuttxDut': NuttxDut,
+        'NuttxSerialDut': NuttxSerialDut,
         'NuttxEspDut': NuttxEspDut,  # requires 'esp' service
+        'NuttxQemuDut': NuttxQemuDut,  # requires 'qemu' service
     },
     {
         'NuttxApp': '.app',  # requires 'esp' service
         'NuttxSerial': '.serial',  # requires 'esp' service
     },
 )
 
-__all__ = ['NuttxApp', 'NuttxSerial', 'NuttxEspDut', 'NuttxDut']
+__all__ = ['NuttxApp', 'NuttxSerial', 'NuttxSerialDut', 'NuttxQemuDut', 'NuttxEspDut', 'NuttxDut']
 
 __version__ = '1.12.0'

pytest-embedded-nuttx/pytest_embedded_nuttx/dut.py

@@ -4,26 +4,23 @@
 from typing import TYPE_CHECKING, AnyStr
 
 import pexpect
+from pytest_embedded.dut import Dut
+from pytest_embedded_qemu.dut import QemuDut
+from pytest_embedded_qemu.qemu import Qemu
 from pytest_embedded_serial.dut import SerialDut
 
 if TYPE_CHECKING:
     from .app import NuttxApp
 
 
-class NuttxDut(SerialDut):
+class NuttxDut(Dut):
     """
     Generic DUT class for use with NuttX RTOS.
     """
 
     PROMPT_NSH = 'nsh>'
     PROMPT_TIMEOUT_S = 30
 
-    def __init__(
-        self,
-        **kwargs,
-    ) -> None:
-        super().__init__(**kwargs)
-
     def write(self, data: str) -> None:
         """
         Write to NuttShell and sleep for a few hundred milliseconds to
@@ -45,11 +42,14 @@ def return_code(self, timeout: int = PROMPT_TIMEOUT_S) -> int:
         Matches the 'echo $?' response and extracts the integer value
         corresponding to the last program return code.
 
+        The first regex option on expect is for serial interface,
+        while the second will match QEMU.
+
         Returns:
             int: return code.
         """
         self.write('echo $?')
-        echo_match = self.expect(r'echo \$\?\r\n(\d+)', timeout=timeout)
+        echo_match = self.expect([r'echo \$\?\r\n(\d+)', r'echo \$\?\n*(\d+)\n'], timeout=timeout)
         ret_code = re.findall(r'\d+', echo_match.group().decode())
 
         if not ret_code:
@@ -75,8 +75,78 @@ def write_and_return(self, data: str, timeout: int = 2) -> AnyStr:
         ans = self.expect(pexpect.TIMEOUT, timeout=timeout)
         return ans.rstrip().decode()
 
+    def reset_to_nsh(self, ready_prompt: str = PROMPT_NSH) -> None:
+        """
+        Resets the board and waits until the Nuttshell prompt appears.
+        Defaults to 'nsh>'.
+
+        Args:
+            ready_prompt (str): string on prompt that signals completion.
+
+        Returns:
+            None
+        """
+        if self.reset:
+            logging.info('Resetting board')
+            self.reset()
+        else:
+            logging.error('Resetting method not available')
+        self.expect(ready_prompt, timeout=self.PROMPT_TIMEOUT_S)
+
+
+class NuttxSerialDut(SerialDut, NuttxDut):
+    """
+    DUT class for serial ports connected to generic boards running NuttX
+    with NuttX RTOS.
+    """
+
+    def __init__(
+        self,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+
+    def reset(self) -> None:
+        """Reset the DUT by toggling the DTR line.
+
+        Args:
+            None.
+
+        Returns:
+            None.
+        """
+        self.serial.proc.dtr = False
+        sleep(0.2)
+        self.serial.proc.dtr = True
+
+
+class NuttxQemuDut(QemuDut, NuttxDut):
+    """
+    DUT class for QEMU usage of the NuttX RTOS.
+    """
+
+    def __init__(
+        self,
+        qemu: Qemu,
+        **kwargs,
+    ) -> None:
+        self.qemu = qemu
+
+        super().__init__(qemu=qemu, **kwargs)
 
-class NuttxEspDut(NuttxDut):
+    def reset(self) -> None:
+        """Hard reset the DUT.
+
+        Args:
+            None.
+
+        Returns:
+            None.
+        """
+        self.hard_reset()
+
+
+class NuttxEspDut(NuttxSerialDut):
     """
     DUT class for serial ports connected to Espressif boards which are
     flashed with NuttX RTOS.
@@ -91,16 +161,14 @@ def __init__(
 
         super().__init__(app=app, **kwargs)
 
-    def reset_to_nsh(self, ready_prompt: str = NuttxDut.PROMPT_NSH) -> None:
+    def reset(self) -> None:
         """
-        Resets the board and waits until the Nuttshell prompt appears.
-        Defaults to 'nsh>'.
+        Resets the board.
 
         Args:
-            ready_prompt (str): string on prompt that signals completion.
+            None.
 
         Returns:
-            None
+            None.
         """
         self.serial.hard_reset()
-        self.expect(ready_prompt, timeout=NuttxDut.PROMPT_TIMEOUT_S)

pytest-embedded/pytest_embedded/dut_factory.py

@@ -362,12 +362,20 @@ def _fixture_classes_and_options_fn(
                 else:
                     raise SystemExit('wokwi service should be used together with idf or arduino service')
             elif 'qemu' in _services:
-                from pytest_embedded_qemu import QemuDut
+                if 'nuttx' in _services:
+                    from pytest_embedded_nuttx import NuttxQemuDut
 
-                classes[fixture] = QemuDut
-                kwargs[fixture].update({
-                    'qemu': None,
-                })
+                    classes[fixture] = NuttxQemuDut
+                    kwargs[fixture].update({
+                        'qemu': None,
+                    })
+                else:
+                    from pytest_embedded_qemu import QemuDut
+
+                    classes[fixture] = QemuDut
+                    kwargs[fixture].update({
+                        'qemu': None,
+                    })
             elif 'jtag' in _services:
                 if 'idf' in _services:
                     from pytest_embedded_idf import IdfDut
@@ -400,9 +408,9 @@ def _fixture_classes_and_options_fn(
                         'serial': None,
                     })
                 elif 'nuttx' in _services:
-                    from pytest_embedded_nuttx import NuttxDut
+                    from pytest_embedded_nuttx import NuttxSerialDut
 
-                    classes[fixture] = NuttxDut
+                    classes[fixture] = NuttxSerialDut
                     kwargs[fixture].update({
                         'serial': None,
                     })