136 lines
5.4 KiB
Python
136 lines
5.4 KiB
Python
"""Android wire compatibility and complete host-side flashing transactions."""
|
|
from dataclasses import replace
|
|
from types import SimpleNamespace
|
|
import struct
|
|
import zlib
|
|
|
|
import pytest
|
|
|
|
from protocan import tms_firmware as tms
|
|
from protocan.can_boot import CanBootTarget, CanBootTransfer
|
|
from protocan.protocan import ProtoCanId
|
|
from protocan.transport import build_frame
|
|
|
|
|
|
def reply(command, payload=b"", tail=2):
|
|
return tms.packet(10, command, payload) + bytes(tail)
|
|
|
|
|
|
def test_android_packet_layout_and_crc():
|
|
assert tms.crc16_modbus(b"123456789") == 0x4B37
|
|
assert tms.init_load(10, 0xA0000, 0x1234)[:10] == bytes.fromhex("0a3a00000a0034120000")
|
|
assert tms.load_data(10, b"\1\2\3\4")[:6] == bytes.fromhex("0a3301020304")
|
|
assert tms.tflash(10, 0xA0000, 0x110000, 0x123456)[:14] == bytes.fromhex("0a3700000a000000110056341200")
|
|
assert tms.extend(10, 0xA0000, 0, 2, 10, 3)[:16] == bytes.fromhex("0a3c00000a0000000000020000000a03")
|
|
|
|
|
|
@pytest.mark.parametrize("command,payload,tail", [(58, b"", 2), (52, b"abcdef", 4)])
|
|
def test_only_reserved_tail_may_be_missing(command, payload, tail):
|
|
full = reply(command, payload, tail)
|
|
for n in range(tail + 1):
|
|
assert tms.normalize_reply(full[:len(full)-n], 10, command, len(full)) == full
|
|
assert tms.normalize_reply(full[:len(full)-tail-1], 10, command, len(full)) is None
|
|
corrupt = bytearray(full)
|
|
corrupt[2] ^= 1
|
|
assert tms.normalize_reply(bytes(corrupt), 10, command, len(full)) is None
|
|
assert tms.normalize_reply(full, 11, command, len(full)) is None
|
|
|
|
|
|
def run_tms(data, target, corrupt=False):
|
|
plan = tms.programming_steps(data, target)
|
|
response = None
|
|
requests = []
|
|
while True:
|
|
try:
|
|
request, size, timeout, title, percent = plan.send(response)
|
|
except StopIteration:
|
|
return requests
|
|
requests.append(request)
|
|
command = request[1]
|
|
if command == 52:
|
|
address, count = struct.unpack_from("<II", request, 2)
|
|
offset = (address-target.flash)*2
|
|
chunk = data[offset:offset+count].ljust(count, b"\xff")
|
|
if corrupt:
|
|
chunk = bytes([chunk[0] ^ 1]) + chunk[1:]
|
|
response = reply(command, chunk, 4)
|
|
assert len(response) == size
|
|
elif command == 60:
|
|
response = reply(command, struct.pack("<III", (len(data)+1)//2, 0, 0))
|
|
else:
|
|
response = reply(command)
|
|
|
|
|
|
def test_tms_word_offsets_and_odd_final_byte_readback():
|
|
requests = run_tms(bytes(range(256)) + b"x", tms.TmsTarget())
|
|
assert [p[1] for p in requests] == [58, 51, 58, 51, 55, 52, 52]
|
|
assert struct.unpack_from("<II", requests[2], 2) == (0xA0080, 1)
|
|
assert struct.unpack_from("<II", requests[-1], 2) == (0x110080, 2)
|
|
with pytest.raises(ValueError, match="не пройдена"):
|
|
run_tms(b"test", tms.TmsTarget(), corrupt=True)
|
|
|
|
|
|
@pytest.mark.parametrize("kind,codes", [("spartan2e", [6, 17]), ("spartan6", [10])])
|
|
def test_peripheral_flashing_and_verification(kind, codes):
|
|
packets = run_tms(b"test", tms.TmsTarget(kind=kind, flash=0, board=3))
|
|
assert [p[14] for p in packets if p[1] == 60] == codes
|
|
assert all(p[15] == 3 for p in packets if p[1] == 60)
|
|
|
|
|
|
def test_load_only_never_writes_flash():
|
|
assert [p[1] for p in run_tms(b"test", tms.TmsTarget(load_only=True))] == [58, 51]
|
|
|
|
|
|
@pytest.mark.parametrize("target,size", [(tms.TmsTarget(flash=0), 4), (tms.TmsTarget(), 0),
|
|
(tms.TmsTarget(block_size=257), 4), (tms.TmsTarget(flash=0x17FFFF), 3),
|
|
(tms.TmsTarget(ram=0xFFFFFFFF), 4)])
|
|
def test_invalid_targets_fail_before_transmission(target, size):
|
|
with pytest.raises(ValueError):
|
|
target.validate(size)
|
|
|
|
|
|
def image(data):
|
|
return SimpleNamespace(data=data, crc32=zlib.crc32(data), version=7)
|
|
|
|
|
|
def status(t, cmd, code=0, slot=1, expected=0):
|
|
target = t.target
|
|
can_id = ProtoCanId.build(1, 1, target.device_type, target.device, 12, target.session_id << 8 | cmd)
|
|
return build_frame(can_id, struct.pack("<BBHI", code, slot, expected, 0), to_can=False)
|
|
|
|
|
|
def enter_data(t):
|
|
assert ProtoCanId.parse(t.start()[0].can_id).msg_type == 9
|
|
for cmd in (2, 3, 4):
|
|
t.handle_status(status(t, cmd))
|
|
return t.handle_status(status(t, 5))[0]
|
|
|
|
|
|
def test_legacy_can_full_transaction_and_retransmission():
|
|
t = CanBootTransfer(image(bytes(range(130))), CanBootTarget())
|
|
frames = enter_data(t)
|
|
assert len(frames) == 16
|
|
assert all(ProtoCanId.parse(f.can_id).msg_type == 11 for f in frames)
|
|
frames, _ = t.handle_status(status(t, 0, code=8, expected=8))
|
|
assert ProtoCanId.parse(frames[0].can_id).body == 8
|
|
assert frames[-1].data == bytes([128,129]) + b"\xff" * 6
|
|
frames, _ = t.handle_status(status(t, 0, expected=17))
|
|
assert t.stage == "verify"
|
|
for cmd in (6, 7, 9):
|
|
t.handle_status(status(t, cmd))
|
|
assert t.finished and t.percent == 100
|
|
|
|
|
|
def test_can_rejects_unsent_blocks_wrong_slot_and_foreign_session():
|
|
t = CanBootTransfer(image(bytes(200)), CanBootTarget())
|
|
enter_data(t)
|
|
with pytest.raises(RuntimeError, match="непереданный"):
|
|
t.handle_status(status(t, 0, expected=20))
|
|
foreign = CanBootTransfer(t.image, replace(t.target, session_id=2))
|
|
assert not t.accepts(status(foreign, 0))
|
|
t = CanBootTransfer(image(bytes(8)), CanBootTarget())
|
|
t.handle_status(status(t, 2))
|
|
t.handle_status(status(t, 3))
|
|
with pytest.raises(RuntimeError, match="слот"):
|
|
t.handle_status(status(t, 4, slot=2))
|