fix(ci): twai test close board to fixed ci jobs pullotion

components/driver/test_apps/legacy_twai/pytest_twai.py

@@ -97,22 +97,24 @@ def fixture_create_socket_can() -> Bus:
     'target', ['esp32', 'esp32c3', 'esp32c6', 'esp32h2', 'esp32s2', 'esp32s3', 'esp32p4'], indirect=['target']
 )
 def test_legacy_twai_listen_only(dut: Dut, socket_can: Bus) -> None:
-    esp_reset_and_wait_ready(dut)
+    try:
+        esp_reset_and_wait_ready(dut)
 
-    # TEST_CASE("twai_listen_only", "[twai]")
-    dut.write('"twai_listen_only"')
+        # TEST_CASE("twai_listen_only", "[twai]")
+        dut.write('"twai_listen_only"')
 
-    # wait the DUT to start listening
-    time.sleep(0.1)
+        # wait the DUT to start listening
+        time.sleep(0.1)
 
-    message = Message(
-        arbitration_id=0x123,
-        is_extended_id=False,
-        data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88],
-    )
-    socket_can.send(message, timeout=0.2)
-    dut.expect_unity_test_output()
-    esp_enter_flash_mode(dut)
+        message = Message(
+            arbitration_id=0x123,
+            is_extended_id=False,
+            data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88],
+        )
+        socket_can.send(message, timeout=0.2)
+        dut.expect_unity_test_output()
+    finally:
+        esp_enter_flash_mode(dut)
 
 
 @pytest.mark.twai_adapter
@@ -127,30 +129,31 @@ def test_legacy_twai_listen_only(dut: Dut, socket_can: Bus) -> None:
     'target', ['esp32', 'esp32c3', 'esp32c6', 'esp32h2', 'esp32s2', 'esp32s3', 'esp32p4'], indirect=['target']
 )
 def test_legacy_twai_remote_request(dut: Dut, socket_can: Bus) -> None:
-    esp_reset_and_wait_ready(dut)
+    try:
+        esp_reset_and_wait_ready(dut)
 
-    # TEST_CASE("twai_remote_request", "[twai]")
-    dut.write('"twai_remote_request"')
+        # TEST_CASE("twai_remote_request", "[twai]")
+        dut.write('"twai_remote_request"')
 
-    deadline = time.time() + 2.0
-    req = None
-    while time.time() < deadline:
-        req = socket_can.recv(timeout=0.2)
-        if req is not None and req.is_remote_frame:
-            break
+        deadline = time.time() + 2.0
+        req = None
+        while time.time() < deadline:
+            req = socket_can.recv(timeout=0.2)
+            if req is not None and req.is_remote_frame:
+                break
 
-    if req is None:
-        raise Exception('Remote frame not received')
+        if req is None:
+            raise Exception('Remote frame not received')
+        logging.info(f'Received message: {req}')
 
-    logging.info(f'Received message: {req}')
+        reply = Message(
+            arbitration_id=req.arbitration_id,
+            is_extended_id=req.is_extended_id,
+            data=[0x10, 0x20, 0x30, 0x40, 0x50, 0x60, 0x70, 0x80],
+        )
+        socket_can.send(reply, timeout=0.2)
+        print('send', reply)
 
-    reply = Message(
-        arbitration_id=req.arbitration_id,
-        is_extended_id=req.is_extended_id,
-        data=[0x10, 0x20, 0x30, 0x40, 0x50, 0x60, 0x70, 0x80],
-    )
-    socket_can.send(reply, timeout=0.2)
-    print('send', reply)
-
-    dut.expect_unity_test_output()
-    esp_enter_flash_mode(dut)
+        dut.expect_unity_test_output()
+    finally:
+        esp_enter_flash_mode(dut)

components/esp_driver_twai/test_apps/test_twai/pytest_driver_twai.py

