tests: on_target: introduce prepare_dut fixture and UART live output

Add a shared `prepare_dut` fixture to `conftest.py` that consolidates
the repeated flash/factory-reset/boot sequence used across all tests.
All test files are updated to use it instead of duplicating the steps.

Add structured UART live output control with a pretest suppression
phase: output is suppressed during setup and automatically re-enabled
when the boot banner is detected or a timeout is hit. Add
`wait_for_app_boot` as a named helper for the boot banner check.

Add lifecycle hooks that print structured markers around each test
phase, and enable `--capture=tee-sys` in CI to capture and show
stdout in test reports.

Signed-off-by: Simen S. Røstad <simen.rostad@nordicsemi.no>

Author: simensrostad

.github/workflows/target-test.yml

@@ -159,6 +159,7 @@ jobs:
           fi
 
           pytest -v "${pytest_marker[@]}" \
+            --capture=tee-sys \
             --junit-xml=results/test-results.xml \
             --html=results/test-results.html --self-contained-html \
             ${PYTEST_PATH}

tests/on_target/tests/conftest.py

@@ -7,14 +7,15 @@
 import re
 import pytest
 import types
-from utils.flash_tools import recover_device
+from utils.flash_tools import recover_device, flash_device, reset_device
 from utils.uart import Uart, UartBinary
 import sys
 sys.path.append(os.getcwd())
 from utils.logger import get_logger
 from utils.nrfcloud import NRFCloud, NRFCloudFOTA
 
 logger = get_logger()
+TEST_OUTCOMES = {}
 
 UART_TIMEOUT = 60 * 30
 
@@ -30,10 +31,8 @@ def get_uarts():
 
     if platform.system() == "Darwin":  # macOS
         base_path = "/dev"
-        pattern = "tty.*"
     else:  # Linux
         base_path = "/dev/serial/by-id"
-        pattern = None
 
     try:
         if platform.system() == "Darwin":
@@ -57,12 +56,87 @@ def scan_log_for_assertions(log):
 
 @pytest.hookimpl(tryfirst=True)
 def pytest_runtest_logstart(nodeid, location):
+    print(f"\n===== TEST_ITEM_START {nodeid} =====", flush=True)
     logger.info(f"Starting test: {nodeid}")
 
 @pytest.hookimpl(trylast=True)
 def pytest_runtest_logfinish(nodeid, location):
+    print(f"===== TEST_ITEM_FINISH {nodeid} =====\n", flush=True)
     logger.info(f"Finished test: {nodeid}")
 
+@pytest.hookimpl(tryfirst=True)
+def pytest_runtest_setup(item):
+    print(f"----- TEST_SETUP_START {item.nodeid} -----", flush=True)
+
+@pytest.hookimpl(tryfirst=True)
+def pytest_runtest_call(item):
+    print(f"----- TEST_CALL_START {item.nodeid} -----", flush=True)
+    # Tests using prepare_dut enable UART output after setup completes.
+    if "prepare_dut" not in item.fixturenames:
+        uart = _get_uart_for_item(item)
+        if uart:
+            uart.begin_test_run(item.nodeid)
+
+@pytest.hookimpl(tryfirst=True)
+def pytest_runtest_teardown(item, nextitem):
+    status = TEST_OUTCOMES.get(item.nodeid, "unknown")
+    uart = _get_uart_for_item(item)
+    if uart:
+        if uart.is_pretest_phase_active():
+            uart.end_pretest_phase(reason="teardown_fallback")
+        uart.mark(f"UART_TEST_END {item.nodeid} status={status}")
+        uart.set_live_output(False)
+    print(f"----- TEST_TEARDOWN_END {item.nodeid} status={status} -----", flush=True)
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    outcome = yield
+    report = outcome.get_result()
+    if report.when == "setup" and report.failed:
+        TEST_OUTCOMES[item.nodeid] = "failed_setup"
+    elif report.when == "call":
+        TEST_OUTCOMES[item.nodeid] = report.outcome
+        print(f"----- TEST_CALL_END {item.nodeid} outcome={report.outcome} -----", flush=True)
+    elif report.when == "teardown" and item.nodeid not in TEST_OUTCOMES:
+        TEST_OUTCOMES[item.nodeid] = report.outcome
+
+def _get_uart_for_item(item):
+    funcargs = getattr(item, "funcargs", {})
+    for fixture_name in ("dut_cloud", "dut_fota", "dut_board", "dut_traces"):
+        fixture = funcargs.get(fixture_name)
+        if fixture and hasattr(fixture, "uart"):
+            return fixture.uart
+    return None
+
+@pytest.fixture(scope="function")
+def prepare_dut(request):
+    nodeid = request.node.nodeid
+
+    def _prepare_dut(dut, hex_file_path, boot_timeout=240, max_boot_retries=3):
+        uart = dut.uart
+        uart.begin_pretest_phase(nodeid)
+        try:
+            flash_device(os.path.abspath(hex_file_path))
+            uart.xfactoryreset()
+            uart.flush()
+            reset_device()
+            for attempt in range(max_boot_retries):
+                try:
+                    uart.wait_for_app_boot(timeout=boot_timeout)
+                    break
+                except AssertionError:
+                    if attempt == max_boot_retries - 1:
+                        raise
+                    reset_device()
+            uart.end_pretest_phase(reason="setup_complete")
+            uart.begin_test_run(nodeid)
+        except Exception:
+            if uart.is_pretest_phase_active():
+                uart.end_pretest_phase(reason="setup_failed")
+            raise
+
+    return _prepare_dut
+
 @pytest.fixture(scope="session", autouse=True)
 def _purge_pending_fota_jobs():
     """Cancel leftover FOTA jobs queued for the test device before any test runs. """
