124 lines
5.7 KiB
Python
124 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the actual firmware CAN service/send functions against a host HAL stub."""
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
|
|
source = (Path(__file__).resolve().parents[1] /
|
|
'A11_F407_SDC_ACQ_DAC_CubeMX/Src/main.c').read_text()
|
|
|
|
def function(name):
|
|
start = source.index('static void ' + name + '(', source.index('static void ServiceCanHealth(CAN_HandleTypeDef *hcan)\n{'))
|
|
brace = source.index('{', start)
|
|
depth = 1
|
|
end = brace + 1
|
|
while depth:
|
|
depth += (source[end] == '{') - (source[end] == '}')
|
|
end += 1
|
|
return source[start:end]
|
|
|
|
state_start = source.index('typedef struct\n{\n uint32_t last_esr;')
|
|
state_end = source.index('static bool can_tx_abort_requested[2][3];', state_start)
|
|
state_end += len('static bool can_tx_abort_requested[2][3];')
|
|
harness = r'''
|
|
#include <assert.h>
|
|
#include <stdbool.h>
|
|
#include <stdint.h>
|
|
#include <string.h>
|
|
#include <stdio.h>
|
|
#define CAN_TX_TIMEOUT_MS 100U
|
|
#define CAN_ESR_BOFF 4U
|
|
#define CAN_ESR_EPVF 2U
|
|
#define CAN_TSR_TXOK0 2U
|
|
#define CAN_ID_STD 0U
|
|
#define CAN_RTR_DATA 0U
|
|
#define DISABLE 0U
|
|
#define HAL_OK 0U
|
|
#define HAL_ERROR 1U
|
|
typedef struct { uint32_t ESR, TSR, pending; } CAN_TypeDef;
|
|
typedef struct { CAN_TypeDef *Instance; } CAN_HandleTypeDef;
|
|
typedef struct { uint32_t StdId, ExtId, IDE, RTR, DLC, TransmitGlobalTime; } CAN_TxHeaderTypeDef;
|
|
static CAN_TypeDef regs[2];
|
|
static CAN_HandleTypeDef hcan1 = {®s[0]}, hcan2 = {®s[1]};
|
|
static uint32_t now, abort_calls, add_result, abort_result;
|
|
static bool race_completion;
|
|
static uint32_t HAL_GetTick(void) { return now; }
|
|
static uint32_t HAL_CAN_IsTxMessagePending(CAN_HandleTypeDef *h, uint32_t mask) {
|
|
return (h->Instance->pending & mask) != 0;
|
|
}
|
|
static uint32_t HAL_CAN_GetTxMailboxesFreeLevel(CAN_HandleTypeDef *h) {
|
|
if (race_completion) { h->Instance->pending = 0; race_completion = false; }
|
|
return 3 - __builtin_popcount(h->Instance->pending);
|
|
}
|
|
static uint32_t HAL_CAN_AbortTxRequest(CAN_HandleTypeDef *h, uint32_t mask) {
|
|
(void)h; (void)mask; abort_calls++; return abort_result;
|
|
}
|
|
static uint32_t HAL_CAN_AddTxMessage(CAN_HandleTypeDef *h, const CAN_TxHeaderTypeDef *hdr,
|
|
uint8_t *data, uint32_t *mailbox) {
|
|
(void)hdr; (void)data;
|
|
if (add_result) return add_result;
|
|
for (uint32_t i = 0; i < 3; i++) if (!(h->Instance->pending & (1U << i))) {
|
|
*mailbox = 1U << i; h->Instance->pending |= *mailbox;
|
|
h->Instance->TSR &= ~(CAN_TSR_TXOK0 << (8 * i)); return HAL_OK;
|
|
}
|
|
return HAL_ERROR;
|
|
}
|
|
'''
|
|
harness += source[state_start:state_end] + '\n'
|
|
harness += function('ServiceCanHealth') + '\n' + function('SendCanFrame')
|
|
harness += r'''
|
|
static void reset(void) {
|
|
memset(regs, 0, sizeof regs);
|
|
memset((void *)can_diagnostics, 0, sizeof can_diagnostics);
|
|
memset(can_tx_tracked, 0, sizeof can_tx_tracked);
|
|
memset(can_tx_abort_requested, 0, sizeof can_tx_abort_requested);
|
|
now = abort_calls = add_result = abort_result = 0; race_completion = false;
|
|
}
|
|
static void send(CAN_HandleTypeDef *h) { uint8_t data[8] = {0}; SendCanFrame(h, 0x6A0, data, 8); }
|
|
int main(void) {
|
|
reset(); send(&hcan1);
|
|
assert(can_diagnostics[0].tx_queued == 1 && can_diagnostics[0].tx_completed == 0);
|
|
regs[0].pending = 0; regs[0].TSR = CAN_TSR_TXOK0;
|
|
ServiceCanHealth(&hcan1); ServiceCanHealth(&hcan1);
|
|
assert(can_diagnostics[0].tx_completed == 1); /* counted only once */
|
|
send(&hcan1); regs[0].pending = 0; ServiceCanHealth(&hcan1);
|
|
assert(can_diagnostics[0].tx_failed == 1); /* HAL_OK is not delivery */
|
|
reset(); send(&hcan2); now = 99; ServiceCanHealth(&hcan2); assert(abort_calls == 0);
|
|
now = 100; ServiceCanHealth(&hcan2); ServiceCanHealth(&hcan2);
|
|
assert(abort_calls == 1 && can_diagnostics[1].abort_requests == 1);
|
|
send(&hcan1); assert(can_diagnostics[0].tx_queued == 1); /* other bus independent */
|
|
regs[1].pending = 0; ServiceCanHealth(&hcan2);
|
|
assert(can_diagnostics[1].tx_failed == 1);
|
|
reset(); now = UINT32_MAX - 49; send(&hcan1); now = 49;
|
|
ServiceCanHealth(&hcan1); assert(abort_calls == 0);
|
|
now = 50; ServiceCanHealth(&hcan1); assert(abort_calls == 1); /* tick wrap */
|
|
reset(); send(&hcan1); regs[0].ESR = CAN_ESR_BOFF | CAN_ESR_EPVF;
|
|
send(&hcan1); ServiceCanHealth(&hcan1);
|
|
assert(can_diagnostics[0].bus_off_events == 1 && can_diagnostics[0].passive_events == 1);
|
|
assert(can_diagnostics[0].tx_dropped == 1 && abort_calls == 1);
|
|
regs[0].pending = 0; regs[0].ESR = 0; send(&hcan1);
|
|
assert(can_diagnostics[0].recovery_events == 1 && can_diagnostics[0].tx_queued == 2);
|
|
reset(); send(&hcan1); send(&hcan1); send(&hcan1); send(&hcan1);
|
|
assert(can_diagnostics[0].tx_queued == 3 && can_diagnostics[0].tx_dropped == 1);
|
|
reset(); add_result = HAL_ERROR; send(&hcan1);
|
|
assert(can_diagnostics[0].tx_dropped == 1 && !can_tx_tracked[0][0]);
|
|
reset(); send(&hcan1); now = 100; abort_result = HAL_ERROR; ServiceCanHealth(&hcan1);
|
|
assert(can_diagnostics[0].abort_failures == 1 && !can_tx_abort_requested[0][0]);
|
|
abort_result = HAL_OK; ServiceCanHealth(&hcan1); assert(can_diagnostics[0].abort_requests == 1);
|
|
reset(); send(&hcan1); race_completion = true; send(&hcan1);
|
|
assert(can_diagnostics[0].tx_unobserved == 1 && can_diagnostics[0].tx_queued == 2);
|
|
/* Validate success bit selection for mailboxes 1 and 2. */
|
|
reset(); send(&hcan1); send(&hcan1); send(&hcan1);
|
|
regs[0].pending = 0; regs[0].TSR = (CAN_TSR_TXOK0 << 8) | (CAN_TSR_TXOK0 << 16);
|
|
ServiceCanHealth(&hcan1);
|
|
assert(can_diagnostics[0].tx_completed == 2 && can_diagnostics[0].tx_failed == 1);
|
|
puts("CAN health tests passed");
|
|
}
|
|
'''
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp)
|
|
(root / 'test.c').write_text(harness)
|
|
subprocess.run(['cc', '-std=c11', '-Wall', '-Wextra', '-Werror',
|
|
'-fsanitize=undefined', str(root / 'test.c'), '-o', str(root / 'test')], check=True)
|
|
subprocess.run([str(root / 'test')], check=True)
|