From 4ab13e58948745d2d23a2d2bd9140b4f47620fed Mon Sep 17 00:00:00 2001 From: wanckl Date: Mon, 22 Jun 2026 20:18:58 +0800 Subject: [PATCH] fix(ci): twai test close board to fixed ci jobs pullotion --- .../test_apps/legacy_twai/pytest_twai.py | 73 ++++++++-------- .../test_apps/test_twai/pytest_driver_twai.py | 71 ++++++++-------- .../twai/twai_network/pytest_twai_network.py | 85 +++++++++++-------- .../twai/twai_utils/pytest_twai_utils.py | 54 ++++++------ pytest.ini | 1 - 5 files changed, 153 insertions(+), 131 deletions(-) diff --git a/components/driver/test_apps/legacy_twai/pytest_twai.py b/components/driver/test_apps/legacy_twai/pytest_twai.py index 04eef1b5b38..b8103449311 100644 --- a/components/driver/test_apps/legacy_twai/pytest_twai.py +++ b/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) diff --git a/components/esp_driver_twai/test_apps/test_twai/pytest_driver_twai.py b/components/esp_driver_twai/test_apps/test_twai/pytest_driver_twai.py index 05ebee81ea1..966a277cd32 100644 --- a/components/esp_driver_twai/test_apps/test_twai/pytest_driver_twai.py +++ b/components/esp_driver_twai/test_apps/test_twai/pytest_driver_twai.py @@ -84,20 +84,22 @@ def fixture_create_socket_can() -> Bus: @idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target']) @pytest.mark.temp_skip_ci(targets=['esp32h4'], reason='no runner') # TODO: IDFCI-11110 def test_driver_twai_listen_only(dut: Dut, socket_can: Bus) -> None: - dut.expect_exact('Press ENTER to see the list of tests') - dut.write('"twai_listen_only"') + try: + dut.expect_exact('Press ENTER to see the list of tests') + 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 @@ -106,29 +108,30 @@ def test_driver_twai_listen_only(dut: Dut, socket_can: Bus) -> None: @idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target']) @pytest.mark.temp_skip_ci(targets=['esp32h4'], reason='no runner') # TODO: IDFCI-11110 def test_driver_twai_remote_request(dut: Dut, socket_can: Bus) -> None: - dut.expect_exact('Press ENTER to see the list of tests') - dut.write('"twai_remote_request"') + try: + dut.expect_exact('Press ENTER to see the list of tests') + 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') + if req is None: + raise Exception('Remote frame not received') + print(f'USB Socket CAN Received: {req}') - 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) - 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) diff --git a/examples/peripherals/twai/twai_network/pytest_twai_network.py b/examples/peripherals/twai/twai_network/pytest_twai_network.py index c24c4318052..eee23823af0 100644 --- a/examples/peripherals/twai/twai_network/pytest_twai_network.py +++ b/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]) diff --git a/examples/peripherals/twai/twai_utils/pytest_twai_utils.py b/examples/peripherals/twai/twai_utils/pytest_twai_utils.py index 3f4dc6f3cd1..6eece3d69cf 100644 --- a/examples/peripherals/twai/twai_utils/pytest_twai_utils.py +++ b/examples/peripherals/twai/twai_utils/pytest_twai_utils.py @@ -617,34 +617,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)}') diff --git a/pytest.ini b/pytest.ini index fecb28943c3..f2f3d27720e 100644 --- a/pytest.ini +++ b/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