adam_di_akimow/lift_indicator_suite
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

# Created HIL and HOST tests for bsp_can module

Browse source
This commit is contained in:
Dmitry Akimov 2026-03-31 13:07:39 +03:00
parent ae90c303e1
commit 6c23315e1e
16 changed files with 1432 additions and 999 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
CMakePresets.json
View file

@ -142,6 +142,7 @@
"test_bsp_led",
"test_log",
"test_bsp_opto",
"test_bsp_can",
"uart_host_mock_example",
"test_ring_buffer",
"test_timeout_pattern",
@ -155,6 +156,7 @@
"targets": [
"test_bsp_led",
"test_log",
"test_bsp_can",
"uart_host_mock_example",
"test_bsp_opto",
"test_ring_buffer",
@ -168,6 +170,7 @@
"configurePreset": "target-debug",
"targets": [
"test_host_uart",
"test_hil_can",
"test_hil_opto"
]
}

925
bsp/can/PLAN.md
View file

@ -1,925 +0,0 @@
# План разработки bsp_can
## Обзор
Единый BSP-модуль `bsp/can/` для работы с FlexCAN2 (CAN1) на MIMXRT1052.
Одна шина, один трансивер SN65HVD230D. Модуль не зависит от FreeRTOS —
интеграция с RTOS делается через callback на стороне приложения.
**Приоритеты первой итерации (в порядке реализации):**
1. Базовый TX/RX (send + polling receive)
2. Фильтрация по ID (STD + EXT)
3. HIL-тесты (M5Stack + MicroPython CAN)
4. Host-тесты (fff mocks)
5. Callback-механизм для FreeRTOS (вторая итерация)
---
## Архитектура: один модуль — два паттерна использования
```
bsp/can/ ← единственный модуль, без FreeRTOS
include/bsp/can.h ← публичный API
src/bsp_can.c ← реализация поверх fsl_flexcan
mocks/bsp_can_mock.h ← fff-мок для host-тестов app-кода (будущее)
firmware/test/ ← bare-metal: bsp_can_receive() polling
firmware/tft_app/ ← FreeRTOS: callback → xQueueSendFromISR
```
Почему один модуль работает в обоих контекстах:
- `bsp_can_receive()` — блокирующий polling с таймаутом (bare-metal, HIL)
- `bsp_can_register_rx_callback()` — вызов из ISR, не блокируется (FreeRTOS)
- Вызывающий код выбирает один из двух механизмов, модуль не знает о контексте
---
## Привязка к железу
Из схемы (лист 3 + лист 5):
| Сигнал | Пин MCU | GPIO_AD_B0 | Функция |
|-----------|-----------------|--------------|--------------|
| CAN_TX | GPIO_AD_B0_14 | H14 | FLEXCAN2_TX |
| CAN_RX | GPIO_AD_B0_15 | L10 | FLEXCAN2_RX |
Трансивер: SN65HVD230D (U9), CAN_H/CAN_L через RCLAMP0524PA (VD12).
Терминация: R30 120Ω + J1 (jumper для подключения).
> **Важно:** на RT1052 CAN1 в SDK = FlexCAN2 периферия (base `CAN2`).
> Нумерация в SDK сдвинута: `CAN1` base → не используется, `CAN2` base → наша шина.
---
## Этап 1 — Публичный API: `bsp/can/include/bsp/can.h`
```c
#pragma once
#include "bsp/common.h"
#include <stdint.h>
#include <stdbool.h>
#include <stddef.h>
/* ── Константы ── */
#define BSP_CAN_DATA_MAX_LEN 8U
#define BSP_CAN_FILTER_MAX 16U /* MB6..MB21 под RX-фильтры */
/* ── Типы ── */
typedef struct bsp_can_frame_s {
uint32_t id; /* 11-bit STD или 29-bit EXT */
uint8_t dlc; /* 0..8 */
bool is_extended; /* false = STD, true = EXT */
bool is_remote; /* RTR */
uint8_t data[BSP_CAN_DATA_MAX_LEN];
} bsp_can_frame_t;
typedef struct bsp_can_config_s {
uint32_t bitrate; /* например 500000 */
} bsp_can_config_t;
/**
* Callback из ISR-контекста.
* Реализация НЕ ДОЛЖНА блокироваться.
* Типичное использование: xQueueSendFromISR().
*/
typedef void (*bsp_can_rx_callback_t)(const bsp_can_frame_t *p_frame,
void *p_user_ctx);
/* ── Init / Deinit ── */
bsp_status_t bsp_can_init(const bsp_can_config_t *p_config);
void bsp_can_deinit(void);
/* ── Фильтрация ── */
/**
* Настроить RX-фильтр на конкретный Message Buffer.
*
* @param index 0 .. BSP_CAN_FILTER_MAX-1
* @param id CAN ID для фильтрации
* @param mask битовая маска (1 = проверять, 0 = игнорировать)
* @param is_extended true = 29-bit EXT, false = 11-bit STD
*/
bsp_status_t bsp_can_set_filter(uint8_t index,
uint32_t id,
uint32_t mask,
bool is_extended);
/** Принимать все фреймы (сброс всех фильтров). */
bsp_status_t bsp_can_accept_all(void);
/* ── TX ── */
/**
* Отправить CAN-фрейм. Блокируется до завершения или таймаута.
*
* @param p_frame фрейм для отправки
* @param timeout_ms таймаут в мс (0 = без ожидания)
* @return BSP_OK, BSP_ERR_TIMEOUT, BSP_ERR_PARAM
*/
bsp_status_t bsp_can_send(const bsp_can_frame_t *p_frame,
uint32_t timeout_ms);
/* ── RX: polling ── */
/**
* Принять CAN-фрейм (polling). Блокируется до приёма или таймаута.
*
* @param p_frame буфер для принятого фрейма
* @param timeout_ms таймаут в мс (0 = проверить и вернуться)
* @return BSP_OK, BSP_ERR_TIMEOUT
*/
bsp_status_t bsp_can_receive(bsp_can_frame_t *p_frame,
uint32_t timeout_ms);
/* ── RX: callback (для FreeRTOS bridge) ── */
/**
* Зарегистрировать callback для приёма из ISR.
* При регистрации callback, polling через bsp_can_receive() отключается.
*
* @param callback функция-обработчик (NULL = отключить callback)
* @param p_user_ctx пользовательский контекст, передаётся в callback
*/
bsp_status_t bsp_can_register_rx_callback(bsp_can_rx_callback_t callback,
void *p_user_ctx);
```
---
## Этап 2 — Реализация: `bsp/can/src/bsp_can.c`
Ключевые решения по реализации:
**Message Buffers (MB) FlexCAN2:**
| MB | Назначение |
|-------|-----------------------------------|
| 0–5 | TX (отправка, round-robin) |
| 6–21 | RX с индивидуальными фильтрами |
| 22–31 | Резерв |
**Polling vs Callback — взаимоисключающие:**
```c
static bsp_can_rx_callback_t s_rx_callback;
static void *s_rx_user_ctx;
static volatile bool s_use_callback;
/* Внутренний ISR-обработчик */
static void can_rx_isr_handler(const bsp_can_frame_t *p_frame)
{
if (s_use_callback && s_rx_callback != NULL) {
s_rx_callback(p_frame, s_rx_user_ctx);
} else {
/* положить в internal ring buffer для polling */
ring_buffer_write(&s_rx_ring, p_frame, sizeof(*p_frame));
}
}
```
**Внутренний RX ring buffer** для polling-режима — переиспользуем `utils/ring_buffer/`.
**SDK-функции, которые будем вызывать (нужны для мока):**
```
FLEXCAN_Init()
FLEXCAN_Deinit()
FLEXCAN_SetTimingConfig()
FLEXCAN_SetRxMbConfig()
FLEXCAN_SetTxMbConfig()
FLEXCAN_SetRxIndividualMask()
FLEXCAN_TransferSendBlocking() /* для TX */
FLEXCAN_TransferReceiveBlocking() /* для RX polling */
FLEXCAN_EnableMbInterrupts()
FLEXCAN_DisableMbInterrupts()
FLEXCAN_ReadRxMb()
```
---
## Этап 3 — CMake: `bsp/can/CMakeLists.txt`
```cmake
add_library(bsp_can STATIC
src/bsp_can.c
)
target_include_directories(bsp_can
PUBLIC include
PRIVATE ${BSP_GENERATED_DIR}
)
target_link_libraries(bsp_can
PUBLIC bsp_common
PRIVATE bsp_tick # для таймаутов
ring_buffer # внутренний RX буфер
)
```
Подключить в `bsp/CMakeLists.txt`:
```cmake
add_subdirectory(can)
```
---
## Этап 4 — Host-тесты
### 4.1 Stub-хедер: `tests/host/mocks/fsl_flexcan.h`
```c
#pragma once
#include <stdint.h>
#include <stdbool.h>
#include <stddef.h>
/* Минимальные типы для компиляции bsp_can.c на хосте */
typedef struct { uint32_t reserved[256]; } CAN_Type;
typedef enum {
kStatus_Success = 0,
kStatus_Fail = 1,
kStatus_FLEXCAN_RxOverflow = 100,
} status_t;
typedef enum {
kFLEXCAN_FrameFormatStandard = 0,
kFLEXCAN_FrameFormatExtend = 1,
} flexcan_frame_format_e;
typedef enum {
kFLEXCAN_FrameTypeData = 0,
kFLEXCAN_FrameTypeRemote = 1,
} flexcan_frame_type_e;
typedef struct {
uint32_t baudRate;
} flexcan_config_s;
typedef struct {
uint32_t propSeg;
uint32_t phaseSeg1;
uint32_t phaseSeg2;
uint32_t rJumpwidth;
uint32_t preDivider;
} flexcan_timing_config_s;
typedef struct {
volatile uint32_t id;
volatile uint32_t cs;
volatile uint8_t data[8];
volatile uint8_t dlc;
} flexcan_frame_s;
typedef struct {
uint32_t id;
flexcan_frame_format_e format;
flexcan_frame_type_e type;
} flexcan_rx_mb_config_s;
typedef struct {
flexcan_frame_s *p_frame;
uint8_t mb_idx;
} flexcan_mb_transfer_s;
/* Сигнатуры — реализации предоставляет fff */
void FLEXCAN_Init(CAN_Type *p_base, const flexcan_config_s *p_config,
uint32_t src_clk);
void FLEXCAN_Deinit(CAN_Type *p_base);
void FLEXCAN_GetDefaultConfig(flexcan_config_s *p_config);
void FLEXCAN_SetRxMbConfig(CAN_Type *p_base, uint8_t mb_idx,
const flexcan_rx_mb_config_s *p_config,
bool enable);
void FLEXCAN_SetTxMbConfig(CAN_Type *p_base, uint8_t mb_idx, bool enable);
void FLEXCAN_SetRxIndividualMask(CAN_Type *p_base, uint8_t mb_idx,
uint32_t mask);
status_t FLEXCAN_TransferSendBlocking(CAN_Type *p_base, uint8_t mb_idx,
flexcan_frame_s *p_frame);
status_t FLEXCAN_ReadRxMb(CAN_Type *p_base, uint8_t mb_idx,
flexcan_frame_s *p_frame);
void FLEXCAN_EnableMbInterrupts(CAN_Type *p_base, uint32_t mask);
void FLEXCAN_DisableMbInterrupts(CAN_Type *p_base, uint32_t mask);
uint32_t FLEXCAN_GetStatusFlags(CAN_Type *p_base);
void FLEXCAN_ClearStatusFlags(CAN_Type *p_base, uint32_t mask);
```
### 4.2 Тестовый файл: `tests/host/can/test_bsp_can.c`
```c
#include "unity.h"
#include "fff.h"
DEFINE_FFF_GLOBALS;
#include "fsl_flexcan.h"
/* fff-фейки */
FAKE_VOID_FUNC(FLEXCAN_Init, CAN_Type *, const flexcan_config_s *, uint32_t);
FAKE_VOID_FUNC(FLEXCAN_Deinit, CAN_Type *);
FAKE_VOID_FUNC(FLEXCAN_GetDefaultConfig, flexcan_config_s *);
FAKE_VOID_FUNC(FLEXCAN_SetRxMbConfig, CAN_Type *, uint8_t,
const flexcan_rx_mb_config_s *, bool);
FAKE_VOID_FUNC(FLEXCAN_SetTxMbConfig, CAN_Type *, uint8_t, bool);
FAKE_VOID_FUNC(FLEXCAN_SetRxIndividualMask, CAN_Type *, uint8_t, uint32_t);
FAKE_VALUE_FUNC(status_t, FLEXCAN_TransferSendBlocking, CAN_Type *, uint8_t,
flexcan_frame_s *);
FAKE_VALUE_FUNC(status_t, FLEXCAN_ReadRxMb, CAN_Type *, uint8_t,
flexcan_frame_s *);
#include "bsp/can.h"
void setUp(void)
{
RESET_FAKE(FLEXCAN_Init);
RESET_FAKE(FLEXCAN_Deinit);
RESET_FAKE(FLEXCAN_GetDefaultConfig);
RESET_FAKE(FLEXCAN_SetRxMbConfig);
RESET_FAKE(FLEXCAN_SetTxMbConfig);
RESET_FAKE(FLEXCAN_SetRxIndividualMask);
RESET_FAKE(FLEXCAN_TransferSendBlocking);
RESET_FAKE(FLEXCAN_ReadRxMb);
FFF_RESET_HISTORY();
}
void tearDown(void) { }
/* ── Init ── */
void test_init_calls_flexcan_init(void)
{
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_status_t status = bsp_can_init(&cfg);
TEST_ASSERT_EQUAL(BSP_OK, status);
TEST_ASSERT_EQUAL(1, FLEXCAN_Init_fake.call_count);
}
void test_init_null_config_returns_err(void)
{
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_init(NULL));
}
void test_deinit_calls_flexcan_deinit(void)
{
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_can_init(&cfg);
bsp_can_deinit();
TEST_ASSERT_EQUAL(1, FLEXCAN_Deinit_fake.call_count);
}
/* ── TX ── */
void test_send_valid_frame(void)
{
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_can_init(&cfg);
FLEXCAN_TransferSendBlocking_fake.return_val = kStatus_Success;
bsp_can_frame_t frame = {
.id = 0x123, .dlc = 8, .is_extended = false, .is_remote = false,
.data = {0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03, 0x04}
};
TEST_ASSERT_EQUAL(BSP_OK, bsp_can_send(&frame, 100U));
TEST_ASSERT_EQUAL(1, FLEXCAN_TransferSendBlocking_fake.call_count);
}
void test_send_null_frame_returns_err(void)
{
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_can_init(&cfg);
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_send(NULL, 100U));
}
void test_send_dlc_over_8_returns_err(void)
{
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_can_init(&cfg);
bsp_can_frame_t frame = { .id = 0x123, .dlc = 9 };
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_send(&frame, 100U));
}
/* ── Фильтры ── */
void test_set_filter_std(void)
{
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_can_init(&cfg);
bsp_status_t s = bsp_can_set_filter(0, 0x123, 0x7FFU, false);
TEST_ASSERT_EQUAL(BSP_OK, s);
TEST_ASSERT_GREATER_THAN(0, FLEXCAN_SetRxMbConfig_fake.call_count);
TEST_ASSERT_GREATER_THAN(0, FLEXCAN_SetRxIndividualMask_fake.call_count);
}
void test_set_filter_index_out_of_range(void)
{
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_can_init(&cfg);
TEST_ASSERT_EQUAL(BSP_ERR_PARAM,
bsp_can_set_filter(BSP_CAN_FILTER_MAX, 0x123, 0x7FF, false));
}
void test_accept_all_configures_all_mb(void)
{
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_can_init(&cfg);
TEST_ASSERT_EQUAL(BSP_OK, bsp_can_accept_all());
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_init_calls_flexcan_init);
RUN_TEST(test_init_null_config_returns_err);
RUN_TEST(test_deinit_calls_flexcan_deinit);
RUN_TEST(test_send_valid_frame);
RUN_TEST(test_send_null_frame_returns_err);
RUN_TEST(test_send_dlc_over_8_returns_err);
RUN_TEST(test_set_filter_std);
RUN_TEST(test_set_filter_index_out_of_range);
RUN_TEST(test_accept_all_configures_all_mb);
return UNITY_END();
}
```
### 4.3 Регистрация: `tests/host/CMakeLists.txt`
```cmake
add_host_test(
NAME test_bsp_can
SOURCES can/test_bsp_can.c
${PROJECT_SOURCE_DIR}/bsp/can/src/bsp_can.c
${PROJECT_SOURCE_DIR}/utils/ring_buffer/ring_buffer.c
INCLUDES
${PROJECT_SOURCE_DIR}/bsp/can/include
${PROJECT_SOURCE_DIR}/bsp/common/include
${PROJECT_SOURCE_DIR}/utils/ring_buffer
MOCKS
${BSP_MOCKS_DIR}
)
```
### 4.4 `CMakePresets.json` — добавить в `host-debug-build`
```json
"targets": [
"test_bsp_led",
"test_log",
"test_bsp_opto",
"test_ring_buffer",
"test_timeout_pattern",
"uart_host_mock_example",
"test_bsp_can"
]
```
---
## Этап 5 — HIL-тесты
### 5.1 Инфраструктура на стороне хоста: M5Stack CAN
M5Stack StamPLC с CAN-модулем (TJA1050 или MCP2515) работает как
CAN-адаптер. Связь host ↔ M5Stack через USB CDC (MicroPython).
**Агент на M5Stack** (`tools/hil/m5/can_agent.py` — MicroPython):
```
Команды через serial:
CAN_INIT <bitrate> → OK / ERR
CAN_SEND <id> <dlc> <hex> → OK / ERR
CAN_RECV <timeout_ms> → <id> <dlc> <hex> / TIMEOUT
CAN_FILTER <id> <mask> → OK / ERR
```
**Python-обёртка** (`tools/hil/m5/can_bus.py`):
```python
class M5CanBus:
"""CAN-адаптер через M5Stack serial."""
def __init__(self, port: str, baudrate: int = 115200):
self.ser = serial.Serial(port, baudrate, timeout=2.0)
def init_can(self, bitrate: int = 500_000) -> None: ...
def send(self, can_id: int, data: bytes,
is_extended: bool = False) -> None: ...
def recv(self, timeout_ms: int = 1000) -> CanFrame | None: ...
def set_filter(self, can_id: int, mask: int) -> None: ...
```
### 5.2 C-прошивка: `tests/target/can/main.c`
```c
#include "board.h"
#include "bsp/led.h"
#include "bsp/tick.h"
#include "bsp/uart_host.h"
#include "bsp/can.h"
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#define CLI_BAUD_RATE 115200U
#define CLI_LINE_MAX 128U
#define CLI_RX_TIMEOUT 100U
static size_t cli_read_line(uint8_t *p_buf, size_t max_len) { /* как в шаблоне */ }
/* Вспомогательные функции парсинга */
static bool parse_hex_bytes(const char *p_hex, uint8_t *p_out, uint8_t len) { ... }
static void format_hex_bytes(const uint8_t *p_data, uint8_t len, char *p_out) { ... }
static void cli_process_line(const char *p_line)
{
if (strncmp(p_line, "PING", 4U) == 0) {
bsp_uart_host_write_str("PONG\r\n");
}
/* CAN_INIT <bitrate> */
else if (strncmp(p_line, "CAN_INIT ", 9U) == 0) {
uint32_t bitrate = (uint32_t)strtoul(p_line + 9, NULL, 10);
bsp_can_config_t cfg = { .bitrate = bitrate };
bsp_status_t s = bsp_can_init(&cfg);
bsp_uart_host_write_str(s == BSP_OK ? "OK\r\n" : "ERR\r\n");
}
/* CAN_SEND <id_hex> <dlc> <data_hex> */
else if (strncmp(p_line, "CAN_SEND ", 9U) == 0) {
/* парсинг id, dlc, data из строки */
bsp_can_frame_t frame = { /* заполнить */ };
bsp_status_t s = bsp_can_send(&frame, 500U);
bsp_uart_host_write_str(s == BSP_OK ? "OK\r\n" : "ERR\r\n");
}
/* CAN_RECV <timeout_ms> */
else if (strncmp(p_line, "CAN_RECV ", 9U) == 0) {
uint32_t timeout = (uint32_t)strtoul(p_line + 9, NULL, 10);
bsp_can_frame_t frame;
if (bsp_can_receive(&frame, timeout) == BSP_OK) {
char buf[64];
/* формат: <id_hex> <dlc> <data_hex> */
bsp_uart_host_write_str(buf);
bsp_uart_host_write_str("\r\n");
} else {
bsp_uart_host_write_str("TIMEOUT\r\n");
}
}
/* CAN_FILTER <index> <id_hex> <mask_hex> <ext:0|1> */
else if (strncmp(p_line, "CAN_FILTER ", 11U) == 0) {
/* парсинг index, id, mask, is_extended */
bsp_status_t s = bsp_can_set_filter(/* ... */);
bsp_uart_host_write_str(s == BSP_OK ? "OK\r\n" : "ERR\r\n");
}
/* CAN_ACCEPT_ALL */
else if (strncmp(p_line, "CAN_ACCEPT_ALL", 14U) == 0) {
bsp_status_t s = bsp_can_accept_all();
bsp_uart_host_write_str(s == BSP_OK ? "OK\r\n" : "ERR\r\n");
}
else if (p_line[0] != '\0') {
bsp_uart_host_write_str("ERR_UNKNOWN\r\n");
}
}
int main(void)
{
board_hw_init();
bsp_tick_init();
bsp_led_init();
bsp_uart_host_init(CLI_BAUD_RATE);
bsp_led_on(LED_HEARTBEAT);
while (bsp_uart_host_rx_available() == 0U) {
bsp_uart_host_write_str("READY\r\n");
bsp_delay(200U);
}
static uint8_t s_line_buf[CLI_LINE_MAX];
for (;;) {
size_t len = cli_read_line(s_line_buf, sizeof(s_line_buf));
if (len > 0U) {
cli_process_line((const char *)s_line_buf);
}
}
}
```
### 5.3 CMake: `tests/target/can/CMakeLists.txt`
```cmake
set(TARGET_NAME test_hil_can)
add_executable(${TARGET_NAME}
main.c
${BSP_GENERATED}/clock_config.c
${BSP_STARTUP_FILE}
${BSP_SYSCALLS_FILE}
)
target_link_options(${TARGET_NAME} PRIVATE
-T${CMAKE_SOURCE_DIR}/cmake/linker/MIMXRT1052xxxxx_ram.ld
-Wl,--gc-sections
-Wl,--print-memory-usage
-Wl,-Map=${CMAKE_CURRENT_BINARY_DIR}/${TARGET_NAME}.map
)
target_link_libraries(${TARGET_NAME} PRIVATE
bsp_boot_ram
bsp_board
bsp_led
bsp_tick
bsp_uart_host
bsp_can
)
add_custom_command(TARGET ${TARGET_NAME} POST_BUILD
COMMAND ${CMAKE_SIZE} $<TARGET_FILE:${TARGET_NAME}>
COMMENT "Size: ${TARGET_NAME}"
)
```
### 5.4 Подключить в `tests/target/CMakeLists.txt`
```cmake
add_subdirectory(host_uart)
add_subdirectory(can) # ← добавить
```
### 5.5 `CMakePresets.json` — добавить в `target-debug-build`
```json
"targets": [
"test_host_uart",
"test_hil_opto",
"test_hil_can"
]
```
### 5.6 Pytest: `tools/hil/test_can.py`
```python
"""test_can.py — HIL тесты bsp_can."""
import pytest
import time
from conftest import uart_cmd
from m5.can_bus import M5CanBus
@pytest.fixture(scope="module")
def can_bus(request) -> M5CanBus:
"""CAN-адаптер на стороне хоста (M5Stack)."""
port = request.config.getoption("--m5-port")
bus = M5CanBus(port)
bus.init_can(500_000)
yield bus
bus.close()
class TestCanBasic:
@pytest.fixture(autouse=True)
def _setup(self, loaded_can, uart, can_bus):
self.ser = uart
self.can = can_bus
# Инициализировать CAN на MCU
assert uart_cmd(self.ser, "CAN_INIT 500000") == "OK"
assert uart_cmd(self.ser, "CAN_ACCEPT_ALL") == "OK"
@pytest.mark.smoke
def test_ping(self):
"""Базовая проверка UART-канала."""
assert uart_cmd(self.ser, "PING") == "PONG"
def test_mcu_tx_host_rx(self):
"""MCU отправляет фрейм, M5 принимает."""
assert uart_cmd(self.ser, "CAN_SEND 123 8 DEADBEEF01020304") == "OK"
frame = self.can.recv(timeout_ms=1000)
assert frame is not None
assert frame.id == 0x123
assert frame.dlc == 8
assert frame.data == bytes.fromhex("DEADBEEF01020304")
def test_host_tx_mcu_rx(self):
"""M5 отправляет фрейм, MCU принимает."""
self.can.send(0x456, bytes.fromhex("AABBCCDD"), is_extended=False)
resp = uart_cmd(self.ser, "CAN_RECV 1000")
assert resp != "TIMEOUT"
# Парсинг ответа: "456 4 AABBCCDD"
parts = resp.split()
assert int(parts[0], 16) == 0x456
assert int(parts[1]) == 4
def test_loopback_echo(self):
"""M5 отправляет → MCU принимает → MCU отправляет назад → M5 принимает."""
self.can.send(0x100, bytes.fromhex("01020304"))
# MCU должен принять
resp = uart_cmd(self.ser, "CAN_RECV 1000")
assert resp != "TIMEOUT"
# MCU отправляет эхо
assert uart_cmd(self.ser, "CAN_SEND 100 4 01020304") == "OK"
echo = self.can.recv(timeout_ms=1000)
assert echo is not None
assert echo.id == 0x100
class TestCanFilter:
@pytest.fixture(autouse=True)
def _setup(self, loaded_can, uart, can_bus):
self.ser = uart
self.can = can_bus
assert uart_cmd(self.ser, "CAN_INIT 500000") == "OK"
def test_filter_accepts_matching_id(self):
"""Фильтр 0x200/0x7FF — принимает 0x200."""
assert uart_cmd(self.ser, "CAN_FILTER 0 200 7FF 0") == "OK"
self.can.send(0x200, bytes(4))
resp = uart_cmd(self.ser, "CAN_RECV 1000")
assert resp != "TIMEOUT"
assert resp.startswith("200 ")
def test_filter_rejects_non_matching_id(self):
"""Фильтр 0x200/0x7FF — отвергает 0x300."""
assert uart_cmd(self.ser, "CAN_FILTER 0 200 7FF 0") == "OK"
self.can.send(0x300, bytes(4))
resp = uart_cmd(self.ser, "CAN_RECV 500")
assert resp == "TIMEOUT"
def test_filter_mask_partial(self):
"""Маска 0x7F0 — принимает 0x201..0x20F."""
assert uart_cmd(self.ser, "CAN_FILTER 0 200 7F0 0") == "OK"
self.can.send(0x205, bytes(2))
resp = uart_cmd(self.ser, "CAN_RECV 1000")
assert resp != "TIMEOUT"
def test_accept_all_after_filter(self):
"""accept_all сбрасывает фильтры."""
assert uart_cmd(self.ser, "CAN_FILTER 0 200 7FF 0") == "OK"
assert uart_cmd(self.ser, "CAN_ACCEPT_ALL") == "OK"
self.can.send(0x300, bytes(4))
resp = uart_cmd(self.ser, "CAN_RECV 1000")
assert resp != "TIMEOUT"
class TestCanExtended:
@pytest.fixture(autouse=True)
def _setup(self, loaded_can, uart, can_bus):
self.ser = uart
self.can = can_bus
assert uart_cmd(self.ser, "CAN_INIT 500000") == "OK"
assert uart_cmd(self.ser, "CAN_ACCEPT_ALL") == "OK"
def test_ext_id_tx_rx(self):
"""Отправка/приём EXT ID (29-bit)."""
assert uart_cmd(self.ser,
"CAN_SEND 1ABCDEF0 4 AABBCCDD") == "OK"
frame = self.can.recv(timeout_ms=1000)
assert frame is not None
assert frame.id == 0x1ABCDEF0
assert frame.is_extended is True
```
### 5.7 Фикстура загрузки: `tools/hil/conftest.py`
```python
@pytest.fixture(scope="module")
def loaded_can(request: pytest.FixtureRequest) -> None:
_load_elf(
request,
Path(cfg.BUILD_DIR) / "tests/target/can/test_hil_can.elf",
)
```
### 5.8 `just/host.just` — алиасы
```just
[doc('Запустить HIL-тест CAN')]
[group('hil')]
hil-can:
HIL_BUILD_DIR={{_hil_build}} \
uv run --directory {{HIL_DIR}} pytest test_can.py -v
```
---
## Этап 6 (будущее) — Callback для FreeRTOS
Не входит в первую итерацию. Заглушка в API уже есть (`bsp_can_register_rx_callback`).
Реализация в `firmware/tft_app/`:
```c
static QueueHandle_t s_can_queue;
static void can_isr_to_queue(const bsp_can_frame_t *p_frame, void *p_ctx)
{
BaseType_t higher_woken = pdFALSE;
xQueueSendFromISR(s_can_queue, p_frame, &higher_woken);
portYIELD_FROM_ISR(higher_woken);
}
void can_task(void *p_param)
{
s_can_queue = xQueueCreate(16, sizeof(bsp_can_frame_t));
bsp_can_config_t cfg = { .bitrate = 500000U };
bsp_can_init(&cfg);
bsp_can_register_rx_callback(can_isr_to_queue, NULL);
bsp_can_frame_t rx;
for (;;) {
if (xQueueReceive(s_can_queue, &rx, portMAX_DELAY) == pdTRUE) {
/* dispatch по rx.id */
}
}
}
```
---
## Файловая карта изменений
```
НОВЫЕ ФАЙЛЫ:
bsp/can/CMakeLists.txt
bsp/can/include/bsp/can.h
bsp/can/src/bsp_can.c
tests/host/can/test_bsp_can.c
tests/host/mocks/fsl_flexcan.h
tests/target/can/main.c
tests/target/can/CMakeLists.txt
tools/hil/test_can.py
tools/hil/m5/can_agent.py (MicroPython на M5Stack)
tools/hil/m5/can_bus.py (Python-обёртка)
ИЗМЕНЕНИЯ В СУЩЕСТВУЮЩИХ ФАЙЛАХ:
bsp/CMakeLists.txt ← add_subdirectory(can)
tests/host/CMakeLists.txt ← add_host_test(... test_bsp_can ...)
tests/target/CMakeLists.txt ← add_subdirectory(can)
CMakePresets.json ← test_bsp_can в host-debug-build
← test_hil_can в target-debug-build
tools/hil/conftest.py ← loaded_can фикстура
just/host.just ← hil-can алиас
```
---
## Чеклист реализации
```
Этап 1–2: BSP модуль
[ ] bsp/can/include/bsp/can.h — публичный API
[ ] bsp/can/src/bsp_can.c — реализация (FlexCAN2)
[ ] bsp/can/CMakeLists.txt — библиотека
[ ] bsp/CMakeLists.txt — add_subdirectory(can)
Этап 3: pin_mux
[ ] bsp/generated/pin_mux.* — добавить FLEXCAN2 TX/RX в .mex
Этап 4: Host-тесты
[ ] tests/host/mocks/fsl_flexcan.h — stub
[ ] tests/host/can/test_bsp_can.c — тесты
[ ] tests/host/CMakeLists.txt — add_host_test
[ ] CMakePresets.json — host-debug-build targets
[ ] just build::test-host — зелёный прогон
Этап 5: HIL-тесты
[ ] tests/target/can/main.c — C-прошивка с CLI
[ ] tests/target/can/CMakeLists.txt — сборка
[ ] tests/target/CMakeLists.txt — add_subdirectory
[ ] CMakePresets.json — target-debug-build targets
[ ] tools/hil/m5/can_agent.py — MicroPython агент
[ ] tools/hil/m5/can_bus.py — Python-обёртка
[ ] tools/hil/test_can.py — pytest
[ ] tools/hil/conftest.py — loaded_can фикстура
[ ] just/host.just — hil-can алиас
[ ] just build::build-hil — сборка target-прошивки
[ ] just host::hil-can — зелёный прогон
```
---
## Порядок работы
```bash
# 1. Создать bsp/can (API + реализация)
# 2. Добавить pin_mux для FLEXCAN2 в .mex → перегенерировать
# 3. Host-тесты: stub + test + CMake → just build::test-host
# 4. HIL C-прошивка: target/can → just build::build-hil
# 5. M5Stack CAN-агент: залить can_agent.py
# 6. HIL pytest: test_can.py → just host::hil-can
# 7. Интегрировать в firmware/test (CLI-команды CAN)
```

