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

Merge branch 'ci/twai-runner-update-v6.0' into 'release/v6.0'

Browse Source 
ci: twai runner setup v6.0

See merge request espressif/esp-idf!49803
This commit is contained in:
morris
2026-06-29 10:59:17 +08:00
parent cabc6100a7 35fb412e71
commit d91bff2f0f
5 changed files with 200 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
components/driver/test_apps/legacy_twai/pytest_twai.py
+54 -45
View File
@@ -1,6 +1,7 @@
# SPDX-FileCopyrightText: 2021-2025 Espressif Systems (Shanghai) CO LTD
# SPDX-FileCopyrightText: 2021-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: CC0-1.0
import logging
import os
import subprocess
import time
@@ -14,6 +15,9 @@ from pytest_embedded_idf.utils import idf_parametrize
# Loop Back Tests
# ---------------------------------------------------------------------------
can_env = os.getenv('CAN_PORT', 'can0')
print(f'CAN_PORT={can_env}')
@pytest.mark.generic
@pytest.mark.parametrize(
@@ -54,28 +58,26 @@ def esp_reset_and_wait_ready(dut: Dut) -> None:
@pytest.fixture(name='socket_can')
def fixture_create_socket_can() -> Bus:
# Set up the socket CAN with the bitrate
start_command = 'sudo -n ip link set can0 up type can bitrate 250000 restart-ms 100'
stop_command = 'sudo -n ip link set can0 down'
status_command = 'sudo -n ip -details link show can0'
start_command = f'sudo -n ip link set {can_env} up type can bitrate 250000'
stop_command = f'sudo -n ip link set {can_env} down'
status_command = f'sudo -n ip -details link show {can_env}'
try:
result = subprocess.run(status_command, shell=True, capture_output=True, text=True)
result = subprocess.run(status_command, shell=True, capture_output=True, text=True, timeout=2)
if result.returncode != 0:
raise Exception('CAN interface "can0" not found')
raise Exception(f'CAN interface "{can_env}" not found')
if 'UP' in result.stdout: # Close the bus anyway if it is already up
subprocess.run(stop_command, shell=True, capture_output=True, text=True)
subprocess.run(start_command, shell=True, capture_output=True, text=True)
subprocess.run(stop_command, shell=True, capture_output=True, text=True, timeout=2)
subprocess.run(start_command, shell=True, capture_output=True, text=True, timeout=2)
time.sleep(0.5)
bus = Bus(interface='socketcan', channel='can0', bitrate=250000)
bus = Bus(interface='socketcan', channel=f'{can_env}', bitrate=250000)
yield bus # test invoked here
bus.shutdown()
except Exception as e:
pytest.skip(f'Open usb-can bus Error: {str(e)}')
finally:
subprocess.run(stop_command, shell=True, capture_output=True, text=True)
subprocess.run(stop_command, shell=True, capture_output=True, text=True, timeout=2)
# ---------------------------------------------------------------------------
@@ -83,7 +85,7 @@ def fixture_create_socket_can() -> Bus:
# ---------------------------------------------------------------------------
@pytest.mark.twai_std
@pytest.mark.twai_adapter
@pytest.mark.parametrize(
'config',
[
@@ -95,25 +97,27 @@ 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_std
@pytest.mark.twai_adapter
@pytest.mark.parametrize(
'config',
[
@@ -125,26 +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"')
while True:
req = socket_can.recv(timeout=0.2)
# wait for the remote request frame
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
logging.info(f'Received message: {req}')
if req is None:
raise Exception('Remote frame not received')
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
+50 -42
View File
@@ -1,6 +1,6 @@
# SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import os
import subprocess
import time
@@ -11,6 +11,9 @@ from pytest_embedded import Dut
from pytest_embedded_idf.utils import idf_parametrize
from pytest_embedded_idf.utils import soc_filtered_targets
can_env = os.getenv('CAN_PORT', 'can0')
print(f'CAN_PORT={can_env}')
# ---------------------------------------------------------------------------
# Loop Back Tests
@@ -56,26 +59,24 @@ def esp_reset_and_wait_ready(dut: Dut) -> None:
@pytest.fixture(name='socket_can')
def fixture_create_socket_can() -> Bus:
# Set up the socket CAN with the bitrate
start_command = 'sudo -n ip link set can0 up type can bitrate 250000 restart-ms 100'
stop_command = 'sudo -n ip link set can0 down'
status_command = 'sudo -n ip -details link show can0'
start_command = f'sudo -n ip link set {can_env} up type can bitrate 250000'
stop_command = f'sudo -n ip link set {can_env} down'
status_command = f'sudo -n ip -details link show {can_env}'
try:
result = subprocess.run(status_command, shell=True, capture_output=True, text=True)
if result.returncode != 0:
raise Exception('CAN interface "can0" not found')
raise Exception(f'CAN interface "{can_env}" not found')
if 'UP' in result.stdout: # Close the bus anyway if it is already up
subprocess.run(stop_command, shell=True, capture_output=True, text=True)
subprocess.run(start_command, shell=True, capture_output=True, text=True)
time.sleep(0.5)
bus = Bus(interface='socketcan', channel='can0', bitrate=250000)
bus = Bus(interface='socketcan', channel=f'{can_env}', bitrate=250000)
yield bus # test invoked here
bus.shutdown()
except Exception as e:
pytest.skip(f'Open usb-can bus Error: {str(e)}')
finally:
subprocess.run(stop_command, shell=True, capture_output=True, text=True)
@@ -83,51 +84,58 @@ def fixture_create_socket_can() -> Bus:
# ---------------------------------------------------------------------------
# Interactive Tests
# ---------------------------------------------------------------------------
@pytest.mark.twai_std
@pytest.mark.temp_skip_ci(targets=['esp32h4'], reason='no runner')
@pytest.mark.twai_adapter
@pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner')
@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_std
@pytest.mark.temp_skip_ci(targets=['esp32h4'], reason='no runner')
@pytest.mark.twai_adapter
@pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner')
@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 ...')
while True:
req = socket_can.recv(timeout=0.2)
if req is not None and req.is_remote_frame:
break
print(f'USB Socket CAN Received: {req}')
if req is None:
raise Exception('Remote frame not received')
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)
examples/peripherals/twai/twai_network/pytest_twai_network.py
+60 -42
View File
@@ -1,30 +1,43 @@
# SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: CC0-1.0
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')
print(f'CAN_PORT={can_env}')
# Socket CAN fixture
@pytest.fixture(name='socket_can')
def fixture_create_socket_can() -> Bus:
start_command = 'sudo ip link set can0 up type can bitrate 1000000'
stop_command = 'sudo ip link set can0 down'
try:
subprocess.run(start_command, shell=True, capture_output=True, text=True)
except Exception as e:
print(f'Open bus Error: {e}')
bus = Bus(interface='socketcan', channel='can0', bitrate=1000000)
start_command = f'sudo ip link set {can_env} up type can bitrate 1000000'
stop_command = f'sudo ip link set {can_env} down'
subprocess.run(start_command, shell=True, capture_output=True, text=True)
bus = Bus(interface='socketcan', channel=f'{can_env}', bitrate=1000000)
yield bus # test invoked here
bus.shutdown()
subprocess.run(stop_command, shell=True, capture_output=True, text=True)
@pytest.mark.twai_std
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)
@pytest.mark.parametrize(
@@ -55,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
+35 -33
View File
@@ -1,6 +1,6 @@
# SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import os
import subprocess
import time
from collections.abc import Generator
@@ -18,6 +18,9 @@ from pytest_embedded_idf.utils import soc_filtered_targets
# Constants / Helpers
# ---------------------------------------------------------------------------
can_env = os.getenv('CAN_PORT', 'can0')
print(f'CAN_PORT={can_env}')
PROMPTS = ['twai>']
# Hardware configuration
@@ -253,7 +256,7 @@ class TwaiTestHelper:
class CanBusManager:
"""CAN bus manager for external hardware tests"""
def __init__(self, interface: str = 'can0'):
def __init__(self, interface: str = can_env):
self.interface = interface
self.bus: can.Bus | None = None
@@ -280,8 +283,6 @@ class CanBusManager:
self.bus = can.Bus(interface='socketcan', channel=self.interface)
yield self.bus
except Exception as e:
pytest.skip(f'CAN interface not available: {str(e)}')
finally:
if self.bus:
self.bus.shutdown()
@@ -593,8 +594,9 @@ def test_twai_utils_range_filters(twai: TwaiTestHelper) -> None:
# ---------------------------------------------------------------------------
@pytest.mark.twai_std
@pytest.mark.temp_skip_ci(targets=['esp32h4'], reason='no runner')
@pytest.mark.twai_adapter
@pytest.mark.temp_skip_ci(targets=['esp32h4'], reason='no runner') # TODO: IDFCI-11110
@pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner')
@idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target'])
def test_twai_utils_external_communication(twai: TwaiTestHelper, usb_can: CanBusManager) -> None:
test_frames = [
@@ -611,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
+1 -1
View File
@@ -122,7 +122,7 @@ 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
recovery_bootloader: Runner with recovery bootloader offset set in