rsp2k
96ab07889d
The transport is threaded and the session puts an async face on it, so pings and acks are answered whether or not anyone is awaiting a coroutine. Telemetry is stored per key with a timestamp, because a drone that has stopped reporting otherwise reads identically to one repeating itself. Encoding goes through protocol/codec lazily, so this lands without waiting for that stream; the sim carries a small encoder of its own for the events it sends, which also keeps it from agreeing with the client about a shared mistake. Ported from bebop-2's sim.py, retargeted at the vendored XML and with the identity burst now sent per controller attach rather than once per process.
110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
"""Framing, with no sockets involved.
|
|
|
|
These are the cases that cost real debugging time on the live aircraft: a
|
|
datagram carrying several frames, a size field that would never advance the
|
|
cursor, and the command id being two bytes rather than one.
|
|
"""
|
|
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from mcbebop.arsdk.types import COMMAND_HEADER, FRAME_HEADER, BufferId, DataType, Frame
|
|
|
|
|
|
def test_header_is_seven_bytes():
|
|
# Everything about the size field depends on this, including the sim.
|
|
assert FRAME_HEADER.size == 7
|
|
assert COMMAND_HEADER.size == 4
|
|
|
|
|
|
def test_encode_declares_size_including_header():
|
|
frame = Frame(DataType.DATA, BufferId.C2D_NON_ACK, 3, b"abcd")
|
|
raw = frame.encode()
|
|
data_type, buffer_id, seq, size = FRAME_HEADER.unpack_from(raw)
|
|
assert (data_type, buffer_id, seq) == (2, 10, 3)
|
|
assert size == len(raw) == 11
|
|
|
|
|
|
def test_round_trip():
|
|
frame = Frame(DataType.DATA_WITH_ACK, BufferId.C2D_ACK, 255, bytes(range(16)))
|
|
assert Frame.decode_all(frame.encode()) == [frame]
|
|
|
|
|
|
def test_empty_payload_round_trips():
|
|
frame = Frame(DataType.ACK, 139, 0, b"")
|
|
assert Frame.decode_all(frame.encode()) == [frame]
|
|
|
|
|
|
def test_several_frames_in_one_datagram():
|
|
frames = [
|
|
Frame(DataType.DATA, BufferId.D2C_NON_ACK, 1, b"\x01\x02"),
|
|
Frame(DataType.DATA_WITH_ACK, BufferId.D2C_ACK, 2, b"xyz"),
|
|
Frame(DataType.DATA, BufferId.PING, 3, struct.pack("<Q", 9)),
|
|
]
|
|
assert Frame.decode_all(b"".join(f.encode() for f in frames)) == frames
|
|
|
|
|
|
def test_zero_size_does_not_loop_forever():
|
|
# A size below the header length can never advance the cursor. The parse
|
|
# has to stop rather than spin, which is why decode_all checks it.
|
|
bogus = FRAME_HEADER.pack(2, 126, 0, 0) + b"junk"
|
|
assert Frame.decode_all(bogus) == []
|
|
|
|
|
|
def test_size_shorter_than_header_after_a_good_frame():
|
|
good = Frame(DataType.DATA, BufferId.D2C_NON_ACK, 7, b"ok")
|
|
data = good.encode() + FRAME_HEADER.pack(2, 126, 8, 3)
|
|
assert Frame.decode_all(data) == [good]
|
|
|
|
|
|
def test_truncated_frame_is_dropped_not_guessed():
|
|
# A frame claiming more than the datagram holds is a short read, and
|
|
# half a command decodes into a plausible-looking wrong one.
|
|
whole = Frame(DataType.DATA, BufferId.D2C_NON_ACK, 1, b"0123456789").encode()
|
|
assert Frame.decode_all(whole[:-4]) == []
|
|
|
|
|
|
def test_trailing_bytes_below_header_length_are_ignored():
|
|
good = Frame(DataType.DATA, BufferId.D2C_NON_ACK, 1, b"hi")
|
|
assert Frame.decode_all(good.encode() + b"\x02\x7e") == [good]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("buffer_id", "ack_buffer"),
|
|
[
|
|
(BufferId.C2D_ACK, 139), # the one pyparrot hardcodes, so a good check
|
|
(BufferId.C2D_NON_ACK, 138),
|
|
(BufferId.C2D_HIGH_PRIO, 140),
|
|
(BufferId.D2C_ACK, 255), # 127 + 128 lands on the last valid buffer id
|
|
(BufferId.PING, 128),
|
|
],
|
|
)
|
|
def test_ack_buffer_is_offset_by_128(buffer_id, ack_buffer):
|
|
assert BufferId.ack_for(buffer_id) == ack_buffer
|
|
|
|
|
|
def test_ack_buffer_wraps_rather_than_overflowing():
|
|
# 128 upwards would exceed the one-byte field; the modulo is the point.
|
|
assert BufferId.ack_for(200) == 72
|
|
assert all(0 <= BufferId.ack_for(b) < 256 for b in range(256))
|
|
|
|
|
|
def test_command_id_is_two_bytes():
|
|
# ardrone3.MediaStreaming.VideoEnable is (1, 21, 0); the ids that matter
|
|
# for this are the ones above 255, which a u8 command field would mangle.
|
|
raw = COMMAND_HEADER.pack(1, 21, 0)
|
|
assert COMMAND_HEADER.unpack(raw) == (1, 21, 0)
|
|
assert COMMAND_HEADER.unpack(COMMAND_HEADER.pack(0, 5, 300)) == (0, 5, 300)
|
|
|
|
|
|
def test_command_payload_sits_inside_a_frame():
|
|
payload = COMMAND_HEADER.pack(0, 4, 0) # common.Common.AllStates
|
|
frame = Frame(DataType.DATA_WITH_ACK, BufferId.C2D_ACK, 0, payload)
|
|
(decoded,) = Frame.decode_all(frame.encode())
|
|
assert COMMAND_HEADER.unpack(decoded.payload) == (0, 4, 0)
|
|
|
|
|
|
def test_data_type_numbers_match_the_captures():
|
|
assert (DataType.ACK, DataType.DATA, DataType.LOW_LATENCY, DataType.DATA_WITH_ACK) == (1, 2, 3, 4)
|