tests: drivers: uart: Add pytest based uart passthrough test

Verify the passtrough function with the automated two terminal test

Signed-off-by: Bartosz Miller <[email protected]>

Author: nordic-bami

tests/drivers/uart/uart_passtrough/CMakeLists.txt

@@ -0,0 +1,13 @@
+#
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+#
+
+cmake_minimum_required(VERSION 3.20.0)
+
+find_package(Zephyr REQUIRED HINTS $ENV{ZEPHYR_BASE})
+
+project(uart_passtrough)
+
+target_sources(app PRIVATE src/main.c)

tests/drivers/uart/uart_passtrough/boards/nrf5340dk_nrf5340_cpuapp.overlay

@@ -0,0 +1,18 @@
+/*
+ * Copyright (c) 2025 Nordic Semiconductor ASA
+ *
+ * SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+ */
+
+/ {
+	chosen {
+		uart,passthrough = &uart1;
+	};
+};
+
+&uart1 {
+	compatible = "nordic,nrf-uarte";
+	current-speed = <115200>;
+	status = "okay";
+	hw-flow-control;
+};

tests/drivers/uart/uart_passtrough/boards/nrf54h20dk_nrf54h20_cpuapp.overlay

@@ -0,0 +1,19 @@
+/*
+ * Copyright (c) 2025 Nordic Semiconductor ASA
+ *
+ * SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+ */
+
+/ {
+	chosen {
+		uart,passthrough = &uart135;
+	};
+};
+
+&uart135 {
+	compatible = "nordic,nrf-uarte";
+	current-speed = <115200>;
+	status = "okay";
+	memory-regions = <&cpuapp_dma_region>;
+	hw-flow-control;
+};

tests/drivers/uart/uart_passtrough/prj.conf

@@ -0,0 +1,2 @@
+CONFIG_SERIAL=y
+CONFIG_UART_INTERRUPT_DRIVEN=y

tests/drivers/uart/uart_passtrough/pytest/serial_port.py

@@ -0,0 +1,104 @@
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+
+
+import logging
+import time
+
+import serial
+
+logger = logging.getLogger("serial_port")
+logger.setLevel(logging.DEBUG)
+
+
+class SerialPort:
+    """
+    Class for handling serial port
+    Default settings:
+    8 data bits, 1 stop bit, no parity
+    """
+
+    def __init__(self, serial_port: str, baudrate: int, timeout: int = 15):
+        """
+        :param serial_port: full name of the serial device port
+        :param baudrate: buadrate [bps]
+        :param timeout: read/write timeout [s]
+        """
+        self.port_name = serial_port
+        self.baud_rate = baudrate
+        self._timeout = timeout
+        self.serial_port = self._instanitate_serial_port()
+
+    def _instanitate_serial_port(self) -> serial.Serial:
+        """
+        Auxiliary method for handling serial port handler object instantiation
+        """
+        try:
+            port = serial.Serial(
+                port=self.port_name, baudrate=self.baud_rate, timeout=self._timeout, rtscts=True
+            )
+            port.set_low_latency_mode(True)
+            return port
+        except serial.SerialException as exc:
+            logger.error(f"Opening '{self.port_name}' failed with error: {exc}!")
+            raise exc
+
+    def close(
+        self,
+    ):
+        """
+        Close serial port
+        """
+        if self.serial_port:
+            logger.debug("Closing serial port connection")
+            self.serial_port.close()
+            logger.debug("Serial port connection closed")
+            self.serial_port = None
+
+    def open(self) -> None:
+        """
+        Open serial port
+        """
+        if not self.serial_port.is_open:
+            logger.debug(f"Opening: '{self}'")
+            self.serial_port.open()
+            self.serial_port.reset_input_buffer()
+            self.serial_port.reset_output_buffer()
+
+    def send(self, message: str, get_response: bool = False) -> str:
+        """
+        Send one message (and optionally receive response)
+        :param get_response: if True returns the received response
+        :return: received response or empty string
+        """
+        response: str = ""
+        if self.serial_port.is_open is False:
+            raise IOError("Serial port is closed")
+
+        try:
+            time.sleep(0.25)
+            self.serial_port.write_timeout = self._timeout
+            logger.info(f"[{self.serial_port.port}] Serial --> {message}")
+            self.serial_port.write((message + "\n").encode())
+            if get_response:
+                start = time.time()
+                while (self.serial_port.in_waiting == 0) and (time.time() - start < self._timeout):
+                    # no data in buffer yet
+                    time.sleep(0.1)
+                while (self.serial_port.in_waiting > 0) and (time.time() - start < self._timeout):
+                    time.sleep(0.1)
+                    response += self.serial_port.read_all().decode()
+                logger.info(f"Serial <-- {response}")
+            return response
+
+        except serial.SerialTimeoutException as exc:
+            logger.error(f"Serial port read timeout: {exc}")
+
+        except serial.SerialException as exc:
+            logger.error(f"Serial port read error: {exc}")
+
+        except UnicodeDecodeError as exc:
+            logger.error(f"Decoding error: {exc}")
+
+        return response.rstrip("\n")

tests/drivers/uart/uart_passtrough/pytest/test_passtrough.py