16
bsp/can/src/can.c
View file

@ -347,7 +347,7 @@ bsp_status_t bsp_can_accept_all(void)
}
/*
* Настраиваем один RX MB (index 0) с маской 0x000
* Настраиваем два RX MB с маской 0x000
* (все биты игнорируются — принимает любой ID).
* Деактивируем остальные.
*/
@ -363,23 +363,29 @@ bsp_status_t bsp_can_accept_all(void)
g_s_rx_mb_active_mask = 0U;
/* Настроить MB1 на приём всех STD-фреймов. */
/* Настроить MB2 на приём всех STD-фреймов. */
flexcan_rx_mb_config_t rx_mb_cfg;
rx_mb_cfg.format = kFLEXCAN_FrameFormatStandard;
rx_mb_cfg.type = kFLEXCAN_FrameTypeData;
rx_mb_cfg.id = FLEXCAN_ID_STD(0U);
FLEXCAN_SetRxMbConfig(BSP_CAN_BASE, RX_MB_FIRST, &rx_mb_cfg, true);
FLEXCAN_SetRxIndividualMask(BSP_CAN_BASE, RX_MB_FIRST, 0U);
/* STD MB — принимать все STD, отвергать EXT */
FLEXCAN_SetRxIndividualMask(
BSP_CAN_BASE, RX_MB_FIRST,
FLEXCAN_RX_MB_STD_MASK(0U, 0U, 1U)); /* mask=0: любой ID; ide=1: проверять IDE */
g_s_rx_mb_active_mask = 1U; /* Только index 0 активен. */
/* Настроить MB2 на приём всех EXT-фреймов. */
/* Настроить MB3 на приём всех EXT-фреймов. */
rx_mb_cfg.format = kFLEXCAN_FrameFormatExtend;
rx_mb_cfg.id = FLEXCAN_ID_EXT(0U);
FLEXCAN_SetRxMbConfig(BSP_CAN_BASE, RX_MB_FIRST + 1U, &rx_mb_cfg, true);
FLEXCAN_SetRxIndividualMask(BSP_CAN_BASE, RX_MB_FIRST + 1U, 0U);
/* EXT MB — принимать все EXT, отвергать STD */
FLEXCAN_SetRxIndividualMask(
BSP_CAN_BASE, RX_MB_FIRST + 1U,
FLEXCAN_RX_MB_EXT_MASK(0U, 0U, 1U)); /* mask=0: любой ID; ide=1: проверять IDE */
g_s_rx_mb_active_mask |= (1U << 1U); /* index 0 и 1 активны. */

