Andrey
c6390ae1c9
Обеспечь единый процесс обновления для STM32F103/F407/G474 и TMS320F2812/F28335: discovery, служебные ответы, BUSY, тайм-аут сессии и подтверждение приложения. Добавь самостоятельные проекты CCS, аппаратные хранилища и Python-клиент. Совместимость: структуры конфигурации и порта расширены, поэтому потребители должны пересобраться и обнулить новые поля. Для C28x callbacks используют pcan_boot_byte_t; периодический pcan_boot_poll обслуживает discovery и настроенный тайм-аут.
87 lines
6.4 KiB
Python
87 lines
6.4 KiB
Python
from types import SimpleNamespace
|
|
import struct
|
|
import pytest
|
|
|
|
from protocan.can_boot import CanBootServices, CanBootTarget, CanBootTransfer, CanBootPingExchange
|
|
from protocan.protocan import ProtoCanId
|
|
from protocan.transport import build_frame
|
|
|
|
|
|
def response(command, data, device=7):
|
|
return build_frame(ProtoCanId.build(1, 1, 3, device, 13, command), data, to_can=False)
|
|
|
|
|
|
def test_passport_and_discovery_address_filters():
|
|
service = CanBootServices(CanBootTarget(device_type=3, device=7, product_type=0x2833))
|
|
identify = response(1, struct.pack('<HBBI', 0x2833, 1, 1, 0x01010002))
|
|
assert service.decode(identify)['version'] == (1, 1, 2)
|
|
foreign = response(1, identify.data, device=8)
|
|
assert service.decode(foreign) is None
|
|
assert service.decode(foreign, discovery=True)['device'] == 8
|
|
capabilities = service.decode(response(13, struct.pack('<II', 0xFFF, 130560)))
|
|
assert capabilities['slot_capacity'] == 130560
|
|
assert capabilities['capabilities'] & (1 << 8)
|
|
assert service.decode(response(14, b'F28335-M'))['name_fragment'] == 'F28335-M'
|
|
assert service.decode(response(0x10E, b'95M02\0\0\0'))['name_fragment'] == '95M02'
|
|
memory = service.decode(response(15, struct.pack('<IHBB', 262144, 256, 1, 255)))
|
|
assert memory['bootloader'] and memory['memory_bytes'] == 262144
|
|
assert service.decode(response(0x10F, struct.pack('<II', 512, 131072)))['slot_offsets'] == (512, 131072)
|
|
pending = service.decode(response(0x20F, struct.pack('<BBBBI', 0, 1, 1, 1, 30000)))
|
|
assert pending['trial_attempted'] and pending['pending_slot'] == 1
|
|
assert service.decode(response(0x30F, struct.pack('<II', 0x3FEFC, 258)))['parameters_bytes'] == 258
|
|
assert service.decode(response(0x40F, bytes(8))) is None
|
|
discover = ProtoCanId.parse(service.discover().can_id)
|
|
assert (discover.device_type, discover.device, discover.msg_type, discover.body) == (7, 15, 13, 0)
|
|
|
|
|
|
def test_service_requests_and_transfer_limits():
|
|
target = CanBootTarget(device_type=3, device=7, product_type=0x2833, session_id=15)
|
|
service = CanBootServices(target)
|
|
request = service.request(15, 2)
|
|
assert request.data == b'\x02' and ProtoCanId.parse(request.can_id).body == 0xF0F
|
|
assert ProtoCanId.parse(service.request(8).can_id).body == 0xF08
|
|
with pytest.raises(ValueError):
|
|
service.request(5)
|
|
with pytest.raises(ValueError):
|
|
service.request(1, 3)
|
|
with pytest.raises(ValueError):
|
|
CanBootTransfer(SimpleNamespace(data=bytes(130561), version=1), target)
|
|
transfer = CanBootTransfer(SimpleNamespace(data=bytes(40), version=1), target)
|
|
assert transfer.response_timeout_ms == 15000
|
|
transfer.stage = 'data'
|
|
assert transfer.response_timeout_ms == 1500
|
|
|
|
|
|
def test_ping_exchange_only_accepts_addressed_ping_reply():
|
|
ping = CanBootPingExchange(CanBootTarget(device_type=3, device=7))
|
|
assert ProtoCanId.parse(ping.request()[0].can_id).body & 255 == 12
|
|
assert ping.feed(response(1, bytes(8))) is None
|
|
assert ping.feed(response(12, bytes([1, 0, 0, 255, 0, 2, 0, 0]), device=8)) is None
|
|
assert ping.feed(response(12, bytes([1, 0, 0, 255, 0, 2, 0, 0])))['slot_count'] == 2
|
|
|
|
|
|
def test_confirm_and_progress_status_correlate_session():
|
|
service = CanBootServices(CanBootTarget(device_type=3, device=7, session_id=5))
|
|
frame = build_frame(ProtoCanId.build(1, 1, 3, 7, 12, 0x508),
|
|
struct.pack('<BBHI', 0, 1, 123, 0x12345678), to_can=False)
|
|
reply = service.decode(frame)
|
|
assert reply['status'] == 0 and reply['command'] == 8 and reply['next_block'] == 123
|
|
assert service.decode(frame, discovery=True) is None
|
|
assert CanBootServices(CanBootTarget(device_type=3, device=7, session_id=6)).decode(frame) is None
|
|
|
|
|
|
@pytest.mark.parametrize("code,expected", [(0, 0), (0, 8), (0, 16), (8, 0), (8, 8), (8, 16)]) # Проверяет повторные и запоздалые OK/SEQUENCE_ERROR после подтверждённого первого окна.
|
|
def test_duplicate_data_status_does_not_resend_or_rewind_window(code, expected): # Один поздний ответ не должен порождать дополнительные 16 кадров в очереди MCU.
|
|
target = CanBootTarget(device_type=1, device=8, product_type=0x2833) # Использует адрес задатчика, на котором наблюдалось зависание передачи.
|
|
client = CanBootTransfer(SimpleNamespace(data=bytes(512), version=1, crc32=0), target) # Несколько окон позволяют проверить ответ до завершения передачи.
|
|
def status_frame(command, status_code=0, next_block=0): # Формирует тестовый STATUS в той же сессии, что запрос клиента.
|
|
return build_frame(ProtoCanId.build(1, 1, 1, 8, 12, 0x100 | command), struct.pack("<BBHI", status_code, 0, next_block, 0), to_can=False) # Session 1 и слот A отделяют проверку повторов от адресной фильтрации.
|
|
client.start() # Начинает обычную последовательность ENTER_BOOT и метаданных.
|
|
for command in (2, 3, 4, 5): # Переводит production-клиент к первому окну через штатные подтверждения этапов.
|
|
client.handle_status(status_frame(command)) # Каждый управляющий ответ разрешает только следующий этап протокола.
|
|
outgoing, _ = client.handle_status(status_frame(0, next_block=16)) # Подтверждает первое окно и ставит в отправку блоки 16–31.
|
|
assert len(outgoing) == 16 and client.next_block == 16 # Контрольное состояние должно содержать новое неподтверждённое окно.
|
|
for _ in range(16): # Остаток прежнего окна способен вернуть до 16 одинаковых ошибок последовательности.
|
|
assert client.handle_status(status_frame(0, code, expected)) == ([], "") # Дубликаты не пересылают текущее окно и не отодвигают транспортный тайм-аут.
|
|
assert client.next_block == 16 and client.stage == "data" # Запоздалый SEQUENCE_ERROR не откатывает уже подтверждённые блоки.
|