fix(driver_twai): add fd test on ci

components/esp_driver_twai/test_apps/test_twai/main/test_twai_network.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: 2025 Espressif Systems (Shanghai) CO LTD
+ * SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
  *
  * SPDX-License-Identifier: Apache-2.0
  */
@@ -116,3 +116,63 @@ TEST_CASE("twai_remote_request", "[twai_net]")
     TEST_ESP_OK(twai_node_disable(node_hdl));
     TEST_ESP_OK(twai_node_delete(node_hdl));
 }
+
+#define TEST_RANDOM_FRAME_NUM       100
+TEST_CASE("twai_fd_echo_random_trans", "[twai_net]")
+{
+    twai_node_handle_t node_hdl;
+    twai_onchip_node_config_t node_config = {};
+    node_config.io_cfg.tx = TEST_TX_GPIO;
+    node_config.io_cfg.rx = TEST_RX_GPIO;
+    node_config.io_cfg.quanta_clk_out = GPIO_NUM_NC;
+    node_config.io_cfg.bus_off_indicator = GPIO_NUM_NC;
+    node_config.bit_timing.bitrate = 250000;
+    node_config.data_timing.bitrate = 4000000;
+    node_config.tx_queue_depth = 3;
+
+    TEST_ESP_OK(twai_new_node_onchip(&node_config, &node_hdl));
+    ESP_LOGI("Test", "driver installed");
+
+    uint8_t rx_buffer[TWAIFD_FRAME_MAX_LEN] = {0};
+    twai_frame_t rx_frame = {};
+    rx_frame.buffer = rx_buffer;
+    rx_frame.buffer_len = sizeof(rx_buffer);
+    uint8_t rx_msg_cnt = 0;
+    void *user_data[2] = {&rx_msg_cnt, &rx_frame};
+
+    twai_event_callbacks_t user_cbs = {};
+    user_cbs.on_rx_done = test_listen_only_rx_cb;
+    TEST_ESP_OK(twai_node_register_event_callbacks(node_hdl, &user_cbs, user_data));
+    TEST_ESP_OK(twai_node_enable(node_hdl));
+
+    uint8_t tx_buffer[TWAIFD_FRAME_MAX_LEN] = {0};
+    twai_frame_t tx_frame = {};
+    tx_frame.buffer = tx_buffer;
+
+    for (size_t i = 0; i < TEST_RANDOM_FRAME_NUM; i++) {
+        ESP_LOGI("Test", "waiting random frame %d ...", (int)i);
+        while (rx_msg_cnt <= i) {
+            vTaskDelay(1);
+        }
+
+        const size_t rx_len = twaifd_dlc2len(rx_frame.header.dlc);
+        ESP_LOGI("Test", "RX: %lx [%d] fd %d, brs %d, ext %d",
+                 rx_frame.header.id, (int)rx_len, rx_frame.header.fdf, rx_frame.header.brs, rx_frame.header.ide);
+        ESP_LOG_BUFFER_HEX("Data", rx_frame.buffer, rx_len);
+        TEST_ASSERT_LESS_OR_EQUAL(sizeof(rx_buffer), rx_len);
+        TEST_ASSERT_TRUE(rx_frame.header.fdf || rx_len <= 8);
+
+        // ESP only change data and send it back, PC will check the frame is correct
+        for (size_t j = 0; j < rx_len; j++) {
+            tx_buffer[j] = rx_frame.buffer[j] + 2;
+        }
+        tx_frame.header = rx_frame.header;
+        tx_frame.buffer_len = rx_len;
+        ESP_LOGI("Test", "send echo frame");
+        TEST_ESP_OK(twai_node_transmit(node_hdl, &tx_frame, 1000));
+        TEST_ESP_OK(twai_node_transmit_wait_all_done(node_hdl, 1000));
+    }
+
+    TEST_ESP_OK(twai_node_disable(node_hdl));
+    TEST_ESP_OK(twai_node_delete(node_hdl));
+}

components/esp_driver_twai/test_apps/test_twai/pytest_driver_twai.py

@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: 2025-2026 Espressif Systems (Shanghai) CO LTD
 # SPDX-License-Identifier: Apache-2.0
 import os
+import random
 import subprocess
 import time
 