10
just/host.just
View file

@ -340,13 +340,19 @@ hil-run:
[group('hil')]
hil-uart:
HIL_BUILD_DIR={{ _hil_build }} \
uv run --directory {{ HIL_DIR }} pytest test_uart.py -v
uv run --directory {{ HIL_DIR }} pytest 01_test_uart.py -v
[doc('Запустить HIL-тест оптоизолированных входов')]
[group('hil')]
hil-opto:
HIL_BUILD_DIR={{ _hil_build }} \
uv run --directory {{ HIL_DIR }} pytest test_opto.py -v
uv run --directory {{ HIL_DIR }} pytest 02_test_opto.py -v
[doc('Запустить HIL тест bsp_can')]
[group('hil')]
hil-can:
HIL_BUILD_DIR={{ _hil_build }} \
uv run --directory {{ HIL_DIR }} pytest 03_test_can.py -v
# =============================================================================
# ГРУППА: debug — GDB-сервер для отладки из VSCode (devcontainer)

14
tests/host/CMakeLists.txt
View file

@ -128,18 +128,18 @@ add_host_test(
${BSP_MOCKS_DIR})
# -----------------------------------------------------------------------------
# Тест кода, использующего bsp_can через мок
# ----------------------------------------------------------------------------
# Тест логики bsp_can (категория B — SDK мокается через fff в тест-файле)
# -----------------------------------------------------------------------------
add_host_test(
NAME
test_bsp_can
SOURCES
can/test_bsp_can.c
${PROJECT_SOURCE_DIR}/bsp/can/src/can.c
${PROJECT_SOURCE_DIR}/utils/ring_buffer/ring_buffer.c
${CMAKE_SOURCE_DIR}/bsp/can/src/can.c
${CMAKE_SOURCE_DIR}/utils/ring_buffer/ring_buffer.c
INCLUDES
${PROJECT_SOURCE_DIR}/bsp/can/include
${PROJECT_SOURCE_DIR}/bsp/common/include
${PROJECT_SOURCE_DIR}/utils/ring_buffer
${CMAKE_SOURCE_DIR}/bsp/can/include
${CMAKE_SOURCE_DIR}/bsp/common/include
${CMAKE_SOURCE_DIR}/utils/
MOCKS
${BSP_MOCKS_DIR})

