scripts: pytest: add RTT support in pytest-twister-harness

This commit introduces --device-rtt flag in the Pytest Twister harness,
allowing RTT as a communication protocol between the host PC and the
DUT.

To demonstrate this, a new build configuration was added to the
pytest/shell sample. By following the newly added instructions in the
scripts/pylib/pytest-twister-harness/README.rst, users can first build
the pytest/shell sample with the RTT shell and then run the pytest
with specific flags to run the tests via the RTT.

Two additional notes related to the changes:
- The way the pty process was terminated had to be changed. west rtt
  command spawns two subprocesses, and the previous implementation only
  terminated a single one. The already provided terminate_process
  function was used to terminate all child processes that pty process
  started.
- Previously, exec_command sent out a command string, followed by two
  newline characters. The first one was meant to execute the command on
  the device, the second one was meant to produce the next prompt, which
  would signal that execution of the command finished. In some
  instances, when RTT was used, the device ignored the second newline,
  and thus the prompt was not emitted, bringing the function to a halt,
  until the timeout expired. This was noticed with the `kernel version`
  command, the one that is used in one of the tests in pytest/shell.
  Weirdly enough, that was the only instance of the "newline ignoring"
  that could be found. Testing `kernel uptime`, for example, worked
  fine. The workaround was to send a command string with a single
  newline, wait for the command print and then send the second newline.
  It was confirmed that this approach works both for the UART and RTT.

Signed-off-by: Marko Sagadin <[email protected]>

Author: MarkoSagadin

samples/subsys/testsuite/pytest/shell/testcase.yaml

@@ -18,6 +18,12 @@ tests:
     harness: pytest
     extra_configs:
       - CONFIG_SHELL_VT100_COLORS=n
+  sample.pytest.rtt:
+    harness: pytest
+    extra_configs:
+      - CONFIG_USE_SEGGER_RTT=y
+      - CONFIG_LOG_MODE_DEFERRED=y
+      - CONFIG_SHELL_BACKEND_RTT=y
   sample.harness.shell:
     harness: shell
     harness_config:

scripts/pylib/pytest-twister-harness/README.rst

@@ -45,3 +45,7 @@ or build shell application by west and call pytest directly:
   # hardware
   west build -p -b nrf52840dk/nrf52840
   pytest --twister-harness --device-type=hardware --device-serial=/dev/ttyACM0 --build-dir=build -p twister_harness.plugin
+
+  # hardware over RTT
+  west build -p -b nrf52840dk/nrf52840 . -T sample.pytest.rtt
+  pytest --twister-harness --device-type=hardware --device-rtt=True --build-dir=build -p twister_harness.plugin

scripts/pylib/pytest-twister-harness/src/twister_harness/device/device_adapter.py

@@ -80,9 +80,10 @@ def launch(self) -> None:
         self._device_run.set()
         self._start_reader_thread()
 
-        if self.device_config.flash_before:
-            # For hardware devices with shared USB or software USB, connect after flashing.
-            # Retry for up to 10 seconds for USB-CDC based devices to enumerate.
+        if self.device_config.flash_before or self.device_config.use_rtt:
+            # For hardware devices with shared USB or software USB or RTT, connect after
+            # flashing. Retry for up to 10 seconds for USB-CDC based devices to
+            # enumerate.
             self._flash_and_run()
             self.connect(retry_s = 10)
         else:

scripts/pylib/pytest-twister-harness/src/twister_harness/device/hardware_adapter.py

@@ -32,7 +32,7 @@ def __init__(self, device_config: DeviceConfig) -> None:
         super().__init__(device_config)
         self._flashing_timeout: float = device_config.flash_timeout
         self._serial_connection: serial.Serial | None = None
-        self._serial_pty_proc: subprocess.Popen | None = None
+        self._proc: subprocess.Popen | None = None
         self._serial_buffer: bytearray = bytearray()
 
         self.device_log_path: Path = device_config.build_dir / 'device.log'
@@ -76,6 +76,25 @@ def generate_command(self) -> None:
             command.extend(command_extra_args)
         self.command = command
 
+    def generate_rtt_command(self):
+        """Return command to connect to the device via RTT."""
+
+        command = ["west", "rtt", "--skip-rebuild", "-d",
+                               str(self.device_config.build_dir)]
+
+        rtt_runner = self.device_config.rtt_runner
+        if rtt_runner:
+            command.append("--runner")
+            command.append(rtt_runner)
+
+            if rtt_runner in ("jlink", "pyocd"):
+                command.append("--dev-id")
+                command.append(self.device_config.id)
+
+        # Since _start_pty expects a string, we need to convert the command list into
+        # one.
+        return " ".join(command)
+
     def _prepare_runner_args(self) -> tuple[list[str], list[str]]:
         base_args: list[str] = []
         extra_args: list[str] = []