@@ -80,6 +154,7 @@ def dut_board():
         pytest.fail("No UARTs found")
     log_uart_string = all_uarts[0]
     uart = Uart(log_uart_string, timeout=UART_TIMEOUT)
+    uart.set_live_output(False)
 
     yield types.SimpleNamespace(
         uart=uart,

tests/on_target/tests/test_functional/test_buffer_flash.py

@@ -7,7 +7,7 @@
 import sys
 import pytest
 from time import sleep
-from utils.flash_tools import flash_device, reset_device
+from utils.flash_tools import reset_device
 
 sys.path.append(os.getcwd())
 from utils.logger import get_logger
@@ -26,7 +26,7 @@ def get_header_str(datatype):
     return "Header file /att_storage/" + datatype + ".header already exists"
 
 @pytest.mark.slow
-def test_buffer_flash(dut_cloud, hex_file_buffer_flash):
+def test_buffer_flash(dut_cloud, hex_file_buffer_flash, prepare_dut):
 
     # Clear shadow config and command sections before starting the test to ensure a clean slate
     # and deterministic behavior.
@@ -39,8 +39,7 @@ def test_buffer_flash(dut_cloud, hex_file_buffer_flash):
         storage_threshold=FLASH_BUFFER_TEST_STORAGE_THRESHOLD
     )
 
-    flash_device(os.path.abspath(hex_file_buffer_flash))
-    dut_cloud.uart.xfactoryreset()
+    prepare_dut(dut_cloud, hex_file_buffer_flash)
 
     clear_str = "att_storage clear\r\n"
 
@@ -57,10 +56,8 @@ def test_buffer_flash(dut_cloud, hex_file_buffer_flash):
     ]
 
     try:
-        reset_device()
-        start_pos = dut_cloud.uart.get_size()
-
         # Wait for storage automount
+        start_pos = dut_cloud.uart.get_size()
         dut_cloud.uart.wait_for_str("Automount /att_storage succeeded", timeout=60, start_pos=start_pos)
 
         # Clear buffer

tests/on_target/tests/test_functional/test_buffer_ram.py

@@ -6,7 +6,7 @@
 import os
 import sys
 import pytest
-from utils.flash_tools import flash_device, reset_device
+from utils.flash_tools import reset_device
 
 sys.path.append(os.getcwd())
 from utils.logger import get_logger
@@ -26,7 +26,7 @@ def get_storing_str(datatype):
     return f"Stored {datatype} item, count: 1"
 
 @pytest.mark.slow
-def test_buffer_ram(dut_cloud, hex_file_buffer_ram):
+def test_buffer_ram(dut_cloud, hex_file_buffer_ram, prepare_dut):
 
     # Clear shadow config and command sections before starting the test to ensure a clean slate
     # and deterministic behavior.
