rsp2k
ef8c4658e0
Parse Parrot's common.xml and ardrone3.xml into CommandSpec objects rather than transcribing 264 commands by hand. Direction is derived from comment result=/triggered= with a class-name-suffix fallback for the 31 commands that state neither, giving 101 to-drone and 163 from-drone. ardrone3.Piloting.FlatTrim is injected: the firmware accepts it but neither the vendored snapshot nor upstream defines it. codec encodes the <BBH header (the command id is 16 bits, not 8) and reads enums as i32 and strings as NUL-terminated, both of which pyparrot gets wrong. safety assigns the arming tier by class, with Mavlink and Calibration counted as motion because they move the aircraft. expectations parses the undocumented #p-c-m grammar including | alternatives and this.<arg> echoes. 90 tests, counts asserted so a bad parse fails loudly.
245 lines
9.6 KiB
Python
245 lines
9.6 KiB
Python
"""Wire-format tests. The byte counts here are the whole point.
|
|
|
|
Two of them exist because pyparrot gets them wrong and pyparrot is the obvious
|
|
thing to check against: the command id is 16 bits, and an enum is 32.
|
|
"""
|
|
|
|
import struct
|
|
|
|
import pytest
|
|
|
|
from mcbebop.protocol import codec, xml_index
|
|
from mcbebop.protocol.types import ArgSpec, CommandSpec, EnumSpec
|
|
|
|
|
|
def test_header_is_four_bytes_with_a_16_bit_command_id():
|
|
assert codec.HEADER.size == 4
|
|
assert codec.HEADER.pack(1, 0, 2) == b"\x01\x00\x02\x00"
|
|
# An id above 255 must survive, even though no current command has one.
|
|
assert codec.HEADER.unpack(codec.HEADER.pack(1, 4, 300)) == (1, 4, 300)
|
|
|
|
|
|
def test_encode_command_is_header_plus_args():
|
|
spec = xml_index.get("ardrone3.MediaStreaming.VideoEnable")
|
|
assert codec.encode_command(spec, {"enable": 1}) == b"\x01\x15\x00\x00\x01"
|
|
|
|
|
|
def test_no_arg_command_encodes_to_nothing():
|
|
assert codec.encode_args(xml_index.get("ardrone3.Piloting.TakeOff")) == b""
|
|
assert codec.encode_args(xml_index.get("ardrone3.Piloting.FlatTrim"), {}) == b""
|
|
|
|
|
|
def test_enum_is_four_bytes_little_endian():
|
|
spec = xml_index.get("ardrone3.Network.WifiScan")
|
|
assert codec.encode_args(spec, {"band": "5ghz"}) == b"\x01\x00\x00\x00"
|
|
assert codec.encode_args(spec, {"band": "2_4ghz"}) == b"\x00\x00\x00\x00"
|
|
# An int position is accepted too, for replaying captured traffic.
|
|
assert codec.encode_args(spec, {"band": 2}) == b"\x02\x00\x00\x00"
|
|
|
|
|
|
def test_enum_decodes_back_to_the_member_name():
|
|
spec = xml_index.get("ardrone3.Network.WifiScan")
|
|
assert codec.decode_args(spec, b"\x01\x00\x00\x00") == {"WifiScan_band": "5ghz"}
|
|
|
|
|
|
def test_unknown_enum_position_decodes_to_the_number():
|
|
# A firmware member this XML snapshot predates must not raise.
|
|
spec = xml_index.get("ardrone3.Network.WifiScan")
|
|
assert codec.decode_args(spec, b"\x63\x00\x00\x00") == {"WifiScan_band": 99}
|
|
|
|
|
|
def test_string_is_nul_terminated():
|
|
spec = xml_index.get("common.Settings.ProductName")
|
|
assert codec.encode_args(spec, {"name": "Bebop"}) == b"Bebop\x00"
|
|
assert codec.decode_args(spec, b"Bebop\x00") == {"ProductName_name": "Bebop"}
|
|
|
|
|
|
def test_string_arguments_can_be_followed_by_more_arguments():
|
|
# A variable-length field mid-payload is where a fixed-width read goes
|
|
# wrong: everything after the string shifts.
|
|
spec = xml_index.get("ardrone3.NetworkState.WifiScanListChanged")
|
|
assert [a.type for a in spec.args] == ["string", "i16", "enum", "u8"]
|
|
payload = b"MyNet\x00" + struct.pack("<h", -70) + struct.pack("<i", 1) + b"\x0b"
|
|
assert codec.decode_args(spec, payload) == {
|
|
"WifiScanListChanged_ssid": "MyNet",
|
|
"WifiScanListChanged_rssi": -70,
|
|
"WifiScanListChanged_band": "5ghz",
|
|
"WifiScanListChanged_channel": 11,
|
|
}
|
|
|
|
|
|
def test_decoded_keys_match_the_existing_captures():
|
|
spec = xml_index.get("common.CommonState.BatteryStateChanged")
|
|
decoded = codec.decode_args(spec, b"\x4b")
|
|
assert decoded == {"BatteryStateChanged_percent": 75}
|
|
|
|
|
|
def test_decode_event_reads_the_header():
|
|
spec = xml_index.get("common.CommonState.BatteryStateChanged")
|
|
payload = codec.HEADER.pack(*spec.ids) + b"\x4b"
|
|
ids, values = codec.decode_event(payload)
|
|
assert ids == spec.ids
|
|
assert values == {"BatteryStateChanged_percent": 75}
|
|
|
|
|
|
def test_decode_event_rejects_a_truncated_header():
|
|
with pytest.raises(ValueError, match="too short"):
|
|
codec.decode_event(b"\x00\x04\x00")
|
|
|
|
|
|
def test_decode_event_names_an_unknown_command_rather_than_dropping_it():
|
|
with pytest.raises(ValueError, match=r"no command with ids \(7, 7, 7\)"):
|
|
codec.decode_event(codec.HEADER.pack(7, 7, 7))
|
|
|
|
|
|
def test_decode_rejects_a_short_payload():
|
|
spec = xml_index.get("ardrone3.Piloting.PCMD")
|
|
with pytest.raises(ValueError, match="bytes but only"):
|
|
codec.decode_args(spec, b"\x01\x00")
|
|
|
|
|
|
def test_decode_rejects_an_unterminated_string():
|
|
spec = xml_index.get("common.Settings.ProductName")
|
|
with pytest.raises(ValueError, match="unterminated string"):
|
|
codec.decode_args(spec, b"Bebop")
|
|
|
|
|
|
def test_pcmd_round_trip():
|
|
# The stick packet: the one command sent dozens of times a second.
|
|
spec = xml_index.get("ardrone3.Piloting.PCMD")
|
|
args = {"flag": 1, "roll": -20, "pitch": 30, "yaw": 0, "gaz": 10, "timestampAndSeqNum": 12345}
|
|
encoded = codec.encode_args(spec, args)
|
|
assert len(encoded) == 1 + 1 + 1 + 1 + 1 + 4
|
|
decoded = codec.decode_args(spec, encoded)
|
|
assert decoded["PCMD_roll"] == -20
|
|
assert decoded["PCMD_timestampAndSeqNum"] == 12345
|
|
|
|
|
|
def _plausible(arg: ArgSpec):
|
|
"""A value the spec allows, chosen so a round trip is exact."""
|
|
if arg.type == "string":
|
|
return "test"
|
|
if arg.is_enum:
|
|
return arg.members[0].name if arg.members else 0
|
|
if arg.is_bitfield:
|
|
return [arg.members[0].name] if arg.members else 0
|
|
if arg.type in ("float", "double"):
|
|
return 1.5 # exact in both widths, so no tolerance needed
|
|
if arg.type.startswith("i"):
|
|
return -3
|
|
return 3
|
|
|
|
|
|
def test_round_trip_every_command_that_has_arguments():
|
|
checked = 0
|
|
for spec in xml_index.all_commands():
|
|
if not spec.args:
|
|
continue
|
|
sent = {arg.name: _plausible(arg) for arg in spec.args}
|
|
decoded = codec.decode_args(spec, codec.encode_args(spec, sent))
|
|
assert len(decoded) == len(spec.args), spec.full_name
|
|
for arg in spec.args:
|
|
got = decoded[f"{spec.name}_{arg.name}"]
|
|
expected = sent[arg.name]
|
|
if arg.is_enum and not arg.members:
|
|
expected = 0
|
|
assert got == expected, f"{spec.full_name}.{arg.name} ({arg.type}) gave {got!r}"
|
|
checked += 1
|
|
assert checked == 234 # every command that declares at least one argument
|
|
|
|
|
|
def test_encoded_length_matches_the_declared_types():
|
|
sizes = {"u8": 1, "i8": 1, "u16": 2, "i16": 2, "u32": 4, "i32": 4, "u64": 8, "i64": 8}
|
|
sizes |= {"float": 4, "double": 8}
|
|
for spec in xml_index.all_commands():
|
|
if any(a.type == "string" for a in spec.args):
|
|
continue
|
|
expected = sum(4 if a.is_enum else sizes[a.type] for a in spec.args)
|
|
sent = {arg.name: _plausible(arg) for arg in spec.args}
|
|
assert len(codec.encode_args(spec, sent)) == expected, spec.full_name
|
|
|
|
|
|
def test_missing_argument_is_named():
|
|
spec = xml_index.get("ardrone3.Piloting.PCMD")
|
|
with pytest.raises(ValueError, match="requires argument 'flag'"):
|
|
codec.encode_args(spec, {})
|
|
|
|
|
|
def test_unknown_argument_is_named_and_the_real_ones_listed():
|
|
spec = xml_index.get("ardrone3.MediaStreaming.VideoEnable")
|
|
with pytest.raises(ValueError, match="has no argument nope; expected: enable"):
|
|
codec.encode_args(spec, {"enable": 1, "nope": 2})
|
|
|
|
|
|
def test_out_of_range_integer_is_named():
|
|
spec = xml_index.get("ardrone3.MediaStreaming.VideoEnable")
|
|
with pytest.raises(ValueError, match="'enable' \\(u8\\) cannot hold 999"):
|
|
codec.encode_args(spec, {"enable": 999})
|
|
|
|
|
|
def test_wrong_type_is_named():
|
|
spec = xml_index.get("ardrone3.MediaStreaming.VideoEnable")
|
|
with pytest.raises(ValueError, match="'enable' is a u8, got str"):
|
|
codec.encode_args(spec, {"enable": "yes"})
|
|
with pytest.raises(ValueError, match="pass 1 or 0 rather than a bool"):
|
|
codec.encode_args(spec, {"enable": True})
|
|
|
|
|
|
def test_unknown_enum_member_lists_the_allowed_ones():
|
|
spec = xml_index.get("ardrone3.Network.WifiScan")
|
|
with pytest.raises(ValueError, match="no member named 'wifi6'; allowed: 2_4ghz, 5ghz, all"):
|
|
codec.encode_args(spec, {"band": "wifi6"})
|
|
with pytest.raises(ValueError, match="no member with value 9"):
|
|
codec.encode_args(spec, {"band": 9})
|
|
|
|
|
|
def test_embedded_nul_in_a_string_is_refused():
|
|
spec = xml_index.get("common.Settings.ProductName")
|
|
with pytest.raises(ValueError, match="contains a NUL byte"):
|
|
codec.encode_args(spec, {"name": "Beb\x00op"})
|
|
|
|
|
|
def _bitfield_spec(width: str) -> CommandSpec:
|
|
"""Parrot's other projects use bitfields; these two files do not."""
|
|
arg = ArgSpec(
|
|
name="flags",
|
|
type=f"bitfield:{width}:Flags",
|
|
members=(EnumSpec("first", 0), EnumSpec("second", 1), EnumSpec("third", 2)),
|
|
)
|
|
return CommandSpec(project="x", klass="Y", name="Z", ids=(9, 9, 9), args=(arg,))
|
|
|
|
|
|
def test_bitfield_takes_member_names_or_a_mask():
|
|
spec = _bitfield_spec("u8")
|
|
assert codec.encode_args(spec, {"flags": ["first", "third"]}) == b"\x05"
|
|
assert codec.encode_args(spec, {"flags": "second"}) == b"\x02"
|
|
assert codec.encode_args(spec, {"flags": 0xFF}) == b"\xff"
|
|
assert codec.decode_args(spec, b"\x05") == {"Z_flags": 5}
|
|
|
|
|
|
def test_bitfield_width_sets_the_wire_size():
|
|
assert len(codec.encode_args(_bitfield_spec("u16"), {"flags": 1})) == 2
|
|
assert len(codec.encode_args(_bitfield_spec("u32"), {"flags": 1})) == 4
|
|
with pytest.raises(ValueError, match="unknown bitfield width 'u7'"):
|
|
codec.encode_args(_bitfield_spec("u7"), {"flags": 1})
|
|
|
|
|
|
def test_bitfield_rejects_an_unknown_member():
|
|
with pytest.raises(ValueError, match="no member named 'fourth'"):
|
|
codec.encode_args(_bitfield_spec("u8"), {"flags": ["fourth"]})
|
|
|
|
|
|
def test_unsupported_type_is_reported():
|
|
blob = ArgSpec(name="a", type="blob")
|
|
spec = CommandSpec(project="x", klass="Y", name="Z", ids=(9, 9, 8), args=(blob,))
|
|
with pytest.raises(ValueError, match="unsupported type 'blob'"):
|
|
codec.encode_args(spec, {"a": 1})
|
|
|
|
|
|
def test_every_type_in_the_xml_is_encodable():
|
|
declared = {a.type for c in xml_index.all_commands() for a in c.args}
|
|
for type_name in declared:
|
|
if type_name in ("string", "enum"):
|
|
continue
|
|
assert struct.calcsize(codec._FORMATS[type_name]) > 0
|