@@ -54,6 +55,7 @@ def esp_enter_flash_mode(dut: Dut) -> None:
 def fixture_create_socket_can() -> Bus:
     # Set up the socket CAN with the bitrate
     start_command = f'sudo -n ip link set {can_env} up type can bitrate 250000'
+    fdstart_command = f'sudo -n ip link set {can_env} up type can bitrate 250000 dbitrate 4000000 fd on'
     stop_command = f'sudo -n ip link set {can_env} down'
     status_command = f'sudo -n ip -details link show {can_env}'
 
@@ -64,10 +66,15 @@ def fixture_create_socket_can() -> Bus:
 
         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)
+        result = subprocess.run(fdstart_command, shell=True, capture_output=True, text=True)
+        fd_enabled = result.returncode == 0
+        if not fd_enabled:
+            subprocess.run(start_command, shell=True, capture_output=True, text=True)
+        print(f'FD mode {"Enabled" if fd_enabled else "NOT supported"} on USB CAN')
 
         time.sleep(0.5)
-        bus = Bus(interface='socketcan', channel=f'{can_env}', bitrate=250000)
+        bus = Bus(interface='socketcan', channel=f'{can_env}', bitrate=250000, fd=fd_enabled)
+        bus.fd_enabled = fd_enabled
         yield bus  # test invoked here
 
         bus.shutdown()
@@ -135,3 +142,63 @@ def test_driver_twai_remote_request(dut: Dut, socket_can: Bus) -> None:
         dut.expect_unity_test_output(timeout=10)
     finally:
         esp_enter_flash_mode(dut)
+
+
+# Test send 100 random frames and ESP echo them back with data + 2
+# Verify the frame is received correctly
+@pytest.mark.twai_adapter
+@idf_parametrize('target', soc_filtered_targets('SOC_TWAI_FD_SUPPORTED == 1'), indirect=['target'])
+@pytest.mark.temp_skip_ci(targets=['esp32s31', 'esp32h4'], reason='no runner')
+@pytest.mark.parametrize('config', ['release'], indirect=True)
+def test_driver_twai_fd_trans(dut: Dut, socket_can: Bus) -> None:
+    if not socket_can.fd_enabled:
+        pytest.fail(f'CAN interface "{can_env}" does not support CAN FD')
+    try:
+        dut.expect_exact('Press ENTER to see the list of tests')
+        dut.write('"twai_fd_echo_random_trans"')
+
+        # wait the DUT to finish initialize
+        time.sleep(0.5)
+
+        rng = random.Random(0x20260702)
+        canfd_lengths = [0, 1, 2, 3, 4, 5, 6, 7, 8, 12, 16, 20, 24, 32, 48, 64]
+
+        for index in range(100):
+            is_fd = rng.choice([False, True])
+            is_extended_id = rng.choice([False, True])
+            arbitration_id = rng.randrange(0x20000000 if is_extended_id else 0x800)
+            data_len = rng.choice(canfd_lengths if is_fd else list(range(9)))
+            message = Message(
+                arbitration_id=arbitration_id,
+                is_extended_id=is_extended_id,
+                is_fd=is_fd,
+                bitrate_switch=is_fd and rng.choice([False, True]),
+                data=bytes(rng.randrange(256) for _ in range(data_len)),
+            )
+            esp_to_usb_data = bytes((byte + 2) & 0xFF for byte in message.data)
+            print(f'USB TX {index}:', message, 'Return:', socket_can.send(message, timeout=0.2))
+
+            deadline = time.time() + 5.0
+            reply = None
+            while time.time() < deadline:
+                candidate = socket_can.recv(timeout=0.2)
+                if (
+                    candidate is not None
+                    and candidate.arbitration_id == message.arbitration_id
+                    and candidate.is_extended_id == message.is_extended_id
+                ):
+                    reply = candidate
+                    break
+            if reply is None:
+                raise Exception('CAN frame not received')
+            print(f'USB RX {index}: {reply}')
+
+            assert reply.arbitration_id == message.arbitration_id
+            assert reply.is_extended_id == message.is_extended_id
+            assert reply.is_fd == message.is_fd
+            assert reply.bitrate_switch == message.bitrate_switch
+            assert bytes(reply.data) == esp_to_usb_data
+
+        dut.expect_unity_test_output(timeout=10)
+    finally:
+        esp_enter_flash_mode(dut)