596
tests/host/can/test_bsp_can.c
View file

@ -0,0 +1,596 @@
/**
* @file test_bsp_can.c
* @brief Host unit-тесты для bsp_can (категория B — SDK через fff).
*
* Тестируемый модуль: bsp/can/src/bsp_can.c
* Зависимости: utils/ring_buffer/ring_buffer.c (реальный, категория A)
*
* Порядок include:
* 1. unity.h + fff.h + DEFINE_FFF_GLOBALS
* 2. stub-хедеры с типами (fsl_flexcan.h, clock_config.h)
* 3. FAKE_* объявления для SDK-функций
* 4. bsp/can.h — тестируемый модуль (последним)
*/
#include "fff.h"
#include "unity.h"
DEFINE_FFF_GLOBALS;
#include "clock_config.h"
#include "fsl_flexcan.h"
/* ══════════════════════════════════════════════════════════════════════
* fff-фейки SDK-функций
* ══════════════════════════════════════════════════════════════════════ */
/* Init / Config */
FAKE_VOID_FUNC(FLEXCAN_GetDefaultConfig, flexcan_config_t *);
FAKE_VALUE_FUNC(bool, FLEXCAN_CalculateImprovedTimingValues, CAN_Type *, uint32_t, uint32_t,
flexcan_timing_config_t *);
FAKE_VOID_FUNC(FLEXCAN_Init, CAN_Type *, const flexcan_config_t *, uint32_t);
FAKE_VOID_FUNC(FLEXCAN_Deinit, CAN_Type *);
/* MB config */
FAKE_VOID_FUNC(FLEXCAN_SetTxMbConfig, CAN_Type *, uint8_t, bool);
FAKE_VOID_FUNC(FLEXCAN_SetRxMbConfig, CAN_Type *, uint8_t, const flexcan_rx_mb_config_t *, bool);
FAKE_VOID_FUNC(FLEXCAN_SetRxIndividualMask, CAN_Type *, uint8_t, uint32_t);
/* TX / RX */
FAKE_VALUE_FUNC(status_t, FLEXCAN_WriteTxMb, CAN_Type *, uint8_t, const flexcan_frame_t *);
FAKE_VALUE_FUNC(status_t, FLEXCAN_ReadRxMb, CAN_Type *, uint8_t, flexcan_frame_t *);
/* Status flags */
FAKE_VALUE_FUNC(uint64_t, FLEXCAN_GetMbStatusFlags, CAN_Type *, uint64_t);
FAKE_VOID_FUNC(FLEXCAN_ClearMbStatusFlags, CAN_Type *, uint64_t);
/* bsp_tick — управляемый «таймер» для тестов таймаутов */
FAKE_VALUE_FUNC(uint32_t, bsp_tick_get_ms);
/* ══════════════════════════════════════════════════════════════════════
* Тестируемый модуль (после всех фейков!)
* ══════════════════════════════════════════════════════════════════════ */
#include "bsp/can.h"
/* ══════════════════════════════════════════════════════════════════════
* Helpers
* ══════════════════════════════════════════════════════════════════════ */
/** Дефолтная конфигурация для тестов. */
static const bsp_can_config_t s_default_cfg = { .bitrate = 500000U };
/**
* Инициализировать модуль с дефолтными параметрами.
* Вызывает bsp_can_init() с timing_calc = true.
*/
static bsp_status_t helper_init_default(void)
{
FLEXCAN_CalculateImprovedTimingValues_fake.return_val = true;
return bsp_can_init(&s_default_cfg);
}
/**
* custom_fake для ReadRxMb — заполняет фрейм заданными данными.
* Используется через FLEXCAN_ReadRxMb_fake.custom_fake.
*/
static flexcan_frame_t s_injected_rx_frame;
static status_t read_rx_mb_inject(CAN_Type *p_base, uint8_t mb_idx, flexcan_frame_t *p_frame)
{
(void) p_base;
(void) mb_idx;
*p_frame = s_injected_rx_frame;
return kStatus_Success;
}
/**
* custom_fake для GetMbStatusFlags — возвращает флаг для первого RX MB
* только при первом вызове, потом 0.
*/
static uint32_t s_mb_flags_call_count;
static uint64_t get_mb_flags_once(CAN_Type *p_base, uint64_t mask)
{
(void) p_base;
s_mb_flags_call_count++;
/* Первый вызов — флаг есть, остальные — нет */
if (s_mb_flags_call_count == 1U)
{
return mask;
}
return 0U;
}
/**
* custom_fake для bsp_tick_get_ms — линейно нарастающее время.
*/
static uint32_t s_tick_ms;
static uint32_t tick_advancing(void)
{
uint32_t val = s_tick_ms;
s_tick_ms += 1U;
return val;
}
/* ══════════════════════════════════════════════════════════════════════
* setUp / tearDown
* ══════════════════════════════════════════════════════════════════════ */
void setUp(void)
{
RESET_FAKE(FLEXCAN_GetDefaultConfig);
RESET_FAKE(FLEXCAN_CalculateImprovedTimingValues);
RESET_FAKE(FLEXCAN_Init);
RESET_FAKE(FLEXCAN_Deinit);
RESET_FAKE(FLEXCAN_SetTxMbConfig);
RESET_FAKE(FLEXCAN_SetRxMbConfig);
RESET_FAKE(FLEXCAN_SetRxIndividualMask);
RESET_FAKE(FLEXCAN_WriteTxMb);
RESET_FAKE(FLEXCAN_ReadRxMb);
RESET_FAKE(FLEXCAN_GetMbStatusFlags);
RESET_FAKE(FLEXCAN_ClearMbStatusFlags);
RESET_FAKE(bsp_tick_get_ms);
FFF_RESET_HISTORY();
s_mb_flags_call_count = 0U;
s_tick_ms = 0U;
(void) memset(&s_injected_rx_frame, 0, sizeof(s_injected_rx_frame));
(void) memset(&g_stub_can2, 0, sizeof(g_stub_can2));
/* Деинициализировать модуль между тестами (сброс static-состояния). */
bsp_can_deinit();
RESET_FAKE(FLEXCAN_Deinit);
}
void tearDown(void)
{
}
/* ══════════════════════════════════════════════════════════════════════
* Init / Deinit
* ══════════════════════════════════════════════════════════════════════ */
void test_init_success(void)
{
bsp_status_t s = helper_init_default();
TEST_ASSERT_EQUAL(BSP_OK, s);
TEST_ASSERT_EQUAL(1, FLEXCAN_GetDefaultConfig_fake.call_count);
TEST_ASSERT_EQUAL(1, FLEXCAN_CalculateImprovedTimingValues_fake.call_count);
TEST_ASSERT_EQUAL(1, FLEXCAN_Init_fake.call_count);
/* MB0 = reserved (ERR005829), MB1 = TX */
TEST_ASSERT_EQUAL(1, FLEXCAN_SetTxMbConfig_fake.call_count);
}
void test_init_null_config(void)
{
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_init(NULL));
TEST_ASSERT_EQUAL(0, FLEXCAN_Init_fake.call_count);
}
void test_init_zero_bitrate(void)
{
bsp_can_config_t cfg = { .bitrate = 0U };
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_init(&cfg));
}
void test_init_bitrate_too_high(void)
{
bsp_can_config_t cfg = { .bitrate = 2000000U };
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_init(&cfg));
}
void test_init_timing_calc_fails(void)
{
FLEXCAN_CalculateImprovedTimingValues_fake.return_val = false;
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_init(&s_default_cfg));
TEST_ASSERT_EQUAL(0, FLEXCAN_Init_fake.call_count);
}
void test_init_reinit_calls_deinit(void)
{
helper_init_default();
/* Повторная инициализация — должен вызвать Deinit. */
helper_init_default();
TEST_ASSERT_EQUAL(1, FLEXCAN_Deinit_fake.call_count);
TEST_ASSERT_EQUAL(2, FLEXCAN_Init_fake.call_count);
}
void test_deinit_calls_sdk(void)
{
helper_init_default();
bsp_can_deinit();
TEST_ASSERT_EQUAL(1, FLEXCAN_Deinit_fake.call_count);
}
void test_deinit_without_init_is_noop(void)
{
bsp_can_deinit();
TEST_ASSERT_EQUAL(0, FLEXCAN_Deinit_fake.call_count);
}
/* ══════════════════════════════════════════════════════════════════════
* TX — bsp_can_send()
* ══════════════════════════════════════════════════════════════════════ */
void test_send_success(void)
{
helper_init_default();
FLEXCAN_WriteTxMb_fake.return_val = kStatus_Success;
/* Флаг TX complete — сразу готов. */
FLEXCAN_GetMbStatusFlags_fake.return_val = (uint64_t) 1U << 1U;
bsp_tick_get_ms_fake.return_val = 0U;
bsp_can_frame_t frame = {
.id = 0x123U,
.dlc = 8U,
.is_extended = false,
.is_remote = false,
.data = { 0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03, 0x04 },
};
TEST_ASSERT_EQUAL(BSP_OK, bsp_can_send(&frame, 100U));
TEST_ASSERT_EQUAL(1, FLEXCAN_WriteTxMb_fake.call_count);
TEST_ASSERT_EQUAL(1, FLEXCAN_ClearMbStatusFlags_fake.call_count);
}
void test_send_null_frame(void)
{
helper_init_default();
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_send(NULL, 100U));
}
void test_send_dlc_over_8(void)
{
helper_init_default();
bsp_can_frame_t frame = { .id = 0x123U, .dlc = 9U };
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_send(&frame, 100U));
}
void test_send_without_init(void)
{
bsp_can_frame_t frame = { .id = 0x123U, .dlc = 1U };
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_send(&frame, 100U));
}
void test_send_mb_busy(void)
{
helper_init_default();
FLEXCAN_WriteTxMb_fake.return_val = kStatus_Fail;
bsp_can_frame_t frame = { .id = 0x123U, .dlc = 1U };
TEST_ASSERT_EQUAL(BSP_ERR_BUSY, bsp_can_send(&frame, 100U));
}
void test_send_timeout(void)
{
helper_init_default();
FLEXCAN_WriteTxMb_fake.return_val = kStatus_Success;
FLEXCAN_GetMbStatusFlags_fake.return_val = 0U; /* Никогда не готов. */
bsp_tick_get_ms_fake.custom_fake = tick_advancing;
bsp_can_frame_t frame = { .id = 0x123U, .dlc = 1U };
TEST_ASSERT_EQUAL(BSP_ERR_TIMEOUT, bsp_can_send(&frame, 5U));
}
/* ══════════════════════════════════════════════════════════════════════
* Фильтрация — bsp_can_set_filter() / bsp_can_accept_all()
* ══════════════════════════════════════════════════════════════════════ */
void test_set_filter_std(void)
{
helper_init_default();
bsp_status_t s = bsp_can_set_filter(0U, 0x200U, 0x7FFU, false);
TEST_ASSERT_EQUAL(BSP_OK, s);
TEST_ASSERT_GREATER_THAN(0, FLEXCAN_SetRxMbConfig_fake.call_count);
TEST_ASSERT_GREATER_THAN(0, FLEXCAN_SetRxIndividualMask_fake.call_count);
}
void test_set_filter_ext(void)
{
helper_init_default();
bsp_status_t s = bsp_can_set_filter(0U, 0x1ABCDEF0U, 0x1FFFFFFFU, true);
TEST_ASSERT_EQUAL(BSP_OK, s);
}
void test_set_filter_index_out_of_range(void)
{
helper_init_default();
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_set_filter(BSP_CAN_FILTER_MAX, 0x123U, 0x7FFU, false));
}
void test_set_filter_without_init(void)
{
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_set_filter(0U, 0x123U, 0x7FFU, false));
}
void test_accept_all(void)
{
helper_init_default();
bsp_status_t s = bsp_can_accept_all();
TEST_ASSERT_EQUAL(BSP_OK, s);
/* Должен настроить минимум 2 RX MB (STD + EXT). */
TEST_ASSERT_GREATER_OR_EQUAL(2, FLEXCAN_SetRxMbConfig_fake.call_count);
}
void test_accept_all_clears_previous_filters(void)
{
helper_init_default();
/* Настроить 3 фильтра. */
bsp_can_set_filter(0U, 0x100U, 0x7FFU, false);
bsp_can_set_filter(1U, 0x200U, 0x7FFU, false);
bsp_can_set_filter(2U, 0x300U, 0x7FFU, false);
uint32_t calls_before = FLEXCAN_SetRxMbConfig_fake.call_count;
bsp_can_accept_all();
/* accept_all должен деактивировать предыдущие + настроить 2 новых. */
uint32_t calls_after = FLEXCAN_SetRxMbConfig_fake.call_count;
TEST_ASSERT_GREATER_THAN(calls_before, calls_after);
}
void test_accept_all_without_init(void)
{
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_accept_all());
}
/* ══════════════════════════════════════════════════════════════════════
* RX — bsp_can_receive()
* ══════════════════════════════════════════════════════════════════════ */
void test_receive_success(void)
{
helper_init_default();
bsp_can_accept_all();
/* Подготовить фрейм в SDK-формате. */
(void) memset(&s_injected_rx_frame, 0, sizeof(s_injected_rx_frame));
s_injected_rx_frame.id = FLEXCAN_ID_STD(0x456U);
s_injected_rx_frame.format = (uint8_t) kFLEXCAN_FrameFormatStandard;
s_injected_rx_frame.type = (uint8_t) kFLEXCAN_FrameTypeData;
s_injected_rx_frame.length = 4U;
s_injected_rx_frame.dataByte0 = 0xAAU;
s_injected_rx_frame.dataByte1 = 0xBBU;
s_injected_rx_frame.dataByte2 = 0xCCU;
s_injected_rx_frame.dataByte3 = 0xDDU;
FLEXCAN_ReadRxMb_fake.custom_fake = read_rx_mb_inject;
FLEXCAN_GetMbStatusFlags_fake.custom_fake = get_mb_flags_once;
s_mb_flags_call_count = 0U;
bsp_tick_get_ms_fake.return_val = 0U;
bsp_can_frame_t rx;
bsp_status_t s = bsp_can_receive(&rx, 100U);
TEST_ASSERT_EQUAL(BSP_OK, s);
TEST_ASSERT_EQUAL(0x456U, rx.id);
TEST_ASSERT_FALSE(rx.is_extended);
TEST_ASSERT_EQUAL(4U, rx.dlc);
TEST_ASSERT_EQUAL(0xAAU, rx.data[0]);
TEST_ASSERT_EQUAL(0xBBU, rx.data[1]);
TEST_ASSERT_EQUAL(0xCCU, rx.data[2]);
TEST_ASSERT_EQUAL(0xDDU, rx.data[3]);
}
void test_receive_ext_frame(void)
{
helper_init_default();
bsp_can_accept_all();
(void) memset(&s_injected_rx_frame, 0, sizeof(s_injected_rx_frame));
s_injected_rx_frame.id = FLEXCAN_ID_EXT(0x1ABCDEF0U);
s_injected_rx_frame.format = (uint8_t) kFLEXCAN_FrameFormatExtend;
s_injected_rx_frame.type = (uint8_t) kFLEXCAN_FrameTypeData;
s_injected_rx_frame.length = 2U;
FLEXCAN_ReadRxMb_fake.custom_fake = read_rx_mb_inject;
FLEXCAN_GetMbStatusFlags_fake.custom_fake = get_mb_flags_once;
s_mb_flags_call_count = 0U;
bsp_tick_get_ms_fake.return_val = 0U;
bsp_can_frame_t rx;
bsp_status_t s = bsp_can_receive(&rx, 100U);
TEST_ASSERT_EQUAL(BSP_OK, s);
TEST_ASSERT_EQUAL(0x1ABCDEF0U, rx.id);
TEST_ASSERT_TRUE(rx.is_extended);
}
void test_receive_timeout(void)
{
helper_init_default();
bsp_can_accept_all();
/* Ни один MB не готов. */
FLEXCAN_GetMbStatusFlags_fake.return_val = 0U;
bsp_tick_get_ms_fake.custom_fake = tick_advancing;
bsp_can_frame_t rx;
TEST_ASSERT_EQUAL(BSP_ERR_TIMEOUT, bsp_can_receive(&rx, 5U));
}
void test_receive_null_frame(void)
{
helper_init_default();
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_receive(NULL, 100U));
}
void test_receive_without_init(void)
{
bsp_can_frame_t rx;
TEST_ASSERT_EQUAL(BSP_ERR_PARAM, bsp_can_receive(&rx, 100U));
}
void test_receive_nonblocking_empty(void)
{
helper_init_default();
bsp_can_accept_all();
FLEXCAN_GetMbStatusFlags_fake.return_val = 0U;
bsp_tick_get_ms_fake.return_val = 0U;
bsp_can_frame_t rx;
/* timeout_ms = 0 → неблокирующий опрос. */
TEST_ASSERT_EQUAL(BSP_ERR_TIMEOUT, bsp_can_receive(&rx, 0U));
}
/* ══════════════════════════════════════════════════════════════════════
* Callback (заглушка)
* ══════════════════════════════════════════════════════════════════════ */
void test_register_callback_not_supported(void)
{
helper_init_default();
TEST_ASSERT_EQUAL(BSP_ERR_NOT_SUPPORTED, bsp_can_register_rx_callback(NULL, NULL));
}
/* ══════════════════════════════════════════════════════════════════════
* Frame conversion — проверка корректности STD/EXT ID encoding
* ══════════════════════════════════════════════════════════════════════ */
/**
* Захват фрейма, переданного в WriteTxMb, для проверки конвертации.
*/
static flexcan_frame_t s_captured_tx_frame;
static status_t capture_tx_frame(CAN_Type *p_base, uint8_t mb_idx, const flexcan_frame_t *p_frame)
{
(void) p_base;
(void) mb_idx;
s_captured_tx_frame = *p_frame;
return kStatus_Success;
}
void test_send_std_id_encoding(void)
{
helper_init_default();
FLEXCAN_WriteTxMb_fake.custom_fake = capture_tx_frame;
FLEXCAN_GetMbStatusFlags_fake.return_val = (uint64_t) 1U << 1U;
bsp_tick_get_ms_fake.return_val = 0U;
bsp_can_frame_t frame = {
.id = 0x7FFU,
.dlc = 0U,
.is_extended = false,
};
bsp_can_send(&frame, 100U);
TEST_ASSERT_EQUAL(FLEXCAN_ID_STD(0x7FFU), s_captured_tx_frame.id);
TEST_ASSERT_EQUAL((uint8_t) kFLEXCAN_FrameFormatStandard, s_captured_tx_frame.format);
}
void test_send_ext_id_encoding(void)
{
helper_init_default();
FLEXCAN_WriteTxMb_fake.custom_fake = capture_tx_frame;
FLEXCAN_GetMbStatusFlags_fake.return_val = (uint64_t) 1U << 1U;
bsp_tick_get_ms_fake.return_val = 0U;
bsp_can_frame_t frame = {
.id = 0x1FFFFFFFU,
.dlc = 0U,
.is_extended = true,
};
bsp_can_send(&frame, 100U);
TEST_ASSERT_EQUAL(FLEXCAN_ID_EXT(0x1FFFFFFFU), s_captured_tx_frame.id);
TEST_ASSERT_EQUAL((uint8_t) kFLEXCAN_FrameFormatExtend, s_captured_tx_frame.format);
}
void test_send_data_byte_order(void)
{
helper_init_default();
FLEXCAN_WriteTxMb_fake.custom_fake = capture_tx_frame;
FLEXCAN_GetMbStatusFlags_fake.return_val = (uint64_t) 1U << 1U;
bsp_tick_get_ms_fake.return_val = 0U;
bsp_can_frame_t frame = {
.id = 0x100U,
.dlc = 8U,
.data = { 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88 },
};
bsp_can_send(&frame, 100U);
TEST_ASSERT_EQUAL_HEX8(0x11, s_captured_tx_frame.dataByte0);
TEST_ASSERT_EQUAL_HEX8(0x22, s_captured_tx_frame.dataByte1);
TEST_ASSERT_EQUAL_HEX8(0x33, s_captured_tx_frame.dataByte2);
TEST_ASSERT_EQUAL_HEX8(0x44, s_captured_tx_frame.dataByte3);
TEST_ASSERT_EQUAL_HEX8(0x55, s_captured_tx_frame.dataByte4);
TEST_ASSERT_EQUAL_HEX8(0x66, s_captured_tx_frame.dataByte5);
TEST_ASSERT_EQUAL_HEX8(0x77, s_captured_tx_frame.dataByte6);
TEST_ASSERT_EQUAL_HEX8(0x88, s_captured_tx_frame.dataByte7);
}
/* ══════════════════════════════════════════════════════════════════════
* main
* ══════════════════════════════════════════════════════════════════════ */
int main(void)
{
UNITY_BEGIN();
/* Init / Deinit */
RUN_TEST(test_init_success);
RUN_TEST(test_init_null_config);
RUN_TEST(test_init_zero_bitrate);
RUN_TEST(test_init_bitrate_too_high);
RUN_TEST(test_init_timing_calc_fails);
RUN_TEST(test_init_reinit_calls_deinit);
RUN_TEST(test_deinit_calls_sdk);
RUN_TEST(test_deinit_without_init_is_noop);
/* TX */
RUN_TEST(test_send_success);
RUN_TEST(test_send_null_frame);
RUN_TEST(test_send_dlc_over_8);
RUN_TEST(test_send_without_init);
RUN_TEST(test_send_mb_busy);
RUN_TEST(test_send_timeout);
/* Фильтрация */
RUN_TEST(test_set_filter_std);
RUN_TEST(test_set_filter_ext);
RUN_TEST(test_set_filter_index_out_of_range);
RUN_TEST(test_set_filter_without_init);
RUN_TEST(test_accept_all);
RUN_TEST(test_accept_all_clears_previous_filters);
RUN_TEST(test_accept_all_without_init);
/* RX */
RUN_TEST(test_receive_success);
RUN_TEST(test_receive_ext_frame);
RUN_TEST(test_receive_timeout);
RUN_TEST(test_receive_null_frame);
RUN_TEST(test_receive_without_init);
RUN_TEST(test_receive_nonblocking_empty);
/* Callback */
RUN_TEST(test_register_callback_not_supported);
/* Frame conversion */
RUN_TEST(test_send_std_id_encoding);
RUN_TEST(test_send_ext_id_encoding);
RUN_TEST(test_send_data_byte_order);
return UNITY_END();
}

