scripts: twister: add support for device testing via RTT

Add support for using RTT as one of the possible communication
transports (besides UART and serial pty), when executing tests
on hardware with Twister.

Users can enable this by:
- Adding both --device-testing and --device-rtt flags to the Twister
  command line, or
- Setting 'rtt' to True in the hardware map YAML file.

Implementation details:
- RTT support uses the west rtt command to connect to the device.
- If west rtt: should use a different runner, then --rtt-runner flag or
  rtt_runner field can be used.
- Renaming changes in handlers.py were done to remove the ser/serial
  prefix from the variables and functions related to pty, since
  they are now shared by both serial pty and RTT.

The current approach using the west rtt command has one limitation: it
is currently not possible to test multiple devices that use RTT in
parallel. west rtt command will always try a specific set of ports,
so running this command while another one is executing doesn't work.

Closes: #57120

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

Author: MarkoSagadin

doc/develop/test/twister.rst

@@ -1285,6 +1285,25 @@ In this case you can run twister with the following options:
 The script is user-defined and handles delivering the messages which can be
 used by twister to determine the test execution status.
 
+To support devices that communicate via RTT, use the ``--device-rtt`` option. Twister
+will connect to the device with ``west rtt`` command and capture the log messages.
+In this case you can run twister with the following options:
+
+.. tabs::
+
+   .. group-tab:: Linux
+
+      .. code-block:: bash
+
+         scripts/twister --device-testing --device-rtt \
+         -p nrf7002dk/nrf5340/cpuapp -T tests/kernel
+
+   .. group-tab:: Windows
+
+      .. note::
+
+         Not supported on Windows OS
+
 The ``--device-flash-timeout`` option allows to set explicit timeout on the
 device flash operation, for example when device flashing takes significantly
 large time.
@@ -1480,6 +1499,23 @@ work. It is equivalent to following west and twister commands.
   manually according to above example. This is because the serial port
   of the PTY is not fixed and being allocated in the system at runtime.
 
+RTT support using ``--device-rtt``  can also be used in the
+hardware map:
+
+.. code-block:: yaml
+
+   - connected: true
+     id: 001050795550
+     platform: nrf7002dk/nrf5340/cpuapp
+     product: J-Link
+     runner: nrfutil
+     rtt: true
+     rtt_runner: jlink
+
+If a different runner should be used for RTT connection than for flashing, specify
+``rtt_runner`` field, like shown above. If a different runner is not needed, then
+``rtt_runner`` can be omitted.
+
 If west is not available or does not know how to flash your system, a custom
 flash command can be specified using the ``flash-command`` flag. The script is
 called with a ``--build-dir`` with the path of the current build, as well as a

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

@@ -26,6 +26,9 @@ Run exemplary test shell application by Twister:
   # hardware
   ./scripts/twister -p nrf52840dk/nrf52840 --device-testing --device-serial /dev/ttyACM0 -T samples/subsys/testsuite/pytest/shell
 
+  # hardware over RTT
+  ./scripts/twister -p nrf52840dk/nrf52840 --device-testing --device-rtt -T samples/subsys/testsuite/pytest/shell --test sample.pytest.rtt
+
 or build shell application by west and call pytest directly:
 
 .. code-block:: sh

scripts/pylib/twister/twisterlib/environment.py

@@ -174,8 +174,9 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
 
     run_group_option.add_argument(
         "--device-testing", action="store_true",
-        help="Test on device directly. Specify the serial device to "
-             "use with the --device-serial option.")
+        help="Test on device directly. Specify the serial device to use with "
+            "the --device-serial option or specify script to use with the "
+            "--device-serial-pty or specify to use with --device-rtt option.")
 
     run_group_option.add_argument("--generate-hardware-map",
                         help="""Probe serial devices connected to this platform
@@ -204,6 +205,15 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
                         --device-serial-pty <script>
                         """)
 