@@ -0,0 +1,37 @@
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+
+import logging
+
+logger = logging.getLogger("uart_passtrough")
+logger.setLevel(logging.DEBUG)
+
+from twister_harness import DeviceAdapter
+from serial_port import SerialPort
+
+
+def test_uart_passtrough(dut: DeviceAdapter):
+    """
+    Verify the UART darta forwarding capability.
+    Send message from 'passtrough' UART to 'console 'UART' and verify it,
+    then send message from 'console' to 'passtrough' UART and verify it.
+    """
+
+    if "if00" in dut.device_config.serial:
+        second_serial_port_name = dut.device_config.serial.replace("if00", "if02")
+    elif "if02" in dut.device_config.serial:
+        second_serial_port_name = dut.device_config.serial.replace("if02", "if00")
+
+    second_serial_port: SerialPort = SerialPort(
+        serial_port=second_serial_port_name, baudrate=dut.device_config.baud
+    )
+
+    dut.readlines_until(regex="Ready", print_output=True, timeout=2)
+    second_serial_port.open()
+    second_serial_port.send("con->pass")
+    dut.readlines_until(regex="con->pass", print_output=True, timeout=2)
+    dut.write("pass<-con".encode())
+    second_port_data: str = second_serial_port.send("", get_response=True)
+    assert "pass<-con" in second_port_data
+    second_serial_port.close()

tests/drivers/uart/uart_passtrough/src/main.c

@@ -0,0 +1,98 @@
+/*
+ * Copyright (c) 2025 Nordic Semiconductor ASA
+ *
+ * SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+ */
+
+#include <zephyr/kernel.h>
+#include <zephyr/device.h>
+#include <zephyr/drivers/uart.h>
+
+const struct device *const console_dev = DEVICE_DT_GET(DT_CHOSEN(zephyr_console));
+const struct device *const passtrough_dev = DEVICE_DT_GET(DT_CHOSEN(uart_passthrough));
+
+typedef struct {
+	uint8_t *rx_buffer;
+	uint8_t *tx_buffer;
+	const struct device *other_dev;
+
+} uart_pass;
+
+/*
+ * Note: this simple test uses one byte buffer,
+ * CPU must be able to receive and send one character,
+ * before another one is received
+ * This limits the usable buadrates range
+ */
+
+static uint8_t console_buf[1];
+static uint8_t passthrough_buf[1];
+
+void uart_tx_interrupt_service(const struct device *dev, uart_pass *test_data)
+{
+	uart_fifo_fill(dev, test_data->tx_buffer, 1);
+	uart_irq_tx_disable(dev);
+}
+
+void uart_rx_interrupt_service(const struct device *dev, uart_pass *test_data)
+{
+	int rx_data_length = 0;
+
+	do {
+		rx_data_length = uart_fifo_read(dev, test_data->rx_buffer, 1);
+	} while (rx_data_length);
+	uart_irq_tx_enable(test_data->other_dev);
+}
+
+void uart_isr_handler(const struct device *dev, void *user_data)
+{
+	uart_pass *test_data = (uart_pass *)user_data;
+
+	uart_irq_update(dev);
+	while (uart_irq_is_pending(dev)) {
+		if (uart_irq_rx_ready(dev)) {
+			uart_rx_interrupt_service(dev, test_data);
+		}
+		if (uart_irq_tx_ready(dev)) {
+			uart_tx_interrupt_service(dev, test_data);
+		}
+	}
+}
+
+int main(void)
+{
+	int err;
+
+	uart_pass console_dev_data;
+	uart_pass passtrough_dev_data;
+
+	console_dev_data.rx_buffer = console_buf;
+	console_dev_data.tx_buffer = passthrough_buf;
+	console_dev_data.other_dev = passtrough_dev;
+
+	err = uart_irq_callback_set(console_dev, uart_isr_handler);
+	printk("uart_irq_callback_set(console_dev) err=%d\n", err);
+	err = uart_irq_callback_user_data_set(console_dev, uart_isr_handler,
+					      (void *)&console_dev_data);
+	printk("uart_irq_callback_user_data_set(console_dev) err=%d\n", err);
+
+	passtrough_dev_data.rx_buffer = passthrough_buf;
+	passtrough_dev_data.tx_buffer = console_buf;
+	passtrough_dev_data.other_dev = console_dev;
+
+	err = uart_irq_callback_set(passtrough_dev, uart_isr_handler);
+	printk("uart_irq_callback_set(passtrough_dev) err=%d\n", err);
+	err = uart_irq_callback_user_data_set(passtrough_dev, uart_isr_handler,
+					      (void *)&passtrough_dev_data);
+	printk("uart_irq_callback_user_data_set(passtrough_devle_dev) err=%d\n", err);
+
+	k_msleep(10);
+	printk("Ready\n");
+
+	uart_irq_rx_enable(console_dev);
+	uart_irq_rx_enable(passtrough_dev);
+
+	k_sleep(K_FOREVER);
+
+	return 0;
+}

tests/drivers/uart/uart_passtrough/testcase.yaml

@@ -0,0 +1,19 @@
+common:
+  sysbuild: true
+  depends_on: gpio
+  tags:
+    - drivers
+    - uart
+    - ci_tests_drivers_uart
+  harness: pytest
+  harness_config:
+    pytest_dut_scope: session
+  platform_allow:
+    - nrf5340dk/nrf5340/cpuapp
+    - nrf54h20dk/nrf54h20/cpuapp
+  integration_platforms:
+    - nrf5340dk/nrf5340/cpuapp
+    - nrf54h20dk/nrf54h20/cpuapp
+
+tests:
+  drivers.uart_passtrough: {}