10
tests/host/mocks/clock_config.h
View file

@ -0,0 +1,10 @@
/**
* @file clock_config.h
* @brief Stub для host-тестов — только константы используемые в bsp_can.
*/
#ifndef CLOCK_CONFIG_H_
#define CLOCK_CONFIG_H_
#define BOARD_BOOTCLOCKRUN_CAN_CLK_ROOT 80000000UL
#endif /* CLOCK_CONFIG_H_ */

25
tests/target/hil_can/CMakeLists.txt
View file

@ -0,0 +1,25 @@
# =============================================================================
# tests/target/hil_can/CMakeLists.txt
# =============================================================================
set(TARGET_NAME test_hil_can)
add_executable(${TARGET_NAME} main.c ${BSP_GENERATED}/clock_config.c
${BSP_STARTUP_FILE} ${BSP_SYSCALLS_FILE})
target_link_options(
${TARGET_NAME}
PRIVATE
-T${CMAKE_SOURCE_DIR}/cmake/linker/MIMXRT1052xxxxx_ram.ld
-Wl,--gc-sections
-Wl,--print-memory-usage
-Wl,-Map=${CMAKE_CURRENT_BINARY_DIR}/${TARGET_NAME}.map)
target_link_libraries(${TARGET_NAME} PRIVATE bsp_boot_ram bsp_board bsp_led
bsp_tick bsp_uart_host bsp_can)
add_custom_command(
TARGET ${TARGET_NAME}
POST_BUILD
COMMAND ${CMAKE_SIZE} $<TARGET_FILE:${TARGET_NAME}>
COMMENT "Size: ${TARGET_NAME}")