@@ -154,7 +173,13 @@ def _flash_and_run(self) -> None:
                 raise TwisterHarnessException(msg)
 
     def _connect_device(self) -> None:
-        serial_name = self._open_serial_pty() or self.device_config.serial
+        if self.device_config.serial_pty:
+            serial_name = self._open_pty(self.device_config.serial_pty)
+        elif self.device_config.use_rtt:
+            serial_name = self._open_pty(self.generate_rtt_command())
+        else:
+            serial_name = self.device_config.serial
+
         logger.debug('Opening serial connection for %s', serial_name)
         try:
             self._serial_connection = serial.Serial(
@@ -167,17 +192,15 @@ def _connect_device(self) -> None:
             )
         except serial.SerialException as exc:
             logger.exception('Cannot open connection: %s', exc)
-            self._close_serial_pty()
+            self._close_pty()
             raise
 
         self._serial_connection.flush()
         self._serial_connection.reset_input_buffer()
         self._serial_connection.reset_output_buffer()
 
-    def _open_serial_pty(self) -> str | None:
-        """Open a pty pair, run process and return tty name"""
-        if not self.device_config.serial_pty:
-            return None
+    def _open_pty(self, command: str) -> str | None:
+        """Open a pty pair, run process with command and return tty name."""
 
         try:
             master, slave = pty.openpty()
@@ -186,32 +209,37 @@ def _open_serial_pty(self) -> str | None:
             raise exc
 
         try:
-            self._serial_pty_proc = subprocess.Popen(
-                re.split(',| ', self.device_config.serial_pty),
+            self._proc = subprocess.Popen(
+                re.split(',| ', command),
                 stdout=master,
                 stdin=master,
                 stderr=master
             )
         except subprocess.CalledProcessError as exc:
-            logger.exception('Failed to run subprocess %s, error %s', self.device_config.serial_pty, str(exc))
+            logger.exception('Failed to run subprocess %s, error %s', command, str(exc))
             raise
         return os.ttyname(slave)
 
     def _disconnect_device(self) -> None:
         if self._serial_connection:
             serial_name = self._serial_connection.port
             self._serial_connection.close()
-            # self._serial_connection = None
             logger.debug('Closed serial connection for %s', serial_name)
-        self._close_serial_pty()
+        self._close_pty()
+
+    def _close_pty(self) -> None:
+        """Terminate the process opened for serial pty script or RTT."""
+        if self._proc:
+            terminate_process(self._proc)
+            self._proc.communicate(timeout=self.base_timeout)
+
+            if self.device_config.serial_pty:
+                proc = self.device_config.serial_pty
+            else:
+                proc = "west rtt"
 
-    def _close_serial_pty(self) -> None:
-        """Terminate the process opened for serial pty script"""
-        if self._serial_pty_proc:
-            self._serial_pty_proc.terminate()
-            self._serial_pty_proc.communicate(timeout=self.base_timeout)
-            logger.debug('Process %s terminated', self.device_config.serial_pty)
-            self._serial_pty_proc = None
+            logger.debug('Process %s terminated', proc)
+            self._proc = None
 
     def _close_device(self) -> None:
         if self.device_config.post_script:
@@ -282,7 +310,7 @@ def _flush_device_output(self) -> None:
     def _clear_internal_resources(self) -> None:
         super()._clear_internal_resources()
         self._serial_connection = None
-        self._serial_pty_proc = None
+        self._proc = None
         self._serial_buffer.clear()
 
     @staticmethod

scripts/pylib/pytest-twister-harness/src/twister_harness/fixtures.py

@@ -73,8 +73,11 @@ def dut(request: pytest.FixtureRequest, device_object: DeviceAdapter) -> Generat
 def shell(dut: DeviceAdapter) -> Shell:
     """Return ready to use shell interface"""
     shell = Shell(dut, timeout=20.0)
+
+    symbol = 'CONFIG_SHELL_PROMPT_RTT' if dut.device_config.use_rtt else 'CONFIG_SHELL_PROMPT_UART'
+
     if prompt := find_in_config(Path(dut.device_config.app_build_dir) / 'zephyr' / '.config',
-                                'CONFIG_SHELL_PROMPT_UART'):
+                                symbol):
         shell.prompt = prompt
     logger.info('Wait for prompt')
     if not shell.wait_for_prompt():

scripts/pylib/pytest-twister-harness/src/twister_harness/helpers/shell.py

