Mirrors/esp-idf
1
0
Fork 0
mirror of https://github.com/espressif/esp-idf.git synced 2026-10-02 11:10:54 +03:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
wanckl
2026-06-22 20:18:58 +08:00
parent 32a17773f4
commit 4ab13e5894
5 changed files with 153 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/driver/test_apps/legacy_twai/pytest_twai.py
+38 -35
View File
@@ -97,22 +97,24 @@ def fixture_create_socket_can() -> Bus:
'target', ['esp32', 'esp32c3', 'esp32c6', 'esp32h2', 'esp32s2', 'esp32s3', 'esp32p4'], indirect=['target'] 'target', ['esp32', 'esp32c3', 'esp32c6', 'esp32h2', 'esp32s2', 'esp32s3', 'esp32p4'], indirect=['target']
) )
def test_legacy_twai_listen_only(dut: Dut, socket_can: Bus) -> None: 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]") # TEST_CASE("twai_listen_only", "[twai]")
dut.write('"twai_listen_only"') dut.write('"twai_listen_only"')
# wait the DUT to start listening # wait the DUT to start listening
time.sleep(0.1) time.sleep(0.1)
message = Message( message = Message(
arbitration_id=0x123, arbitration_id=0x123,
is_extended_id=False, is_extended_id=False,
data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88], data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88],
) )
socket_can.send(message, timeout=0.2) socket_can.send(message, timeout=0.2)
dut.expect_unity_test_output() dut.expect_unity_test_output()
esp_enter_flash_mode(dut) finally:
esp_enter_flash_mode(dut)
@pytest.mark.twai_adapter @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'] 'target', ['esp32', 'esp32c3', 'esp32c6', 'esp32h2', 'esp32s2', 'esp32s3', 'esp32p4'], indirect=['target']
) )
def test_legacy_twai_remote_request(dut: Dut, socket_can: Bus) -> None: 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]") # TEST_CASE("twai_remote_request", "[twai]")
dut.write('"twai_remote_request"') dut.write('"twai_remote_request"')
deadline = time.time() + 2.0 deadline = time.time() + 2.0
req = None req = None
while time.time() < deadline: while time.time() < deadline:
req = socket_can.recv(timeout=0.2) req = socket_can.recv(timeout=0.2)
if req is not None and req.is_remote_frame: if req is not None and req.is_remote_frame:
break break
if req is None: if req is None:
raise Exception('Remote frame not received') 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( dut.expect_unity_test_output()
arbitration_id=req.arbitration_id, finally:
is_extended_id=req.is_extended_id, esp_enter_flash_mode(dut)
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)
components/esp_driver_twai/test_apps/test_twai/pytest_driver_twai.py
+37 -34
View File
@@ -84,20 +84,22 @@ def fixture_create_socket_can() -> Bus:
@idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target']) @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 @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: def test_driver_twai_listen_only(dut: Dut, socket_can: Bus) -> None:
dut.expect_exact('Press ENTER to see the list of tests') try:
dut.write('"twai_listen_only"') dut.expect_exact('Press ENTER to see the list of tests')
dut.write('"twai_listen_only"')
# wait the DUT to finish initialize # wait the DUT to finish initialize
time.sleep(0.1) time.sleep(0.1)
message = Message( message = Message(
arbitration_id=0x6688, arbitration_id=0x6688,
is_extended_id=True, is_extended_id=True,
data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88], data=[0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88],
) )
print('USB Socket CAN Send:', message, 'Return:', socket_can.send(message)) print('USB Socket CAN Send:', message, 'Return:', socket_can.send(message))
dut.expect_unity_test_output(timeout=10) dut.expect_unity_test_output(timeout=10)
esp_enter_flash_mode(dut) finally:
esp_enter_flash_mode(dut)
@pytest.mark.twai_adapter @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']) @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 @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: def test_driver_twai_remote_request(dut: Dut, socket_can: Bus) -> None:
dut.expect_exact('Press ENTER to see the list of tests') try:
dut.write('"twai_remote_request"') dut.expect_exact('Press ENTER to see the list of tests')
dut.write('"twai_remote_request"')
print('Waiting remote frame ...') print('Waiting remote frame ...')
deadline = time.time() + 5.0 deadline = time.time() + 5.0
req = None req = None
while time.time() < deadline: while time.time() < deadline:
req = socket_can.recv(timeout=0.2) req = socket_can.recv(timeout=0.2)
if req is not None and req.is_remote_frame: if req is not None and req.is_remote_frame:
break break
if req is None: if req is None:
raise Exception('Remote frame not received') 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( dut.expect_unity_test_output(timeout=10)
arbitration_id=req.arbitration_id, finally:
is_extended_id=req.is_extended_id, esp_enter_flash_mode(dut)
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)
examples/peripherals/twai/twai_network/pytest_twai_network.py
+51 -34
View File
@@ -3,10 +3,12 @@
import os import os
import os.path import os.path
import subprocess import subprocess
import time
import pytest import pytest
from can import Bus from can import Bus
from can import Message from can import Message
from pytest_embedded import Dut
from pytest_embedded_idf import IdfDut from pytest_embedded_idf import IdfDut
can_env = os.getenv('CAN_PORT', 'can0') 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) 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.twai_adapter
@pytest.mark.parametrize('count', [2], indirect=True) @pytest.mark.parametrize('count', [2], indirect=True)
@pytest.mark.timeout(120) @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 - dut[1]: twai_sender
""" """
# Print chip information for debugging try:
print(f'===> Pytest testing with chips: {dut[0].app.target} (listener), {dut[1].app.target} (sender)') # 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 # Initialize listener node first
dut[0].expect('===================TWAI Listen Only Example Starting...===================') dut[0].expect('===================TWAI Listen Only Example Starting...===================')
dut[0].expect('TWAI start listening...') dut[0].expect('TWAI start listening...')
# Initialize sender node and start communication # Initialize sender node and start communication
dut[1].expect('===================TWAI Sender Example Starting...===================') dut[1].expect('===================TWAI Sender Example Starting...===================')
dut[1].expect('TWAI Sender started successfully') dut[1].expect('TWAI Sender started successfully')
# Verify communication is working # Verify communication is working
# Wait for sender to send messages # Wait for sender to send messages
dut[1].expect('Sending heartbeat message:', timeout=10) dut[1].expect('Sending heartbeat message:', timeout=10)
# Check that listener is receiving data # Check that listener is receiving data
dut[0].expect('RX:', timeout=15) # Listener should see filtered messages dut[0].expect('RX:', timeout=15) # Listener should see filtered messages
# Check if socket receive any messages # Check if socket receive any messages
socket_rcv_cnt = 0 socket_rcv_cnt = 0
for i in range(100): for i in range(100):
msg = socket_can.recv(timeout=1) msg = socket_can.recv(timeout=1)
if msg is not None: if msg is not None:
socket_rcv_cnt += 1 socket_rcv_cnt += 1
print(f'Socket receive {socket_rcv_cnt} messages') print(f'Socket receive {socket_rcv_cnt} messages')
assert socket_rcv_cnt > 50, 'Socket NO messages' assert socket_rcv_cnt > 50, 'Socket NO messages'
# Wait a bit more to ensure stable communication # Wait a bit more to ensure stable communication
dut[1].expect('Sending packet of', timeout=10) dut[1].expect('Sending packet of', timeout=10)
dut[0].expect('RX:', timeout=10) dut[0].expect('RX:', timeout=10)
# Check if esp32 receive messages from usb can # Check if esp32 receive messages from usb can
message = Message( message = Message(
arbitration_id=0x10A, arbitration_id=0x10A,
is_extended_id=False, is_extended_id=False,
data=b'Hi ESP32', data=b'Hi ESP32',
) )
print('USB CAN Send:', message) print('USB CAN Send:', message)
socket_can.send(message, timeout=0.2) 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 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
+27 -27
View File
@@ -617,34 +617,34 @@ def test_twai_utils_external_communication(twai: TwaiTestHelper, usb_can: CanBus
bitrate=DEFAULT_BITRATE, bitrate=DEFAULT_BITRATE,
start_dump=False, 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: 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: 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) 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)}') print(f'\nPC->ESP sending frame: {msg}, Return: {can_bus.send(msg)}')
pytest.ini
-1
View File
@@ -123,7 +123,6 @@ env_markers =
sdio_multidev_32_c61: Test sdio multi board, esp32+esp32c61 sdio_multidev_32_c61: Test sdio multi board, esp32+esp32c61
sdio_multidev_p4_c5: Test sdio multi board, esp32p4+esp32c5 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) 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 twai_adapter: runner with multiple twai_std
lp_i2s: lp_i2s runner tested with hp_i2s lp_i2s: lp_i2s runner tested with hp_i2s
ram_app: ram_app runners ram_app: ram_app runners