@@ -41,8 +41,7 @@ def test_buffer_ram(dut_cloud, hex_file_buffer_ram):
 
 
     try:
-        flash_device(os.path.abspath(hex_file_buffer_ram))
-        dut_cloud.uart.xfactoryreset()
+        prepare_dut(dut_cloud, hex_file_buffer_ram)
 
         initialization_list = [
                 get_initialized_str("BATTERY"),
@@ -56,9 +55,6 @@ def test_buffer_ram(dut_cloud, hex_file_buffer_ram):
                 get_storing_str("LOCATION")
         ]
 
-        dut_cloud.uart.flush()
-        reset_device()
-
         # Wait for storage initialization
         dut_cloud.uart.wait_for_str(initialization_list, timeout=60)
 

tests/on_target/tests/test_functional/test_config.py

@@ -5,7 +5,7 @@
 
 import os
 import time
-from utils.flash_tools import flash_device, reset_device
+from utils.flash_tools import reset_device
 import sys
 sys.path.append(os.getcwd())
 from utils.logger import get_logger
@@ -59,7 +59,7 @@ def wait_for_config_reported(cloud, device_id, expected_sample, expected_thresho
     )
 
 
-def test_config(dut_cloud, hex_file):
+def test_config(dut_cloud, hex_file, prepare_dut):
     '''
     Test that verifies shadow changes are applied and reported back for all config
     parameters. Tests both boot-time config pickup and runtime config updates.
@@ -77,10 +77,7 @@ def test_config(dut_cloud, hex_file):
         storage_threshold=BOOT_STORAGE_THRESHOLD
     )
 
-    flash_device(os.path.abspath(hex_file))
-    dut_cloud.uart.xfactoryreset()
-    dut_cloud.uart.flush()
-    reset_device()
+    prepare_dut(dut_cloud, hex_file)
 
     dut_cloud.uart.wait_for_str_with_retries(
         "Connected to Cloud",

tests/on_target/tests/test_functional/test_fota.py

@@ -110,13 +110,13 @@ def run_fota_resumption(dut_fota, fota_type):
         # First disconnect at 20%
         timeout_20_percent = FULL_MFW_FOTA_TIMEOUT * 0.2
         dut_fota.uart.wait_for_str("20%", timeout=timeout_20_percent)
-        logger.info("Performing first disconnect/reconnect at 20%")
+        logger.info("Performing first disconnect/reconnect at 20 percent")
         perform_disconnect_reconnect(dut_fota, 20)
 
         # Second disconnect at 80%
         timeout_80_percent = FULL_MFW_FOTA_TIMEOUT * 0.6  # Additional 60% of total timeout
         dut_fota.uart.wait_for_str("80%", timeout=timeout_80_percent)
-        logger.info("Performing second disconnect/reconnect at 80%")
+        logger.info("Performing second disconnect/reconnect at 80 percent")
         perform_disconnect_reconnect(dut_fota, 80)
 
 def run_fota_reschedule(dut_fota, fota_type):
@@ -153,12 +153,9 @@ def run_fota_reschedule(dut_fota, fota_type):
         raise AssertionError(f"Fota update not available after {i} attempts")
 
 @pytest.fixture
-def run_fota_fixture(dut_fota, hex_file, reschedule=False):
+def run_fota_fixture(dut_fota, hex_file, prepare_dut, reschedule=False):
     def _run_fota(bundle_id="", fota_type="app", fotatimeout=APP_FOTA_TIMEOUT, new_version=TEST_APP_VERSION, reschedule=False):
-        flash_device(os.path.abspath(hex_file))
-        dut_fota.uart.xfactoryreset()
-        dut_fota.uart.flush()
-        reset_device()
+        prepare_dut(dut_fota, hex_file)
 
         dut_fota.uart.wait_for_str_with_retries("Connected to Cloud", max_retries=3, timeout=240, reset_func=reset_device)
 

tests/on_target/tests/test_functional/test_memfault.py

@@ -8,7 +8,7 @@
 import requests
 import time
 import subprocess
-from utils.flash_tools import flash_device, reset_device
+from utils.flash_tools import reset_device
 from utils.logger import get_logger
 from datetime import datetime, timezone
 
@@ -26,7 +26,7 @@
 def convert_binary_trace_to_pcap(binary_trace, pcapng_file):
     logger.info(f"Converting modem trace to pcap")
     try:
-        result = subprocess.run(
+        subprocess.run(
             ['nrfutil', 'trace', 'lte', '--input-file', binary_trace, '--output-pcapng', pcapng_file],
             check=True,
             text=True,
@@ -64,7 +64,7 @@ def fetch_recent_modem_trace(device_id, start_time, end_time):
         reason = recording.get("reason")
 
         if reason == "modem traces":
-                return recording.get("id")
+            return recording.get("id")
 
     return None
 
@@ -89,17 +89,14 @@ def timestamp(event):
     )
 
 @pytest.mark.slow
-def test_memfault(dut_board, debug_hex_file):
+def test_memfault(dut_board, debug_hex_file, prepare_dut):
     # Save timestamp of latest coredump
     coredumps = get_latest_coredump_traces(UUID)
     logger.debug(f"Found coredumps: {coredumps}")
     timestamp_old_coredump = timestamp(coredumps[0]) if coredumps else  None
     logger.debug(f"Timestamp old coredump: {timestamp_old_coredump}")
 
-    flash_device(os.path.abspath(debug_hex_file))
-    dut_board.uart.xfactoryreset()
-    dut_board.uart.flush()
-    reset_device()
+    prepare_dut(dut_board, debug_hex_file)
 
     dut_board.uart.wait_for_str_with_retries("Connected to Cloud", max_retries=3, timeout=240, reset_func=reset_device)
 
@@ -115,19 +112,19 @@ def test_memfault(dut_board, debug_hex_file):
     # Wait for upload to be reported to memfault api
     start = time.time()
     while time.time() - start < MEMFAULT_TIMEOUT:
-            time.sleep(5)
-            coredumps = get_latest_coredump_traces(UUID)
-            logger.debug(f"Found coredumps: {coredumps}")
-            timestamp_new_coredump = timestamp(coredumps[0]) if coredumps else  None
-            logger.debug(f"Timestamp new coredump: {timestamp_new_coredump}")
-
-            if not coredumps:
-                continue
-            # Check that we have an upload with newer timestamp
-            if not timestamp_old_coredump:
-                break
-            if timestamp(coredumps[0]) > timestamp_old_coredump:
-                break
+        time.sleep(5)
+        coredumps = get_latest_coredump_traces(UUID)
+        logger.debug(f"Found coredumps: {coredumps}")
+        timestamp_new_coredump = timestamp(coredumps[0]) if coredumps else  None
+        logger.debug(f"Timestamp new coredump: {timestamp_new_coredump}")
+
+        if not coredumps:
+            continue
+        # Check that we have an upload with newer timestamp
+        if not timestamp_old_coredump:
+            break
+        if timestamp(coredumps[0]) > timestamp_old_coredump:
+            break
     else:
         raise RuntimeError("No new coredump observed")
 

tests/on_target/tests/test_functional/test_mqtt.py

@@ -4,7 +4,7 @@
 ##########################################################################################
 
 import os
-from utils.flash_tools import flash_device, reset_device
+from utils.flash_tools import reset_device
 import sys
 import pytest
 
@@ -14,25 +14,20 @@
 logger = get_logger()
 
 @pytest.mark.slow
-def test_mqtt_firmware(dut_board, hex_file_mqtt):
+def test_mqtt_firmware(dut_board, hex_file_mqtt, prepare_dut):
     """Test the firmware with cloud MQTT module."""
     # Only run this test for thingy91x devices
     devicetype = os.getenv("DUT_DEVICE_TYPE")
     if devicetype != "thingy91x":
         pytest.skip("This test is only for thingy91x devices")
 
-    # Flash the firmware
-    flash_device(os.path.abspath(hex_file_mqtt))
-    dut_board.uart.xfactoryreset()
+    # Flash and initialize the firmware
+    prepare_dut(dut_board, hex_file_mqtt)
 
     # Log patterns to check
     pattern_connect_to_broker = "network: lte_lc_evt_handler: PDN connection activated"
     pattern_cloud = "cloud: on_mqtt_connack: MQTT connection established, session present: 0"
 
-    # Cloud connection
-    dut_board.uart.flush()
-    reset_device()
-
     dut_board.uart.wait_for_str_with_retries(pattern_cloud, max_retries=3, timeout=240, reset_func=reset_device)
 
     # Wait for shadow request