+    device.add_argument("--device-rtt", action="store_true", default=False,
+                        help="""Use RTT as communication transport.
+                        Instead of using serial console, Twister will connect to the
+                        device with "west rtt" command to capture logs and interact with
+                        the test application. Setting this option automatically sets
+                        --flash-before option, as we can connect to the device only
+                        after it was flashed.
+                        """)
+
     device.add_argument("--hardware-map",
                         help="""Load hardware map from a file. This will be used
                         for testing on hardware that is listed in the file.
@@ -959,20 +969,23 @@ def parse_arguments(
 
     if (
         (not options.device_testing)
-        and (options.device_serial or options.device_serial_pty or options.hardware_map)
+        and (options.device_serial or options.device_serial_pty or options.device_rtt or
+             options.hardware_map)
     ):
         logger.error(
-            "Use --device-testing with --device-serial, or --device-serial-pty, or --hardware-map."
+            "Use --device-testing with --device-serial, or --device-serial-pty, or "
+            "--device-rtt, or --hardware-map."
         )
         sys.exit(1)
 
     if (
         options.device_testing
-        and (options.device_serial or options.device_serial_pty) and len(options.platform) != 1
+        and (options.device_serial or options.device_serial_pty or
+             options.device_rtt) and len(options.platform) != 1
     ):
         logger.error("When --device-testing is used with --device-serial "
-                     "or --device-serial-pty, exactly one platform must "
-                     "be specified")
+                     "or --device-serial-pty, or --device-rtt, "
+                     "exactly one platform must be specified")
         sys.exit(1)
 
     if options.device_flash_with_test and not options.device_testing:

scripts/pylib/twister/twisterlib/handlers.py

@@ -515,7 +515,8 @@ def device_is_available(self, instance):
         for d in self.duts:
             if fixture and fixture not in map(lambda f: f.split(sep=':')[0], d.fixtures):
                 continue
-            if d.platform != device or (d.serial is None and d.serial_pty is None):
+            if d.platform != device or (d.serial is None and d.serial_pty is None and
+                                        d.use_rtt is False):
                 continue
             duts_found.append(d)
 
@@ -565,7 +566,7 @@ def run_custom_script(script, timeout):
                 proc.communicate()
                 logger.error(f"{script} timed out")
 
-    def _create_flash_command(self, hardware):
+    def _create_flash_command_from_option(self, hardware):
         flash_command = next(csv.reader([self.options.flash_command]))
 
         command = [flash_command[0]]
@@ -579,9 +580,28 @@ def _create_flash_command(self, hardware):
 
         return command
 
-    def _create_command(self, runner, hardware):
+    def _create_rtt_command(self, hardware):
+
+        command = ["west", "rtt", "--skip-rebuild", "-d", self.build_dir]
+        rtt_runner = hardware.rtt_runner
+
+        if rtt_runner:
+            command.append("--runner")
+            command.append(rtt_runner)
+
+            if rtt_runner in ("jlink", "pyocd"):
+                board_id = hardware.probe_id or hardware.id
+
+                command.append("--dev-id")
+                command.append(board_id)
+
+        # Since _start_pty expects a string, we need to convert the command list into
+        # one.
+        return " ".join(command)
+
+    def _create_flash_command(self, runner, hardware):
         if self.options.flash_command:
-            return self._create_flash_command(hardware)
+            return self._create_flash_command_from_option(hardware)
 
         command = ["west"]
         if self.options.verbose > 2:
@@ -648,18 +668,18 @@ def _create_command(self, runner, hardware):
 
         return command
 
-    def _terminate_pty(self, ser_pty, ser_pty_process):
-        logger.debug(f"Terminating serial-pty:'{ser_pty}'")
-        terminate_process(ser_pty_process)
+    def _terminate_pty(self, pty_name, pty_process):
+        logger.debug(f"Terminating pty:'{pty_name}'")
+        terminate_process(pty_process)
         try:
-            (stdout, stderr) = ser_pty_process.communicate(timeout=self.get_test_timeout())
-            logger.debug(f"Terminated serial-pty:'{ser_pty}', stdout:'{stdout}', stderr:'{stderr}'")
+            (stdout, stderr) = pty_process.communicate(timeout=self.get_test_timeout())
+            logger.debug(f"Terminated pty:'{pty_name}', stdout:'{stdout}', stderr:'{stderr}'")
         except subprocess.TimeoutExpired:
-            logger.debug(f"Terminated serial-pty:'{ser_pty}'")
-    #
+            logger.debug(f"Terminated pty:'{pty_name}'")
+
 
     def _create_serial_connection(self, dut, serial_device, hardware_baud,
-                                  flash_timeout, serial_pty, ser_pty_process):
+                                  flash_timeout, pty_name, pty_process):
         try:
             ser = serial.Serial(
                 serial_device,
@@ -671,20 +691,20 @@ def _create_serial_connection(self, dut, serial_device, hardware_baud,
                 timeout=max(flash_timeout, self.get_test_timeout())
             )
         except serial.SerialException as e:
-            self._handle_serial_exception(e, dut, serial_pty, ser_pty_process)
+            self._handle_serial_exception(e, dut, pty_name, pty_process)
             raise
 
         return ser
 
 
-    def _handle_serial_exception(self, exception, dut, serial_pty, ser_pty_process):
+    def _handle_serial_exception(self, exception, dut, pty_name, pty_process):
         self.instance.status = TwisterStatus.FAIL
         self.instance.reason = "Serial Device Error"
         logger.error(f"Serial device error: {exception!s}")
 
         self.instance.add_missing_case_status(TwisterStatus.BLOCK, "Serial Device Error")
-        if serial_pty and ser_pty_process:
-            self._terminate_pty(serial_pty, ser_pty_process)
+        if pty_name and pty_process:
+            self._terminate_pty(pty_name, pty_process)
 
         self.make_dut_available(dut)
 
@@ -706,21 +726,21 @@ def get_hardware(self):
             logger.error(self.instance.reason)
         return hardware
 
-    def _start_serial_pty(self, serial_pty, serial_pty_master):
-        ser_pty_process = None
+    def _start_pty(self, command, pty_master):
+        pty_process = None
         try:
-            ser_pty_process = subprocess.Popen(
-                re.split('[, ]', serial_pty),
-                stdout=serial_pty_master,
-                stdin=serial_pty_master,
-                stderr=serial_pty_master
+            pty_process = subprocess.Popen(
+                re.split('[, ]', command),
+                stdout=pty_master,
+                stdin=pty_master,
+                stderr=pty_master
             )
         except subprocess.CalledProcessError as error:
             logger.error(
-                f"Failed to run subprocess {serial_pty}, error {error.output}"
+                f"Failed to run subprocess {command}, error {error.output}"
             )
 
-        return ser_pty_process
+        return pty_process
 
     def handle(self, harness):
         runner = None
@@ -732,16 +752,19 @@ def handle(self, harness):
 
         runner = hardware.runner or self.options.west_runner
         serial_pty = hardware.serial_pty
-
-        if not serial_pty:
-            serial_device = hardware.serial
+        use_rtt  = hardware.use_rtt
+        pty_name = None
+
+        if serial_pty or use_rtt:
+            pty_master, slave = pty.openpty()
+            pty_name = os.ttyname(slave)
+            reason = "RTT" if use_rtt else "serial PTY"
+            logger.debug(f"Using pty {pty_name} for {reason}")
         else:
-            ser_pty_master, slave = pty.openpty()
-            serial_device = os.ttyname(slave)
-
-        logger.debug(f"Using serial device {serial_device} @ {hardware.baud} baud")
+            serial_device = hardware.serial
+            logger.debug(f"Using serial device {serial_device} @ {hardware.baud} baud")
 
-        command = self._create_command(runner, hardware)
+        flash_command = self._create_flash_command(runner, hardware)
 
         pre_script = hardware.pre_script
         post_flash_script = hardware.post_flash_script
@@ -759,20 +782,24 @@ def handle(self, harness):
             flash_timeout += self.get_test_timeout()
 
         serial_port = None
-        ser_pty_process = None
+        pty_process = None
         if hardware.flash_before is False:
             if serial_pty:
-                ser_pty_process = self._start_serial_pty(serial_pty, ser_pty_master)
-            serial_port = serial_device
+                pty_process = self._start_pty(serial_pty, pty_master)
+                serial_port = pty_name
+            else:
+                serial_port = serial_device
 
+        # Open connection to serial device or if using serial pty just create ser object
+        # and open it later
         try:
             ser = self._create_serial_connection(
                 hardware,
                 serial_port,
                 hardware.baud,
                 flash_timeout,
                 serial_pty,
-                ser_pty_process
+                pty_process
             )
         except serial.SerialException:
             return
@@ -785,11 +812,13 @@ def handle(self, harness):
         t.start()
 
         d_log = f"{self.instance.build_dir}/device.log"
-        logger.debug(f'Flash command: {command}', )
+        logger.debug(f'Flash command: {flash_command}')
         failure_type = Handler.FailureType.NONE
         try:
             stdout = stderr = None
-            with subprocess.Popen(command, stderr=subprocess.PIPE, stdout=subprocess.PIPE) as proc:
+            with subprocess.Popen(flash_command,
+                                  stderr=subprocess.PIPE,
+                                  stdout=subprocess.PIPE) as proc:
                 try:
                     (stdout, stderr) = proc.communicate(timeout=flash_timeout)
                     # ignore unencodable unicode chars
@@ -828,10 +857,18 @@ def handle(self, harness):
         # Connect to device after flashing it
         if hardware.flash_before:
             try:
-                if serial_pty:
-                    ser_pty_process = self._start_serial_pty(serial_pty, ser_pty_master)
-                logger.debug(f"Attach serial device {serial_device} @ {hardware.baud} baud")
-                ser.port = serial_device
+                if use_rtt:
+                    rtt_command = self._create_rtt_command(hardware)
+                    pty_process = self._start_pty(rtt_command, pty_master)
+                    logger.debug(f"Start RTT connection on pty {pty_name}")
+                    ser.port = pty_name
+                elif serial_pty:
+                    pty_process = self._start_pty(serial_pty, pty_master)
+                    logger.debug(f"Start serial connection on pty {pty_name}")
+                    ser.port = pty_name
+                else:
+                    logger.debug(f"Attach serial device {serial_device} @ {hardware.baud} baud")
+                    ser.port = serial_device
 
                 # Apply ESP32-specific RTS/DTR reset logic
                 if runner == "esp32":
@@ -854,7 +891,7 @@ def handle(self, harness):
                     ser.open()
 
             except serial.SerialException as e:
-                self._handle_serial_exception(e, hardware, serial_pty, ser_pty_process)
+                self._handle_serial_exception(e, hardware, pty_name, pty_process)
                 return
 
         if failure_type != Handler.FailureType.FLASH:
@@ -876,8 +913,8 @@ def handle(self, harness):
         if ser.isOpen():
             ser.close()
 
-        if serial_pty:
-            self._terminate_pty(serial_pty, ser_pty_process)
+        if pty_name:
+            self._terminate_pty(pty_name, pty_process)
 
         self.execution_time = time.time() - start_time