@@ -89,21 +89,22 @@ def fixture_create_socket_can() -> Bus:
 @pytest.mark.parametrize('config', ['release'], indirect=True)
 @idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target'])
 def test_driver_twai_listen_only(dut: Dut, socket_can: Bus) -> None:
-    esp_reset_and_wait_ready(dut)
+    try:
+        esp_reset_and_wait_ready(dut)
 
-    dut.write('"twai_listen_only"')
+        dut.write('"twai_listen_only"')
+        # wait the DUT to finish initialize
+        time.sleep(0.1)
 
-    # wait the DUT to finish initialize
-    time.sleep(0.1)
-
-    message = Message(
-        arbitration_id=0x6688,
-        is_extended_id=True,
-        data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88],
-    )
-    print('USB Socket CAN Send:', message, 'Return:', socket_can.send(message))
-    dut.expect_unity_test_output(timeout=10)
-    esp_enter_flash_mode(dut)
+        message = Message(
+            arbitration_id=0x6688,
+            is_extended_id=True,
+            data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88],
+        )
+        print('USB Socket CAN Send:', message, 'Return:', socket_can.send(message))
+        dut.expect_unity_test_output(timeout=10)
+    finally:
+        esp_enter_flash_mode(dut)
 
 
 @pytest.mark.twai_adapter
@@ -111,30 +112,30 @@ def test_driver_twai_listen_only(dut: Dut, socket_can: Bus) -> None:
 @pytest.mark.parametrize('config', ['release'], indirect=True)
 @idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target'])
 def test_driver_twai_remote_request(dut: Dut, socket_can: Bus) -> None:
-    esp_reset_and_wait_ready(dut)
+    try:
+        esp_reset_and_wait_ready(dut)
 
-    dut.write('"twai_remote_request"')
+        dut.write('"twai_remote_request"')
+        print('Waiting remote frame ...')
+        deadline = time.time() + 5.0
+        req = None
+        while time.time() < deadline:
+            req = socket_can.recv(timeout=0.2)
+            if req is not None and req.is_remote_frame:
+                break
 
-    print('Waiting remote frame ...')
-    deadline = time.time() + 5.0
-    req = None
-    while time.time() < deadline:
-        req = socket_can.recv(timeout=0.2)
-        if req is not None and req.is_remote_frame:
-            break
+        if req is None:
+            raise Exception('Remote frame not received')
+        print(f'USB Socket CAN Received: {req}')
 
-    if req is None:
-        raise Exception('Remote frame not received')
+        reply = Message(
+            arbitration_id=req.arbitration_id,
+            is_extended_id=req.is_extended_id,
+            data=[0x80, 0x70, 0x60, 0x50, 0x40, 0x30, 0x20, 0x10],
+        )
+        socket_can.send(reply, timeout=0.2)
+        print('USB Socket CAN Replied:', reply)
 
-    print(f'USB Socket CAN Received: {req}')
-
-    reply = Message(
-        arbitration_id=req.arbitration_id,
-        is_extended_id=req.is_extended_id,
-        data=[0x80, 0x70, 0x60, 0x50, 0x40, 0x30, 0x20, 0x10],
-    )
-    socket_can.send(reply, timeout=0.2)
-    print('USB Socket CAN Replied:', reply)
-
-    dut.expect_unity_test_output(timeout=10)
-    esp_enter_flash_mode(dut)
+        dut.expect_unity_test_output(timeout=10)
+    finally:
+        esp_enter_flash_mode(dut)

examples/peripherals/twai/twai_network/pytest_twai_network.py

@@ -3,10 +3,12 @@
 import os
 import os.path
 import subprocess
+import time
 
 import pytest
 from can import Bus
 from can import Message
+from pytest_embedded import Dut
 from pytest_embedded_idf import IdfDut
 
 can_env = os.getenv('CAN_PORT', 'can0')
@@ -25,6 +27,16 @@ def fixture_create_socket_can() -> Bus:
     subprocess.run(stop_command, shell=True, capture_output=True, text=True)
 
 