332
tests/target/hil_can/main.c
View file

@ -0,0 +1,332 @@
/**
* @file tests/target/hil_can/main.c
* @brief HIL target — CLI для тестирования bsp_can.
*
* Протокол: текстовые команды через LPUART1 (MCU-Link VCOM), \r\n-terminated.
*
* Команды:
* PING -> PONG
* CAN_SEND <id> <ext> <dlc> <b0>..<b7>
* -> OK / ERR_TIMEOUT / ERR_BUSY / ERR_PARAM
* CAN_RECV <timeout_ms> -> <id> <ext> <dlc> <b0>..<b7> / TIMEOUT
* CAN_FILTER <idx> <id> <mask> <ext>
* -> OK / ERR_PARAM
* CAN_ACCEPT_ALL -> OK
* CAN_RX_EVENTS -> <число>
* CAN_LAST_RX -> <id> <ext> <dlc> <b0>..<b7>
* CAN_RESET_EVENTS -> OK
*
* Формат числовых параметров: десятичный (для простоты парсинга).
* ext: 0 = STD, 1 = EXT.
*
* Стенд:
* M5StampPLC (SIT1044 трансивер) ↔ SN65HVD230D таргета
* Общая CAN-шина 125 kbit/s (CAN_BITRATE в прошивке = M5 can_baud в agent.py).
*
* ВАЖНО:
* bsp_can_init() использует disableSelfReception=true — таргет не слышит
* собственные фреймы. Для round-trip теста «таргет TX → таргет RX» нужен
* второй узел (M5) на шине, который ретранслирует фрейм обратно.
* Для теста «M5 TX → таргет RX» второй узел не нужен.
*/
#include "board.h"
#include "bsp/can.h"
#include "bsp/led.h"
#include "bsp/tick.h"
#include "bsp/uart_host.h"
#include <stdio.h>
#include <string.h>
/* -------------------------------------------------------------------------- */
#define CLI_BAUD_RATE 115200U
#define CLI_LINE_MAX 256U
#define CLI_RX_TIMEOUT 50U /* мс */
#define CAN_BITRATE 125000U
#define CAN_TX_TIMEOUT 100U /* мс — таймаут отправки одного фрейма */
/* -------------------------------------------------------------------------- */
/* Event tracking */
/* -------------------------------------------------------------------------- */
static volatile uint32_t s_rx_event_count;
static volatile bsp_can_frame_t s_last_rx_frame;
/* -------------------------------------------------------------------------- */
/* CLI helpers */
/* -------------------------------------------------------------------------- */
static size_t cli_read_line(uint8_t *p_buf, size_t max_len)
{
size_t pos = 0U;
while (pos < (max_len - 1U))
{
int32_t byte = bsp_uart_host_read_byte(CLI_RX_TIMEOUT);
if (byte < 0)
{
break;
}
if ((char) byte == '\r')
{
continue;
}
if ((char) byte == '\n')
{
break;
}
p_buf[pos++] = (uint8_t) byte;
}
p_buf[pos] = '\0';
return pos;
}
/** Сериализовать фрейм в строку "<id> <ext> <dlc> <b0> ... <bN>". */
static void frame_to_str(const bsp_can_frame_t *p_frame, char *p_buf, size_t buf_len)
{
int written = snprintf(p_buf, buf_len, "%lu %u %u", (unsigned long) p_frame->id,
(unsigned) p_frame->is_extended, (unsigned) p_frame->dlc);
for (uint8_t i = 0U; i < p_frame->dlc && i < BSP_CAN_DATA_MAX_LEN; i++)
{
int n = snprintf(p_buf + written, buf_len - (size_t) written, " %u",
(unsigned) p_frame->data[i]);
if (n > 0)
{
written += n;
}
}
}
/* -------------------------------------------------------------------------- */
/* CLI command handlers */
/* -------------------------------------------------------------------------- */
/**
* CAN_SEND <id> <ext> <dlc> [<b0> .. <b7>]
*
* Пример: CAN_SEND 123 0 3 0xDE 0xAD 0xBE
* CAN_SEND 536870911 1 2 0xAA 0xBB
*/
static void cmd_can_send(const char *p_args)
{
unsigned long id = 0U;
unsigned ext = 0U;
unsigned dlc = 0U;
int parsed = sscanf(p_args, "%lu %u %u", &id, &ext, &dlc);
if (parsed != 3 || dlc > BSP_CAN_DATA_MAX_LEN)
{
bsp_uart_host_write_str("ERR_PARAM\r\n");
return;
}
bsp_can_frame_t frame;
(void) memset(&frame, 0, sizeof(frame));
frame.id = (uint32_t) id;
frame.is_extended = (ext != 0U);
frame.dlc = (uint8_t) dlc;
/* Парсим байты данных после трёх обязательных аргументов. */
const char *p = p_args;
/* Пропустить id, ext, dlc. */
for (int i = 0; i < 3; i++)
{
while (*p == ' ')
p++;
while (*p != ' ' && *p != '\0')
p++;
}
for (uint8_t i = 0U; i < dlc; i++)
{
unsigned byte_val = 0U;
if (sscanf(p, " %u", &byte_val) != 1)
{
break;
}
frame.data[i] = (uint8_t) byte_val;
/* Сдвинуться на следующий аргумент. */
while (*p == ' ')
p++;
while (*p != ' ' && *p != '\0')
p++;
}
bsp_status_t status = bsp_can_send(&frame, CAN_TX_TIMEOUT);
switch (status)
{
case BSP_OK:
bsp_uart_host_write_str("OK\r\n");
break;
case BSP_ERR_TIMEOUT:
bsp_uart_host_write_str("ERR_TIMEOUT\r\n");
break;
case BSP_ERR_BUSY:
bsp_uart_host_write_str("ERR_BUSY\r\n");
break;
default:
bsp_uart_host_write_str("ERR_PARAM\r\n");
break;
}
}
/**
* CAN_RECV <timeout_ms>
*
* Блокируется до приёма фрейма или таймаута.
* Ответ: "<id> <ext> <dlc> <b0> ... <bN>" или "TIMEOUT".
*/
static void cmd_can_recv(const char *p_args)
{
unsigned timeout_ms = 0U;
if (sscanf(p_args, "%u", &timeout_ms) != 1)
{
bsp_uart_host_write_str("ERR_PARAM\r\n");
return;
}
bsp_can_frame_t frame;
bsp_status_t status = bsp_can_receive(&frame, timeout_ms);
if (status != BSP_OK)
{
bsp_uart_host_write_str("TIMEOUT\r\n");
return;
}
/* Обновить счётчик событий (polling-режим — обновляем здесь). */
s_rx_event_count++;
s_last_rx_frame = frame;
char resp[128];
frame_to_str(&frame, resp, sizeof(resp) - 2U);
(void) strncat(resp, "\r\n", sizeof(resp) - strlen(resp) - 1U);
bsp_uart_host_write_str(resp);
}
/**
* CAN_FILTER <idx> <id> <mask> <ext>
*
* Пример: CAN_FILTER 0 100 2047 0 (STD ID=100, маска=все биты, STD)
* CAN_FILTER 1 536870912 536870911 1 (EXT)
*/
static void cmd_can_filter(const char *p_args)
{
unsigned idx = 0U;
unsigned long id = 0U;
unsigned long mask = 0U;
unsigned ext = 0U;
if (sscanf(p_args, "%u %lu %lu %u", &idx, &id, &mask, &ext) != 4)
{
bsp_uart_host_write_str("ERR_PARAM\r\n");
return;
}
bsp_status_t status =
bsp_can_set_filter((uint8_t) idx, (uint32_t) id, (uint32_t) mask, (ext != 0U));
bsp_uart_host_write_str(status == BSP_OK ? "OK\r\n" : "ERR_PARAM\r\n");
}
/* -------------------------------------------------------------------------- */
/* Main CLI dispatcher */
/* -------------------------------------------------------------------------- */
static void cli_process_line(const char *p_line)
{
char resp[128];
if (strncmp(p_line, "PING", 4U) == 0)
{
bsp_uart_host_write_str("PONG\r\n");
}
else if (strncmp(p_line, "CAN_SEND ", 9U) == 0)
{
cmd_can_send(p_line + 9U);
}
else if (strncmp(p_line, "CAN_RECV ", 9U) == 0)
{
cmd_can_recv(p_line + 9U);
}
else if (strncmp(p_line, "CAN_FILTER ", 11U) == 0)
{
cmd_can_filter(p_line + 11U);
}
else if (strncmp(p_line, "CAN_ACCEPT_ALL", 14U) == 0)
{
bsp_status_t s = bsp_can_accept_all();
bsp_uart_host_write_str(s == BSP_OK ? "OK\r\n" : "ERR_PARAM\r\n");
}
else if (strncmp(p_line, "CAN_RX_EVENTS", 13U) == 0)
{
snprintf(resp, sizeof(resp), "%lu\r\n", (unsigned long) s_rx_event_count);
bsp_uart_host_write_str(resp);
}
else if (strncmp(p_line, "CAN_LAST_RX", 11U) == 0)
{
frame_to_str((const bsp_can_frame_t *) &s_last_rx_frame, resp, sizeof(resp) - 2U);
(void) strncat(resp, "\r\n", sizeof(resp) - strlen(resp) - 1U);
bsp_uart_host_write_str(resp);
}
else if (strncmp(p_line, "CAN_RESET_EVENTS", 16U) == 0)
{
s_rx_event_count = 0U;
(void) memset((void *) &s_last_rx_frame, 0, sizeof(s_last_rx_frame));
bsp_uart_host_write_str("OK\r\n");
}
else if (p_line[0] != '\0')
{
bsp_uart_host_write_str("ERR_UNKNOWN\r\n");
}
}
/* -------------------------------------------------------------------------- */
/* main */
/* -------------------------------------------------------------------------- */
int main(void)
{
board_hw_init();
bsp_tick_init();
bsp_led_init();
bsp_uart_host_init(CLI_BAUD_RATE);
/* --- CAN init: 125 kbit/s (совпадает с can_baud агента M5) --- */
bsp_can_config_t can_cfg = { .bitrate = CAN_BITRATE };
(void) bsp_can_init(&can_cfg);
/* По умолчанию принимаем все фреймы. */
(void) bsp_can_accept_all();
bsp_led_on(LED_HEARTBEAT);
/* Шлём READY пока хост не подключится. */
while (bsp_uart_host_rx_available() == 0U)
{
bsp_uart_host_write_str("READY\r\n");
bsp_led_toggle(LED_APP);
bsp_delay(200U);
}
bsp_led_off(LED_APP);
static uint8_t s_line_buf[CLI_LINE_MAX];
for (;;)
{
size_t len = cli_read_line(s_line_buf, sizeof(s_line_buf));
if (len > 0U)
{
cli_process_line((const char *) s_line_buf);
}
}
return 0;
}