@@ -54,24 +54,30 @@ def exec_command(
         self, command: str, timeout: float | None = None, print_output: bool = True
     ) -> list[str]:
         """
-        Send shell command to a device and return response. Passed command
-        is extended by double enter sings - first one to execute this command
-        on a device, second one to receive next prompt what is a signal that
-        execution was finished. Method returns printout of the executed command.
+        Send shell command to a device and return response. Method returns printout of
+        the executed command.
         """
         timeout = timeout or self.base_timeout
-        command_ext = f'{command}\n\n'
+        command_ext = f'{command}\n'
         regex_prompt = re.escape(self.prompt)
         regex_command = f'.*{re.escape(command)}'
+
+        # Execute command
         self._device.clear_buffer()
         self._device.write(command_ext.encode())
         lines: list[str] = []
-        # wait for device command print - it should be done immediately after sending command to device
+        # wait for device command print - it should be done immediately after sending
+        # command to device.
         lines.extend(
             self._device.readlines_until(
                 regex=regex_command, timeout=1.0, print_output=print_output
             )
         )
+
+        # Send single enter to get next prompt after command execution, as it signals
+        # that execution finished.
+        self._device.write(b"\n")
+
         # wait for device command execution
         lines.extend(
             self._device.readlines_until(

scripts/pylib/pytest-twister-harness/src/twister_harness/plugin.py

@@ -70,13 +70,17 @@ def pytest_addoption(parser: pytest.Parser):
     )
     twister_harness_group.addoption(
         '--runner',
-        help='Use the specified west runner (pyocd, nrfjprog, etc.).'
+        help='Use the specified west runner (pyocd, nrfjprog, etc.) when running west flash.'
     )
     twister_harness_group.addoption(
         '--runner-params',
         action='append',
         help='Use the specified west runner params.'
     )
+    twister_harness_group.addoption(
+        '--rtt-runner',
+        help='Use the specified west runner (pyocd, nrfjprog, etc.) when running west rtt.'
+    )
     twister_harness_group.addoption(
         '--device-id',
         help='ID of connected hardware device (for example 000682459367).'
@@ -89,6 +93,11 @@ def pytest_addoption(parser: pytest.Parser):
         '--device-serial-pty',
         help='Script for controlling pseudoterminal.'
     )
+    twister_harness_group.addoption(
+        '--device-rtt',
+        type=bool,
+        help='Use RTT as communication transport.',
+    )
     twister_harness_group.addoption(
         '--flash-before',
         type=bool,

scripts/pylib/pytest-twister-harness/src/twister_harness/twister_harness_config.py

@@ -26,6 +26,8 @@ class DeviceConfig:
     baud: int = 115200
     runner: str = ''
     runner_params: list[str] = field(default_factory=list, repr=False)
+    rtt_runner: str = ''
+    use_rtt: bool = False
     id: str = ''
     product: str = ''
     serial_pty: str = ''
@@ -78,6 +80,8 @@ def create(cls, config: pytest.Config) -> TwisterHarnessConfig:
             baud=config.option.device_serial_baud,
             runner=config.option.runner,
             runner_params=runner_params,
+            rtt_runner=config.option.rtt_runner,
+            use_rtt=config.option.device_rtt,
             id=config.option.device_id,
             product=config.option.device_product,
             serial_pty=config.option.device_serial_pty,

scripts/pylib/pytest-twister-harness/tests/device/hardware_adapter_test.py

@@ -227,6 +227,7 @@ def test_if_hardware_adapter_uses_serial_pty(
 
     monkeypatch.setattr('twister_harness.device.hardware_adapter.pty.openpty', lambda: (123, 456))
     monkeypatch.setattr('twister_harness.device.hardware_adapter.os.ttyname', lambda x: f'/pty/ttytest/{x}')
+    monkeypatch.setattr('twister_harness.device.hardware_adapter.terminate_process', lambda x: None)
 
     serial_mock = mock.Mock()
     serial_mock.port = '/pty/ttytest/456'
@@ -235,7 +236,7 @@ def test_if_hardware_adapter_uses_serial_pty(
     device._device_run.set()
     device.connect()
     assert device._serial_connection.port == '/pty/ttytest/456'  # type: ignore[union-attr]
-    assert device._serial_pty_proc
+    assert device._proc
     patched_popen.assert_called_with(
         ['script.py'],
         stdout=123,
@@ -244,7 +245,7 @@ def test_if_hardware_adapter_uses_serial_pty(
     )
 
     device.disconnect()
-    assert not device._serial_pty_proc
+    assert not device._proc
 
 
 def test_if_hardware_adapter_properly_send_data_to_subprocess(