Merge branch 'ci/twai-runner-migration' into 'master'

ci: twai runner setup

See merge request espressif/esp-idf!49651
This commit is contained in:
Fu Hanxi
2026-06-17 10:11:57 +02:00
5 changed files with 56 additions and 40 deletions
@@ -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 # SPDX-License-Identifier: CC0-1.0
import logging import logging
import os
import subprocess import subprocess
import time import time
@@ -14,6 +15,9 @@ from pytest_embedded_idf.utils import idf_parametrize
# Loop Back Tests # Loop Back Tests
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
can_env = os.getenv('CAN_PORT', 'can0')
print(f'CAN_PORT={can_env}')
@pytest.mark.generic @pytest.mark.generic
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -54,28 +58,26 @@ def esp_reset_and_wait_ready(dut: Dut) -> None:
@pytest.fixture(name='socket_can') @pytest.fixture(name='socket_can')
def fixture_create_socket_can() -> Bus: def fixture_create_socket_can() -> Bus:
# Set up the socket CAN with the bitrate # Set up the socket CAN with the bitrate
start_command = 'sudo -n ip link set can0 up type can bitrate 250000' start_command = f'sudo -n ip link set {can_env} up type can bitrate 250000'
stop_command = 'sudo -n ip link set can0 down' stop_command = f'sudo -n ip link set {can_env} down'
status_command = 'sudo -n ip -details link show can0' status_command = f'sudo -n ip -details link show {can_env}'
try: 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: 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 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(stop_command, shell=True, capture_output=True, text=True, timeout=2)
subprocess.run(start_command, shell=True, capture_output=True, text=True) subprocess.run(start_command, shell=True, capture_output=True, text=True, timeout=2)
time.sleep(0.5) 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 yield bus # test invoked here
bus.shutdown() bus.shutdown()
except Exception as e:
pytest.skip(f'Open usb-can bus Error: {str(e)}')
finally: 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( @pytest.mark.parametrize(
'config', 'config',
[ [
@@ -113,7 +115,7 @@ def test_legacy_twai_listen_only(dut: Dut, socket_can: Bus) -> None:
esp_enter_flash_mode(dut) esp_enter_flash_mode(dut)
@pytest.mark.twai_std @pytest.mark.twai_adapter
@pytest.mark.parametrize( @pytest.mark.parametrize(
'config', 'config',
[ [
@@ -130,12 +132,16 @@ def test_legacy_twai_remote_request(dut: Dut, socket_can: Bus) -> None:
# TEST_CASE("twai_remote_request", "[twai]") # TEST_CASE("twai_remote_request", "[twai]")
dut.write('"twai_remote_request"') dut.write('"twai_remote_request"')
while True: deadline = time.time() + 2.0
req = None
while time.time() < deadline:
req = socket_can.recv(timeout=0.2) req = socket_can.recv(timeout=0.2)
# wait for the remote request frame
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:
raise Exception('Remote frame not received')
logging.info(f'Received message: {req}') logging.info(f'Received message: {req}')
reply = Message( reply = Message(
@@ -1,6 +1,6 @@
# SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD # SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0 # SPDX-License-Identifier: Apache-2.0
import os
import subprocess import subprocess
import time 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 idf_parametrize
from pytest_embedded_idf.utils import soc_filtered_targets 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 # Loop Back Tests
@@ -50,26 +53,24 @@ def esp_enter_flash_mode(dut: Dut) -> None:
@pytest.fixture(name='socket_can') @pytest.fixture(name='socket_can')
def fixture_create_socket_can() -> Bus: def fixture_create_socket_can() -> Bus:
# Set up the socket CAN with the bitrate # Set up the socket CAN with the bitrate
start_command = 'sudo -n ip link set can0 up type can bitrate 250000' start_command = f'sudo -n ip link set {can_env} up type can bitrate 250000'
stop_command = 'sudo -n ip link set can0 down' stop_command = f'sudo -n ip link set {can_env} down'
status_command = 'sudo -n ip -details link show can0' status_command = f'sudo -n ip -details link show {can_env}'
try: 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)
if result.returncode != 0: 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 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(stop_command, shell=True, capture_output=True, text=True)
subprocess.run(start_command, shell=True, capture_output=True, text=True) subprocess.run(start_command, shell=True, capture_output=True, text=True)
time.sleep(0.5) 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 yield bus # test invoked here
bus.shutdown() bus.shutdown()
except Exception as e:
pytest.skip(f'Open usb-can bus Error: {str(e)}')
finally: finally:
subprocess.run(stop_command, shell=True, capture_output=True, text=True) subprocess.run(stop_command, shell=True, capture_output=True, text=True)
@@ -77,7 +78,7 @@ def fixture_create_socket_can() -> Bus:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Interactive Tests # Interactive Tests
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@pytest.mark.twai_std @pytest.mark.twai_adapter
@pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner') @pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner')
@pytest.mark.parametrize('config', ['release'], indirect=True) @pytest.mark.parametrize('config', ['release'], indirect=True)
@idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target']) @idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target'])
@@ -99,7 +100,7 @@ def test_driver_twai_listen_only(dut: Dut, socket_can: Bus) -> None:
esp_enter_flash_mode(dut) esp_enter_flash_mode(dut)
@pytest.mark.twai_std @pytest.mark.twai_adapter
@pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner') @pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner')
@pytest.mark.parametrize('config', ['release'], indirect=True) @pytest.mark.parametrize('config', ['release'], indirect=True)
@idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target']) @idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target'])
@@ -109,10 +110,16 @@ def test_driver_twai_remote_request(dut: Dut, socket_can: Bus) -> None:
dut.write('"twai_remote_request"') dut.write('"twai_remote_request"')
print('Waiting remote frame ...') print('Waiting remote frame ...')
while True: deadline = time.time() + 5.0
req = None
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:
raise Exception('Remote frame not received')
print(f'USB Socket CAN Received: {req}') print(f'USB Socket CAN Received: {req}')
reply = Message( reply = Message(
@@ -1,5 +1,6 @@
# SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD # SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: CC0-1.0 # SPDX-License-Identifier: CC0-1.0
import os
import os.path import os.path
import subprocess import subprocess
@@ -8,23 +9,23 @@ from can import Bus
from can import Message from can import Message
from pytest_embedded_idf import IdfDut from pytest_embedded_idf import IdfDut
can_env = os.getenv('CAN_PORT', 'can0')
print(f'CAN_PORT={can_env}')
# Socket CAN fixture # Socket CAN fixture
@pytest.fixture(name='socket_can') @pytest.fixture(name='socket_can')
def fixture_create_socket_can() -> Bus: def fixture_create_socket_can() -> Bus:
start_command = 'sudo ip link set can0 up type can bitrate 1000000' start_command = f'sudo ip link set {can_env} up type can bitrate 1000000'
stop_command = 'sudo ip link set can0 down' stop_command = f'sudo ip link set {can_env} down'
try: subprocess.run(start_command, shell=True, capture_output=True, text=True)
subprocess.run(start_command, shell=True, capture_output=True, text=True) bus = Bus(interface='socketcan', channel=f'{can_env}', bitrate=1000000)
except Exception as e:
print(f'Open bus Error: {e}')
bus = Bus(interface='socketcan', channel='can0', bitrate=1000000)
yield bus # test invoked here yield bus # test invoked here
bus.shutdown() bus.shutdown()
subprocess.run(stop_command, shell=True, capture_output=True, text=True) subprocess.run(stop_command, shell=True, capture_output=True, text=True)
@pytest.mark.twai_std @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)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -1,6 +1,6 @@
# SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD # SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0 # SPDX-License-Identifier: Apache-2.0
import os
import subprocess import subprocess
import time import time
from collections.abc import Generator from collections.abc import Generator
@@ -18,6 +18,9 @@ from pytest_embedded_idf.utils import soc_filtered_targets
# Constants / Helpers # Constants / Helpers
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
can_env = os.getenv('CAN_PORT', 'can0')
print(f'CAN_PORT={can_env}')
PROMPTS = ['twai>'] PROMPTS = ['twai>']
# Hardware configuration # Hardware configuration
@@ -253,7 +256,7 @@ class TwaiTestHelper:
class CanBusManager: class CanBusManager:
"""CAN bus manager for external hardware tests""" """CAN bus manager for external hardware tests"""
def __init__(self, interface: str = 'can0'): def __init__(self, interface: str = can_env):
self.interface = interface self.interface = interface
self.bus: can.Bus | None = None self.bus: can.Bus | None = None
@@ -280,8 +283,6 @@ class CanBusManager:
self.bus = can.Bus(interface='socketcan', channel=self.interface) self.bus = can.Bus(interface='socketcan', channel=self.interface)
yield self.bus yield self.bus
except Exception as e:
pytest.skip(f'CAN interface not available: {str(e)}')
finally: finally:
if self.bus: if self.bus:
self.bus.shutdown() self.bus.shutdown()
@@ -597,7 +598,7 @@ def test_twai_utils_range_filters(twai: TwaiTestHelper) -> None:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@pytest.mark.twai_std @pytest.mark.twai_adapter
@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
@pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner') @pytest.mark.temp_skip_ci(targets=['esp32s31'], reason='no runner')
@idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target']) @idf_parametrize('target', soc_filtered_targets('SOC_TWAI_SUPPORTED == 1'), indirect=['target'])
+1
View File
@@ -125,6 +125,7 @@ env_markers =
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_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 lp_i2s: lp_i2s runner tested with hp_i2s
ram_app: ram_app runners ram_app: ram_app runners
recovery_bootloader: Runner with recovery bootloader offset set in recovery_bootloader: Runner with recovery bootloader offset set in