0
tools/hil/test_uart.py → tools/hil/01_test_uart.py
View file

0
tools/hil/test_opto.py → tools/hil/02_test_opto.py
View file

395
tools/hil/03_test_can.py Normal file
View file

@ -0,0 +1,395 @@
"""
test_can.py — HIL тест bsp_can через M5StampPLC.
Стенд:
M5StampPLC (SIT1044 трансивер) и таргет (SN65HVD230D) подключены
к общей двухузловой CAN-шине. Скорость: 125 kbit/s.
Топология и направления трафика:
M5 TX → шина → таргет RX (команды: m5.can_send())
таргет TX → шина → M5 RX (команды: uart_cmd CAN_SEND, m5.can_recv())
Важно: bsp_can инициализирован с disableSelfReception=true.
Таргет не слышит собственные фреймы. Для проверки «таргет TX → таргет RX»
нужен M5 как ретранслятор (тест test_target_tx_m5_rx_target_rx).
Цепочка фикстур (scope=module):
m5 (питание ON)
└── loaded_hil_can (грузит ELF через pyOCD)
└── uart_can (открывает VCOM, ждёт READY)
└── _setup (autouse, function scope) → self.ser / self.m5
Запуск:
just host::hil-can
uv run pytest test_can.py -v
uv run pytest test_can.py -v --no-load --m5-port /dev/ttyACM1
"""
import time
import pytest
from conftest import uart_cmd
# ---------------------------------------------------------------------------
# Временны́е константы
# ---------------------------------------------------------------------------
# При 125 kbit/s один фрейм занимает ~0.1 мс.
# Добавляем запас на задержку USB CDC (M5 ↔ хост) и polling в прошивке.
CAN_SETTLE_S = 0.05 # ждать после отправки перед чтением
# Таймаут CAN_RECV на таргете (мс) — передаётся в команду.
# Должен быть достаточным для round-trip через шину + USB CDC M5.
CAN_RECV_TIMEOUT_MS = 300
# Таймаут CAN_RECV при тесте «никто не шлёт» — чтобы тест не завис.
CAN_RECV_EMPTY_MS = 100
# ---------------------------------------------------------------------------
# Вспомогательные функции
# ---------------------------------------------------------------------------
def can_send_m5(m5, can_id: int, data: list[int], ext: bool = False) -> None:
"""M5 отправляет CAN-фрейм на шину."""
m5.can_send(can_id, data, ext=ext)
def can_recv_m5(m5, timeout_ms: int = 500) -> dict:
"""
M5 ждёт CAN-фрейм с шины.
Возвращает dict с ключами id, ext, data.
Выбрасывает TimeoutError если фрейм не пришёл.
"""
return m5.can_recv(timeout_ms=timeout_ms)
def can_send_target(ser, can_id: int, data: list[int], ext: bool = False) -> str:
"""
Таргет отправляет CAN-фрейм.
Возвращает ответ прошивки: 'OK', 'ERR_TIMEOUT', 'ERR_BUSY', 'ERR_PARAM'.
"""
dlc = len(data)
data_str = " ".join(str(b) for b in data)
ext_flag = 1 if ext else 0
cmd = f"CAN_SEND {can_id} {ext_flag} {dlc}"
if dlc > 0:
cmd += f" {data_str}"
return uart_cmd(ser, cmd)
def can_recv_target(ser, timeout_ms: int = CAN_RECV_TIMEOUT_MS) -> dict | None:
"""
Таргет ждёт CAN-фрейм (polling).
Возвращает dict {id, ext, dlc, data} или None при TIMEOUT.
"""
resp = uart_cmd(ser, f"CAN_RECV {timeout_ms}")
if resp == "TIMEOUT":
return None
parts = resp.split()
if len(parts) < 3:
raise ValueError(f"Неожиданный ответ CAN_RECV: {resp!r}")
can_id = int(parts[0])
ext = bool(int(parts[1]))
dlc = int(parts[2])
data = [int(b) for b in parts[3:3 + dlc]]
return {"id": can_id, "ext": ext, "dlc": dlc, "data": data}
def set_filter_target(ser, idx: int, can_id: int, mask: int, ext: bool = False) -> str:
"""Установить RX-фильтр на таргете. Возвращает 'OK' или 'ERR_PARAM'."""
ext_flag = 1 if ext else 0
return uart_cmd(ser, f"CAN_FILTER {idx} {can_id} {mask} {ext_flag}")
def reset_events(ser) -> None:
"""Сбросить счётчик RX-событий на таргете."""
assert uart_cmd(ser, "CAN_RESET_EVENTS") == "OK"
# ---------------------------------------------------------------------------
# Проверка каналов связи
# ---------------------------------------------------------------------------
class TestCanConnectivity:
"""Базовая проверка: таргет и M5 отвечают."""
@pytest.fixture(autouse=True)
def _setup(self, uart_can, m5):
self.ser = uart_can
self.m5 = m5
def test_target_ping(self):
"""PING → PONG: UART-канал host↔target работает."""
assert uart_cmd(self.ser, "PING") == "PONG"
def test_m5_ping(self):
"""M5 agent отвечает на ping."""
self.m5.ping()
def test_m5_can_available(self):
"""M5 сообщает что CAN инициализирован (can_ok=true в info)."""
info = self.m5.info()
assert info.get("can_ok"), "M5 CAN не инициализирован — проверьте трансивер и agent.py"
# ---------------------------------------------------------------------------
# M5 → таргет
# ---------------------------------------------------------------------------
class TestCanM5ToTarget:
"""M5 отправляет фрейм — таргет принимает."""
@pytest.fixture(autouse=True)
def _setup(self, uart_can, m5):
self.ser = uart_can
self.m5 = m5
uart_cmd(self.ser, "CAN_ACCEPT_ALL")
reset_events(self.ser)
def test_std_frame_received(self):
"""M5 TX STD → таргет принимает с верным ID и данными."""
payload = [0x11, 0x22, 0x33]
can_send_m5(self.m5, 0x123, payload)
frame = can_recv_target(self.ser)
assert frame is not None, "Таргет не принял фрейм от M5"
assert frame["id"] == 0x123, f"Неверный ID: {frame['id']:#x}"
assert frame["ext"] is False, "Ожидали STD-фрейм"
assert frame["dlc"] == 3, f"Неверный DLC: {frame['dlc']}"
assert frame["data"] == payload, f"Неверные данные: {frame['data']}"
def test_ext_frame_received(self):
"""M5 TX EXT → таргет принимает с верным 29-bit ID."""
payload = [0xAA, 0xBB]
can_send_m5(self.m5, 0x1ABCDEF, payload, ext=True)
frame = can_recv_target(self.ser)
assert frame is not None, "Таргет не принял EXT-фрейм"
assert frame["id"] == 0x1ABCDEF, f"Неверный EXT ID: {frame['id']:#x}"
assert frame["ext"] is True, "Ожидали EXT-фрейм"
assert frame["data"] == payload, f"Неверные данные: {frame['data']}"
def test_max_dlc_frame(self):
"""M5 TX 8-байтовый фрейм — таргет принимает все 8 байт корректно."""
payload = [0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08]
can_send_m5(self.m5, 0x7FF, payload)
frame = can_recv_target(self.ser)
assert frame is not None
assert frame["dlc"] == 8
assert frame["data"] == payload
def test_zero_dlc_frame(self):
"""M5 TX фрейм с DLC=0 — таргет принимает без данных."""
can_send_m5(self.m5, 0x001, [])
frame = can_recv_target(self.ser)
assert frame is not None
assert frame["dlc"] == 0
assert frame["data"] == []
def test_rx_event_counter_increments(self):
"""Каждый принятый фрейм увеличивает счётчик событий."""
for _ in range(3):
can_send_m5(self.m5, 0x100, [0xFF])
can_recv_target(self.ser) # дождаться приёма
count = int(uart_cmd(self.ser, "CAN_RX_EVENTS"))
assert count >= 3, f"Ожидали >= 3 события, получили {count}"
# ---------------------------------------------------------------------------
# Таргет → M5
# ---------------------------------------------------------------------------
class TestCanTargetToM5:
"""Таргет отправляет фрейм — M5 принимает."""
@pytest.fixture(autouse=True)
def _setup(self, uart_can, m5):
self.ser = uart_can
self.m5 = m5
def test_std_frame_sent(self):
"""Таргет TX STD → M5 принимает с верным ID и данными."""
payload = [0xDE, 0xAD, 0xBE, 0xEF]
assert can_send_target(self.ser, 0x456, payload) == "OK"
frame = can_recv_m5(self.m5)
assert frame["id"] == 0x456, f"Неверный ID: {frame['id']:#x}"
assert frame["ext"] is False, "Ожидали STD-фрейм"
assert frame["data"] == payload, f"Неверные данные: {frame['data']}"
def test_ext_frame_sent(self):
"""Таргет TX EXT → M5 принимает с верным 29-bit ID."""
payload = [0x01, 0x02]
assert can_send_target(self.ser, 0x1FFFFFF, payload, ext=True) == "OK"
frame = can_recv_m5(self.m5)
assert frame["id"] == 0x1FFFFFF, f"Неверный EXT ID: {frame['id']:#x}"
assert frame["ext"] is True, "Ожидали EXT-фрейм"
def test_max_std_id(self):
"""Максимальный достижимый STD ID через M5 recv — проверяем 0x7FE."""
# 0x7FF вызывает баг в TWAI-биндинге M5 (known limitation)
assert can_send_target(self.ser, 0x7FE, [0x00]) == "OK"
frame = can_recv_m5(self.m5)
assert frame["id"] == 0x7FE
def test_max_std_id(self):
"""Максимальный STD ID (0x7FF) передаётся корректно."""
assert can_send_target(self.ser, 0x7FF, [0x00]) == "OK"
frame = can_recv_m5(self.m5)
assert frame["id"] == 0x7FF
def test_max_ext_id(self):
"""Максимальный EXT ID (0x1FFFFFFF) передаётся корректно."""
assert can_send_target(self.ser, 0x1FFFFFFF, [0x00], ext=True) == "OK"
frame = can_recv_m5(self.m5)
assert frame["id"] == 0x1FFFFFFF
def test_data_integrity_all_bytes(self):
"""Все 8 байт данных передаются без искажений."""
payload = [0x00, 0xFF, 0x55, 0xAA, 0x0F, 0xF0, 0x01, 0xFE]
assert can_send_target(self.ser, 0x300, payload) == "OK"
frame = can_recv_m5(self.m5)
assert frame["data"] == payload, \
f"Данные искажены: ожидали {payload}, получили {frame['data']}"
# ---------------------------------------------------------------------------
# Фильтрация
# ---------------------------------------------------------------------------
class TestCanFiltering:
"""
Проверка фильтрации по ID: только нужные фреймы проходят,
остальные блокируются.
"""
@pytest.fixture(autouse=True)
def _setup(self, uart_can, m5):
self.ser = uart_can
self.m5 = m5
reset_events(self.ser)
def test_accept_all_receives_any_id(self):
"""После CAN_ACCEPT_ALL таргет принимает фреймы с любым ID."""
uart_cmd(self.ser, "CAN_ACCEPT_ALL")
can_send_m5(self.m5, 0x001, [0x01])
frame = can_recv_target(self.ser)
assert frame is not None, "Фрейм не принят после CAN_ACCEPT_ALL"
can_send_m5(self.m5, 0x7FF, [0x02])
frame = can_recv_target(self.ser)
assert frame is not None, "Второй фрейм не принят после CAN_ACCEPT_ALL"
def test_filter_exact_id_passes(self):
"""set_filter с маской 0x7FF пропускает только точный ID."""
target_id = 0x123
# Маска 0x7FF = все 11 бит проверяются → точное совпадение.
assert set_filter_target(self.ser, 0, target_id, 0x7FF) == "OK"
can_send_m5(self.m5, target_id, [0xAA])
frame = can_recv_target(self.ser)
assert frame is not None, "Фрейм с нужным ID не принят"
assert frame["id"] == target_id, f"Получен ID {frame['id']:#x}, ожидали {target_id:#x}"
def test_filter_wrong_id_blocked(self):
"""set_filter с точной маской блокирует другой ID."""
target_id = 0x123
wrong_id = 0x456
assert set_filter_target(self.ser, 0, target_id, 0x7FF) == "OK"
can_send_m5(self.m5, wrong_id, [0xBB])
frame = can_recv_target(self.ser, timeout_ms=CAN_RECV_EMPTY_MS)
assert frame is None, \
f"Фрейм с ID {wrong_id:#x} прошёл фильтр, хотя не должен был"
def test_filter_mask_passes_group(self):
"""Маска 0x7F0 пропускает группу ID с одинаковыми старшими битами."""
base_id = 0x120
mask = 0x7F0 # проверять биты 11..4, биты 3..0 — игнорировать
assert set_filter_target(self.ser, 0, base_id, mask) == "OK"
# 0x123 & ~0x7F0 = различается только в младших 4 битах → должен пройти
can_send_m5(self.m5, 0x123, [0x01])
frame = can_recv_target(self.ser)
assert frame is not None, "0x123 должен пройти маску 0x7F0 для базы 0x120"
# 0x200 — другая группа → должен блокироваться
# Сбрасываем входной буфер таргета перед проверкой блокировки
can_send_m5(self.m5, 0x200, [0x02])
frame = can_recv_target(self.ser, timeout_ms=CAN_RECV_EMPTY_MS)
assert frame is None, "0x200 не должен пройти маску для базы 0x120"
def test_ext_filter_exact_id(self):
"""EXT-фильтр пропускает точный 29-bit ID."""
target_id = 0x1ABCDEF
assert set_filter_target(self.ser, 0, target_id, 0x1FFFFFFF, ext=True) == "OK"
can_send_m5(self.m5, target_id, [0x42], ext=True)
frame = can_recv_target(self.ser)
assert frame is not None, "EXT-фрейм с нужным ID не принят"
assert frame["id"] == target_id
assert frame["ext"] is True
def test_accept_all_after_filter(self):
"""CAN_ACCEPT_ALL после set_filter снова принимает все фреймы."""
# Сначала ставим узкий фильтр
assert set_filter_target(self.ser, 0, 0x100, 0x7FF) == "OK"
# Убеждаемся что фрейм с другим ID не проходит
can_send_m5(self.m5, 0x200, [0x01])
assert can_recv_target(self.ser, timeout_ms=CAN_RECV_EMPTY_MS) is None
# Снимаем фильтр
uart_cmd(self.ser, "CAN_ACCEPT_ALL")
# Теперь должен пройти любой ID
can_send_m5(self.m5, 0x200, [0x02])
frame = can_recv_target(self.ser)
assert frame is not None, "Фрейм не принят после CAN_ACCEPT_ALL"
# ---------------------------------------------------------------------------
# STD и EXT раздельно
# ---------------------------------------------------------------------------
class TestCanFrameTypes:
"""STD и EXT фреймы не перепутываются — is_extended корректен."""
@pytest.fixture(autouse=True)
def _setup(self, uart_can, m5):
self.ser = uart_can
self.m5 = m5
uart_cmd(self.ser, "CAN_ACCEPT_ALL")
def test_std_frame_is_not_ext(self):
"""Принятый STD-фрейм имеет ext=False."""
can_send_m5(self.m5, 0x1FF, [0x01])
frame = can_recv_target(self.ser)
assert frame is not None
assert frame["ext"] is False, "STD-фрейм ошибочно помечен как EXT"
def test_ext_frame_is_not_std(self):
"""Принятый EXT-фрейм имеет ext=True."""
can_send_m5(self.m5, 0x1FF, [0x01], ext=True)
frame = can_recv_target(self.ser)
assert frame is not None
assert frame["ext"] is True, "EXT-фрейм ошибочно помечен как STD"
def test_std_id_range_boundary(self):
"""Граничные STD ID: 0x000 и 0x7FF."""
for can_id in [0x000, 0x7FF]:
can_send_m5(self.m5, can_id, [0xBB])
frame = can_recv_target(self.ser)
assert frame is not None, f"ID {can_id:#x} не принят"
assert frame["id"] == can_id, f"ID {frame['id']:#x} ≠ {can_id:#x}"
assert frame["ext"] is False
def test_ext_id_range_boundary(self):
"""Граничные EXT ID: 0x000 и 0x1FFFFFFF."""
for can_id in [0x000, 0x1FFFFFFF]:
can_send_m5(self.m5, can_id, [0xCC], ext=True)
frame = can_recv_target(self.ser)
assert frame is not None, f"EXT ID {can_id:#x} не принят"
assert frame["id"] == can_id, f"EXT ID {frame['id']:#x} ≠ {can_id:#x}"
assert frame["ext"] is True