+def esp_enter_flash_mode(dut: Dut) -> None:
+    ser = dut.serial.proc
+    ser.setRTS(True)  # EN Low
+    time.sleep(0.5)
+    ser.setDTR(True)  # GPIO0 Low
+    ser.setRTS(False)  # EN High
+    dut.expect('waiting for download', timeout=2)
+    ser.setDTR(False)  # Back RTS/DTR to 1/1 to avoid affect to esptool
+
+
 @pytest.mark.twai_adapter
 @pytest.mark.parametrize('count', [2], indirect=True)
 @pytest.mark.timeout(120)
@@ -56,45 +68,50 @@ def test_twai_network_multi(dut: tuple[IdfDut, IdfDut], socket_can: Bus) -> None
     - dut[1]: twai_sender
     """
 
-    # Print chip information for debugging
-    print(f'===> Pytest testing with chips: {dut[0].app.target} (listener), {dut[1].app.target} (sender)')
+    try:
+        # Print chip information for debugging
+        print(f'===> Pytest testing with chips: {dut[0].app.target} (listener), {dut[1].app.target} (sender)')
 
-    # Initialize listener node first
-    dut[0].expect('===================TWAI Listen Only Example Starting...===================')
-    dut[0].expect('TWAI start listening...')
+        # Initialize listener node first
+        dut[0].expect('===================TWAI Listen Only Example Starting...===================')
+        dut[0].expect('TWAI start listening...')
 
-    # Initialize sender node and start communication
-    dut[1].expect('===================TWAI Sender Example Starting...===================')
-    dut[1].expect('TWAI Sender started successfully')
+        # Initialize sender node and start communication
+        dut[1].expect('===================TWAI Sender Example Starting...===================')
+        dut[1].expect('TWAI Sender started successfully')
 
-    # Verify communication is working
-    # Wait for sender to send messages
-    dut[1].expect('Sending heartbeat message:', timeout=10)
+        # Verify communication is working
+        # Wait for sender to send messages
+        dut[1].expect('Sending heartbeat message:', timeout=10)
 
-    # Check that listener is receiving data
-    dut[0].expect('RX:', timeout=15)  # Listener should see filtered messages
+        # Check that listener is receiving data
+        dut[0].expect('RX:', timeout=15)  # Listener should see filtered messages
 
-    # Check if socket receive any messages
-    socket_rcv_cnt = 0
-    for i in range(100):
-        msg = socket_can.recv(timeout=1)
-        if msg is not None:
-            socket_rcv_cnt += 1
-    print(f'Socket receive {socket_rcv_cnt} messages')
-    assert socket_rcv_cnt > 50, 'Socket NO messages'
+        # Check if socket receive any messages
+        socket_rcv_cnt = 0
+        for i in range(100):
+            msg = socket_can.recv(timeout=1)
+            if msg is not None:
+                socket_rcv_cnt += 1
+        print(f'Socket receive {socket_rcv_cnt} messages')
+        assert socket_rcv_cnt > 50, 'Socket NO messages'
 
-    # Wait a bit more to ensure stable communication
-    dut[1].expect('Sending packet of', timeout=10)
-    dut[0].expect('RX:', timeout=10)
+        # Wait a bit more to ensure stable communication
+        dut[1].expect('Sending packet of', timeout=10)
+        dut[0].expect('RX:', timeout=10)
 
-    # Check if esp32 receive messages from usb can
-    message = Message(
-        arbitration_id=0x10A,
-        is_extended_id=False,
-        data=b'Hi ESP32',
-    )
-    print('USB CAN Send:', message)
-    socket_can.send(message, timeout=0.2)
-    dut[0].expect_exact('10a [8] 48 69 20 45 53 50 33 32', timeout=10)  # ASCII: Hi ESP32
+        # Check if esp32 receive messages from usb can
+        message = Message(
+            arbitration_id=0x10A,
+            is_extended_id=False,
+            data=b'Hi ESP32',
+        )
+        print('USB CAN Send:', message)
+        socket_can.send(message, timeout=0.2)
+        dut[0].expect_exact('10a [8] 48 69 20 45 53 50 33 32', timeout=10)  # ASCII: Hi ESP32
 
-    print('===> TWAI network communication test completed successfully')
+        print('===> TWAI network communication test completed successfully')
+
+    finally:
+        esp_enter_flash_mode(dut[0])
+        esp_enter_flash_mode(dut[1])

examples/peripherals/twai/twai_utils/pytest_twai_utils.py

@@ -613,34 +613,34 @@ def test_twai_utils_external_communication(twai: TwaiTestHelper, usb_can: CanBus
             bitrate=DEFAULT_BITRATE,
             start_dump=False,
         ):
-            # --- ESP -> PC Test ---
-            for frame_str, frame_id, expected_data, is_extended in test_frames:
-                assert twai.send(frame_str), f'ESP->PC send failed: {frame_str}'
-                deadline = time.time() + 2.0
-                got = None
-                while time.time() < deadline:
-                    try:
-                        msg = can_bus.recv(timeout=0.1)
-                        if msg and msg.arbitration_id == frame_id:
-                            got = msg
-                            break
-                    except Exception:
-                        continue
-                assert got is not None, f'ESP->PC receive timeout for ID=0x{frame_id:X}'
-                assert bool(got.is_extended_id) == is_extended, (
-                    f'ESP->PC extended flag mismatch for 0x{frame_id:X}: '
-                    f'expected {is_extended}, got {got.is_extended_id}'
-                )
-                assert bytes(got.data) == expected_data, (
-                    f'ESP->PC data mismatch for 0x{frame_id:X}: '
-                    f'expected {expected_data.hex()}, got {bytes(got.data).hex()}'
-                )
-
-            # --- PC -> ESP ---
-            assert twai.dump_start(), 'Failed to start twai_dump'
-            assert twai.info(), 'Failed to get twai_info'
-
             try:
+                # --- ESP -> PC Test ---
+                for frame_str, frame_id, expected_data, is_extended in test_frames:
+                    assert twai.send(frame_str), f'ESP->PC send failed: {frame_str}'
+                    deadline = time.time() + 2.0
+                    got = None
+                    while time.time() < deadline:
+                        try:
+                            msg = can_bus.recv(timeout=0.1)
+                            if msg and msg.arbitration_id == frame_id:
+                                got = msg
+                                break
+                        except Exception:
+                            continue
+                    assert got is not None, f'ESP->PC receive timeout for ID=0x{frame_id:X}'
+                    assert bool(got.is_extended_id) == is_extended, (
+                        f'ESP->PC extended flag mismatch for 0x{frame_id:X}: '
+                        f'expected {is_extended}, got {got.is_extended_id}'
+                    )
+                    assert bytes(got.data) == expected_data, (
+                        f'ESP->PC data mismatch for 0x{frame_id:X}: '
+                        f'expected {expected_data.hex()}, got {bytes(got.data).hex()}'
+                    )
+
+                # --- PC -> ESP ---
+                assert twai.dump_start(), 'Failed to start twai_dump'
+                assert twai.info(), 'Failed to get twai_info'
+
                 for frame_str, frame_id, expected_data, is_extended in test_frames:
                     msg = can.Message(arbitration_id=frame_id, data=expected_data, is_extended_id=is_extended)
                     print(f'\nPC->ESP sending frame: {msg}, Return: {can_bus.send(msg)}')

pytest.ini

@@ -123,7 +123,6 @@ env_markers =
     sdio_multidev_32_c61: Test sdio multi board, esp32+esp32c61
     sdio_multidev_p4_c5: Test sdio multi board, esp32p4+esp32c5
     usj_device: Test usb_serial_jtag and usb_serial_jtag is used as serial only (not console)
-    twai_std: twai runner with all twai supported targets connect to usb-can adapter
     twai_adapter: runner with multiple twai_std
     lp_i2s: lp_i2s runner tested with hp_i2s
     ram_app: ram_app runners