43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
from setprotocol import Frame, FrameFlag, MessageType, build_frame, decode_datagram
|
|
from setprotocol.can import CanAddress, CanFrame, CanReassembler, segment
|
|
|
|
|
|
def test_can_id_roundtrip():
|
|
value = CanAddress(destination=13, source=0, priority=1, channel=7)
|
|
assert CanAddress.unpack(value.pack()) == value
|
|
|
|
|
|
def test_segmented_setp_frame_roundtrip():
|
|
packet = build_frame(
|
|
Frame(
|
|
message_type=MessageType.FW_DATA,
|
|
sequence=0x1234,
|
|
payload=bytes(range(76)),
|
|
flags=FrameFlag.ACK_REQUIRED | FrameFlag.PRIORITY,
|
|
source=0,
|
|
destination=13,
|
|
)
|
|
)
|
|
frames = segment(packet, CanAddress(13, 0, priority=1, channel=1))
|
|
assert len(frames) == 14
|
|
reassembler = CanReassembler()
|
|
restored = None
|
|
for now_ms, frame in enumerate(frames):
|
|
restored = reassembler.feed(frame, now_ms)
|
|
assert restored == packet
|
|
assert decode_datagram(restored).payload == bytes(range(76))
|
|
|
|
|
|
def test_reassembler_rejects_wrong_sequence():
|
|
packet = build_frame(Frame(MessageType.PING, 1, source=0, destination=13))
|
|
frames = segment(packet, CanAddress(13, 0))
|
|
reassembler = CanReassembler()
|
|
assert reassembler.feed(frames[0], 0) is None
|
|
damaged = CanFrame(frames[1].can_id, bytes((0x22,)) + frames[1].data[1:])
|
|
try:
|
|
reassembler.feed(damaged, 1)
|
|
except ValueError as error:
|
|
assert "последовательность" in str(error)
|
|
else:
|
|
raise AssertionError("wrong sequence must fail")
|