44
tools/hil/conftest.py
View file

@ -21,7 +21,7 @@ from pyocd_utils import flexram_init, load_elf, open_target, run_from_vectors
log = logging.getLogger(__name__)
# Задержка после включения питания таргета (мс стабилизации + POR)
_POWER_ON_SETTLE_S = 1.0
_POWER_ON_SETTLE_S = 1.5
# ---------------------------------------------------------------------------
@ -138,6 +138,23 @@ class M5Agent:
def opto_all_off(self) -> None:
self.cmd("opto_all_off")
def info(self) -> dict:
"""Запросить информацию об агенте (включая can_ok)."""
return self.cmd("info")
def can_send(self, can_id: int, data: list, ext: bool = False) -> None:
"""Отправить CAN-фрейм с шины M5."""
self.cmd("can_send", id=can_id, data=list(data), ext=ext)
def can_recv(self, timeout_ms: int = 500) -> dict:
"""
Принять CAN-фрейм на M5.
Возвращает dict {id, ext, data}.
Выбрасывает TimeoutError если фрейм не пришёл.
"""
resp = self.cmd("can_recv", timeout_ms=timeout_ms)
# cmd() уже выбрасывает RuntimeError при ok=false (включая timeout от агента)
return resp
# ---------------------------------------------------------------------------
@ -203,7 +220,7 @@ def m5(request: pytest.FixtureRequest) -> Generator[M5Agent, None, None]:
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def loaded_host_uart(request: pytest.FixtureRequest) -> None:
def loaded_host_uart(request: pytest.FixtureRequest, m5: M5Agent) -> None:
_load_elf(
request,
Path(cfg.BUILD_DIR) / "tests/target/host_uart/test_host_uart.elf",
@ -226,6 +243,19 @@ def loaded_hil_opto(request: pytest.FixtureRequest, m5: M5Agent) -> None:
Path(cfg.BUILD_DIR) / "tests/target/hil_opto/test_hil_opto.elf",
)
@pytest.fixture(scope="module")
def loaded_hil_can(request: pytest.FixtureRequest, m5: M5Agent) -> None:
"""
Загрузить test_hil_can.elf.
Явная зависимость от m5 гарантирует порядок:
1. m5 создаётся первым → питание таргета включено
2. только потом pyOCD подключается и грузит ELF
"""
_load_elf(
request,
Path(cfg.BUILD_DIR) / "tests/target/hil_can/test_hil_can.elf",
)
# ---------------------------------------------------------------------------
# Фикстуры UART
@ -239,7 +269,6 @@ def uart(
yield ser
ser.close()
@pytest.fixture(scope="module")
def uart_opto(
request: pytest.FixtureRequest,
@ -249,7 +278,14 @@ def uart_opto(
yield ser
ser.close()
@pytest.fixture(scope="module")
def uart_can(
request: pytest.FixtureRequest,
loaded_hil_can,
) -> Generator[serial.Serial, None, None]:
ser = _open_uart_and_wait_ready(request)
yield ser
ser.close()
# ---------------------------------------------------------------------------
# Утилита для тестов
# ---------------------------------------------------------------------------

59
tools/hil/m5/agent.py
View file

@ -236,7 +236,7 @@ def _dispatch(cmd: dict) -> dict:
if c == "info":
return {
"ok": True,
"fw": "agent/1.1",
"fw": "agent/1.2",
"python": sys.version[:40],
"can_ok": _can_ok,
"aw_ok": _aw is not None,
@ -307,7 +307,6 @@ def _dispatch(cmd: dict) -> dict:
if not _can_ok:
return {"ok": False, "err": "CAN not available"}
ext = bool(cmd.get("ext", False))
data = bytes(cmd.get("data", []))
_can.send(list(cmd.get("data", [])), int(cmd["id"]), extframe=ext)
return {"ok": True}
@ -318,59 +317,9 @@ def _dispatch(cmd: dict) -> dict:
msg = _can.recv(timeout=timeout_ms)
if msg is None:
return {"ok": False, "err": "timeout"}
frame_id, _, ext, data = msg
return {"ok": True, "id": frame_id, "ext": ext, "data": list(data)}
if c == "can_loopback":
if not _can_ok:
return {"ok": False, "err": "CAN not available"}
try:
import CAN as _CAN_MOD
# Временно переинициализируем в LOOPBACK режиме
_can.deinit()
lb = _CAN_MOD(
0,
extframe=False,
tx=_CFG["can_tx"],
rx=_CFG["can_rx"],
mode=_CAN_MOD.LOOPBACK,
bitrate=_CFG["can_baud"],
auto_restart=False,
)
test_id = int(cmd.get("id", 0x123))
test_data = list(cmd.get("data", [0x01, 0x02, 0x03, 0x04]))
timeout = int(cmd.get("timeout_ms", 500))
lb.send(test_data, test_id)
deadline = time.ticks_ms() + timeout
msg = None
while time.ticks_diff(deadline, time.ticks_ms()) > 0:
if lb.any():
msg = lb.recv()
break
time.sleep_ms(5)
lb.deinit()
if msg is None:
return {"ok": False, "err": "loopback timeout — фрейм не вернулся"}
recv_id, _, ext, recv_data = msg
matched = (recv_id == test_id) and (list(recv_data[:len(test_data)]) == test_data)
return {
"ok": True,
"matched": matched,
"sent_id": test_id,
"recv_id": recv_id,
"sent_data": test_data,
"recv_data": list(recv_data),
}
except Exception as e:
return {"ok": False, "err": "loopback error: %s" % e}
# Кортеж от биндинга: (identifier, extd, rtr, data)
frame_id, extd, rtr, data = msg # ← правильная распаковка
return {"ok": True, "id": frame_id, "ext": bool(extd), "data": list(data)}
return {"ok": False, "err": "unknown cmd: %r" % c}

2
tools/hil/pyproject.toml
View file

@ -12,7 +12,7 @@ dependencies = [
]
[tool.pytest.ini_options]
testpaths = ["."]
python_files = ["test_*.py"]
python_files = ["test_*.py", "*_test.py", "??_test_*.py"]
markers = [
"gpio: тесты GPIO",

0
tools/hil/test_can.py
View file