Merge protocol layer: XML index, codec, safety tiers, expectations

90 tests. Verified against real data from the aircraft: VideoEnable encodes
to the bytes we actually sent, enums encode as 4-byte indices (pyparrot read
one byte and misaligned everything after), and events decode to the same
<Command>_<arg> keys our captures use.

Two counts in the brief were wrong and the stream corrected them with tests:
31 commands state neither result nor triggered (not 33), and 39 give no
answer on Bebop 2 support (not 32).

tests/test_codec.py

@@ -0,0 +1,244 @@
+"""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

tests/test_expectations.py

@@ -0,0 +1,113 @@
+"""The expectations grammar is undocumented, so the examples are the spec."""
+
+import xml.etree.ElementTree as ET
+
+import pytest
+
+from mcbebop.protocol import xml_index
+from mcbebop.protocol.expectations import expectations_for, parse_expectation_text
+from mcbebop.protocol.types import Direction
+
+
+def test_every_block_is_parsed_and_only_sendable_commands_carry_one():
+    with_expectations = [c for c in xml_index.all_commands() if c.expectations]
+    assert len(with_expectations) == 82
+    assert all(c.direction is Direction.TO_DRONE for c in with_expectations)
+
+
+def test_delayed_blocks():
+    delayed = [c.full_name for c in xml_index.all_commands() if any(e.delayed for e in c.expectations)]
+    assert sorted(delayed) == ["ardrone3.Network.WifiScan", "ardrone3.Network.WifiAuthChannel"][::-1]
+
+
+def test_bare_reference():
+    (expectation,) = parse_expectation_text("#0-3-0")
+    assert expectation.ids == (0, 3, 0)
+    assert expectation.fields == {}
+    assert expectation.alternatives == ()
+    assert expectation.delayed is False
+
+
+def test_this_dot_argname_means_the_drone_echoes_what_was_sent():
+    (expectation,) = xml_index.get("common.Settings.ProductName").expectations
+    assert expectation.ids == (0, 3, 2)
+    assert expectation.fields == {"name": "this.name"}
+
+
+def test_whitespace_separated_references_are_all_expected():
+    expectations = xml_index.get("common.Settings.AutoCountry").expectations
+    assert len(expectations) == 2
+    assert expectations[0].ids == (0, 3, 7)
+    assert expectations[0].fields == {"automatic": "this.automatic"}
+    assert expectations[1].ids == (0, 3, 6)
+    assert all(e.alternatives == () for e in expectations)
+
+
+def test_pipe_prefixed_references_are_alternatives():
+    (expectation,) = xml_index.get("ardrone3.Piloting.NavigateHome").expectations
+    assert expectation.ids == (1, 4, 3)
+    assert expectation.fields == {"state": "inProgress", "reason": "userRequest"}
+    assert len(expectation.alternatives) == 2
+    assert [a.fields["state"] for a in expectation.alternatives] == ["pending", "available"]
+    assert all(a.ids == (1, 4, 3) for a in expectation.alternatives)
+
+
+def test_alternatives_can_name_different_commands():
+    # StopPilotedPOI is confirmed by whichever POI event version is in play.
+    (expectation,) = xml_index.get("ardrone3.Piloting.StopPilotedPOI").expectations
+    assert expectation.ids == (1, 4, 14)
+    assert [a.ids for a in expectation.alternatives] == [(1, 4, 22)]
+
+
+def test_literal_enum_member_values():
+    (expectation,) = xml_index.get("ardrone3.Piloting.StartPilotedPOI").expectations
+    assert expectation.fields["status"] == "RUNNING"
+    assert expectation.fields["latitude"] == "this.latitude"
+    assert expectation.alternatives[0].fields["status"] == "PENDING"
+
+
+def test_takeoff_expects_two_states_not_one_of_two():
+    expectations = xml_index.get("ardrone3.Piloting.TakeOff").expectations
+    assert [e.fields["state"] for e in expectations] == ["motor_ramping", "takingoff"]
+
+
+def test_delayed_flag_reaches_the_alternatives_too():
+    parsed = parse_expectation_text("#1-14-0 |#1-14-1", delayed=True)
+    assert parsed[0].delayed is True
+    assert parsed[0].alternatives[0].delayed is True
+
+
+def test_spaced_pipe_still_parses():
+    (expectation,) = parse_expectation_text("#0-1-2 | #0-1-3")
+    assert [a.ids for a in expectation.alternatives] == [(0, 1, 3)]
+
+
+def test_empty_block_is_empty_not_an_error():
+    assert parse_expectation_text("\n\t\t\t") == ()
+
+
+def test_leading_alternative_is_rejected():
+    with pytest.raises(ValueError, match="starts with an alternative"):
+        parse_expectation_text("|#0-1-2")
+
+
+def test_text_without_a_reference_is_rejected():
+    with pytest.raises(ValueError, match="no #p-c-m reference"):
+        parse_expectation_text("ProductNameChanged")
+
+
+def test_malformed_field_is_rejected():
+    with pytest.raises(ValueError, match="is not 'name: value'"):
+        parse_expectation_text("#0-3-2(name)")
+
+
+def test_unknown_expectations_child_is_rejected():
+    cmd = ET.fromstring(
+        "<cmd name='X' id='0'><expectations><eventually>#0-0-0</eventually></expectations></cmd>"
+    )
+    with pytest.raises(ValueError, match="unknown expectations child"):
+        expectations_for(cmd)
+
+
+def test_commands_without_a_block_have_an_empty_tuple():
+    assert xml_index.get("ardrone3.Piloting.PCMD").expectations == ()

tests/test_protocol_index.py

@@ -0,0 +1,222 @@
+"""The index is derived, not transcribed, so these pin the derivation.
+
+Every count here was read off the two vendored XML files. A bad parse that
+still produces plausible-looking objects is the failure mode worth catching,
+so the numbers are asserted rather than the shapes.
+"""
+
+import collections
+import shutil
+
+import pytest
+
+from mcbebop.protocol import xml_index
+from mcbebop.protocol.types import Buffer, Direction, Tier
+
+TOTAL_IN_XML = 264
+TOTAL_WITH_FLAT_TRIM = 265
+
+
+def test_command_count():
+    assert len(xml_index.all_commands()) == TOTAL_WITH_FLAT_TRIM
+    from_xml = [c for c in xml_index.all_commands() if c.name != "FlatTrim"]
+    assert len(from_xml) == TOTAL_IN_XML
+
+
+def test_direction_split():
+    counts = collections.Counter(c.direction for c in xml_index.all_commands() if c.name != "FlatTrim")
+    assert counts[Direction.TO_DRONE] == 101
+    assert counts[Direction.FROM_DRONE] == 163
+    assert sum(counts.values()) == TOTAL_IN_XML
+
+
+def test_direction_fallback_count():
+    """31 commands state neither result nor triggered and fall back to the suffix."""
+    from xml.etree import ElementTree
+
+    directory = xml_index.xml_dir()
+    neither = 0
+    for name in xml_index.XML_FILES:
+        project = ElementTree.parse(directory / name).getroot()
+        for klass in project.iter("class"):
+            for cmd in klass.findall("cmd"):
+                comment = cmd.find("comment")
+                attrs = comment.attrib if comment is not None else {}
+                assert not ("result" in attrs and "triggered" in attrs), cmd.get("name")
+                if "result" not in attrs and "triggered" not in attrs:
+                    neither += 1
+    assert neither == 31
+
+
+def test_mixed_controller_class_is_not_decided_by_suffix():
+    # common.Controller holds one of each, which is why the suffix rule alone
+    # would be wrong there. Both state result/triggered, so it never applies.
+    assert xml_index.get("common.Controller.isPiloting").direction is Direction.TO_DRONE
+    assert xml_index.get("common.Controller.PeerStateChanged").direction is Direction.FROM_DRONE
+
+
+def test_buffers():
+    counts = collections.Counter(c.buffer for c in xml_index.all_commands() if c.name != "FlatTrim")
+    assert counts[Buffer.ACK] == 248
+    assert counts[Buffer.NON_ACK] == 15
+    assert counts[Buffer.HIGH_PRIO] == 1
+    high_prio = [c.full_name for c in xml_index.all_commands() if c.buffer is Buffer.HIGH_PRIO]
+    assert high_prio == ["ardrone3.Piloting.Emergency"]
+
+
+def test_deprecated_count():
+    assert sum(1 for c in xml_index.all_commands() if c.deprecated) == 39
+
+
+def test_expectation_block_count():
+    blocks = sum(1 for c in xml_index.all_commands() if c.expectations)
+    assert blocks == 82
+    assert all(c.direction is Direction.TO_DRONE for c in xml_index.all_commands() if c.expectations)
+
+
+def test_ids_are_unique_and_lookups_agree():
+    for command in xml_index.all_commands():
+        assert xml_index.by_ids(command.ids) is command
+        assert xml_index.get(command.full_name) is command
+    assert xml_index.get("ardrone3.Piloting.Nope") is None
+    assert xml_index.by_ids((9, 9, 9)) is None
+
+
+def test_known_ids_match_the_live_aircraft():
+    # Read off real traffic, see docs/notes/protocol.md.
+    assert xml_index.get("common.Common.AllStates").ids == (0, 4, 0)
+    assert xml_index.get("common.Settings.AllSettings").ids == (0, 2, 0)
+    assert xml_index.get("ardrone3.Piloting.TakeOff").ids == (1, 0, 1)
+    assert xml_index.get("ardrone3.Piloting.PCMD").ids == (1, 0, 2)
+
+
+def test_command_ids_have_gaps_and_document_order_is_not_id_order():
+    """Never infer an id from a command's position in the file."""
+    piloting = [c for c in xml_index.all_commands() if f"{c.project}.{c.klass}" == "ardrone3.Piloting"]
+    in_document_order = [c.ids[2] for c in piloting]
+    assert in_document_order != sorted(in_document_order)
+
+    # ardrone3.PilotingState runs 1..22 with no id 0, so counting is not enough.
+    state = [
+        c.ids[2] for c in xml_index.all_commands() if f"{c.project}.{c.klass}" == "ardrone3.PilotingState"
+    ]
+    assert len(state) < max(state) + 1
+
+
+def test_flat_trim_is_injected():
+    flat_trim = xml_index.get("ardrone3.Piloting.FlatTrim")
+    assert flat_trim is not None
+    assert flat_trim.ids == (1, 0, 0)
+    assert flat_trim.args == ()
+    assert flat_trim.buffer is Buffer.ACK
+    assert flat_trim.direction is Direction.TO_DRONE
+    assert flat_trim.tier is Tier.MOTION
+    assert flat_trim.support is None
+    assert "missing from Parrot's XML" in flat_trim.doc
+
+
+def test_commands_without_a_comment_keep_their_bare_text_as_doc():
+    bare = [
+        "common.CalibrationState.PitotCalibrationStateChanged",
+        "common.ARLibsVersionsState.ControllerLibARCommandsVersion",
+        "common.ARLibsVersionsState.SkyControllerLibARCommandsVersion",
+        "common.ARLibsVersionsState.DeviceLibARCommandsVersion",
+        "ardrone3.SettingsState.CPUID",
+        "ardrone3.MediaStreamingState.VideoStreamModeChanged",
+    ]
+    for name in bare:
+        command = xml_index.get(name)
+        assert command.title == ""
+        assert command.doc, name
+    assert xml_index.get("ardrone3.SettingsState.CPUID").doc == "Product main cpu id"
+
+
+def test_enum_values_are_positions_and_names_stay_verbatim():
+    band = xml_index.get("ardrone3.Network.WifiScan").args[0]
+    assert band.is_enum
+    assert [(m.name, m.value) for m in band.members] == [("2_4ghz", 0), ("5ghz", 1), ("all", 2)]
+    # Member names that are not valid identifiers must survive as strings.
+    framerate = xml_index.get("ardrone3.PictureSettings.VideoFramerate").args[0]
+    assert [m.name for m in framerate.members][:3] == ["24_FPS", "25_FPS", "30_FPS"]
+
+
+def test_only_name_and_type_on_args():
+    # Nothing in the parse may depend on an arg attribute Parrot does not set.
+    for command in xml_index.all_commands():
+        for arg in command.args:
+            assert arg.name and arg.type
+
+
+def test_search_ranks_the_command_above_the_prose():
+    results = xml_index.search("takeoff")
+    assert results
+    assert results[0].name == "TakeOff"
+    assert any(c.full_name == "ardrone3.PilotingState.FlyingStateChanged" for c in xml_index.search("flying"))
+    assert xml_index.search("   ") == []
+
+
+def test_myclass_spelling_also_loads(tmp_path, monkeypatch):
+    """pyparrot renamed <class> to <myclass>; a copy from it must still parse."""
+    source = xml_index.xml_dir()
+    for name in xml_index.XML_FILES:
+        text = (source / name).read_text()
+        text = text.replace("<class ", "<myclass ").replace("</class>", "</myclass>")
+        (tmp_path / name).write_text(text)
+
+    monkeypatch.setenv(xml_index.XML_DIR_ENV, str(tmp_path))
+    xml_index.index.cache_clear()
+    try:
+        assert len(xml_index.all_commands()) == TOTAL_WITH_FLAT_TRIM
+    finally:
+        monkeypatch.delenv(xml_index.XML_DIR_ENV)
+        xml_index.index.cache_clear()
+
+
+def test_missing_xml_dir_says_what_to_set(tmp_path, monkeypatch):
+    real = xml_index.xml_dir()
+    shutil.copy(real / "common.xml", tmp_path / "common.xml")  # one of the two, not both
+    monkeypatch.setenv(xml_index.XML_DIR_ENV, str(tmp_path))
+    xml_index.index.cache_clear()
+    try:
+        with pytest.raises(FileNotFoundError, match=xml_index.XML_DIR_ENV):
+            xml_index.all_commands()
+    finally:
+        monkeypatch.delenv(xml_index.XML_DIR_ENV)
+        xml_index.index.cache_clear()
+
+
+@pytest.mark.parametrize(
+    ("support", "expected"),
+    [
+        ("drones", True),
+        ("none", False),
+        ("0901;090c;090e", True),
+        ("0901;090e", False),
+        ("090c", True),
+        ("090c:4.3.0", True),  # firmware 4.7.1 clears a 4.3.0 minimum
+        ("090c:4.8.0", False),
+        ("0901:2.0.29;090c;090e", True),
+        ("0914:1.6.3;0919:1.6.3", False),
+        ("090e;090c:4.3.0", True),
+        (None, None),
+        ("", None),
+        ("   ", None),
+    ],
+)
+def test_supports_bebop2(support, expected):
+    assert xml_index.supports_bebop2(support) is expected
+
+
+def test_support_versions_compare_as_integers_not_strings():
+    # Lexically "2.0.29" < "2.0.3", which would make this unsupported.
+    assert xml_index.supports_bebop2("090c:2.0.29", firmware=(2, 0, 3)) is False
+    assert xml_index.supports_bebop2("090c:2.0.3", firmware=(2, 0, 29)) is True
+    assert xml_index.supports_bebop2("090c:10.0.0", firmware=(9, 0, 0)) is False
+
+
+def test_unknown_support_is_not_unsupported():
+    unknown = [c for c in xml_index.all_commands() if xml_index.supports_bebop2(c.support) is None]
+    # 30 commands carry a <comment> with no support attribute, 2 carry an empty
+    # one, 6 have no <comment> at all, and FlatTrim is hand-added.
+    assert len(unknown) == 39
+    assert xml_index.get("ardrone3.Piloting.FlatTrim") in unknown

tests/test_safety.py

@@ -0,0 +1,131 @@
+"""Tiering decides what needs arming, so a misfiled command is a safety bug."""
+
+import collections
+
+from mcbebop.protocol import xml_index
+from mcbebop.protocol.safety import ALWAYS_ALLOWED, tier_for
+from mcbebop.protocol.types import Direction, Tier
+
+MOTION = {
+    # The aircraft itself.
+    "ardrone3.Piloting.FlatTrim",
+    "ardrone3.Piloting.TakeOff",
+    "ardrone3.Piloting.PCMD",
+    "ardrone3.Piloting.Landing",
+    "ardrone3.Piloting.Emergency",
+    "ardrone3.Piloting.NavigateHome",
+    "ardrone3.Piloting.AutoTakeOffMode",
+    "ardrone3.Piloting.moveBy",
+    "ardrone3.Piloting.UserTakeOff",
+    "ardrone3.Piloting.Circle",
+    "ardrone3.Piloting.moveTo",
+    "ardrone3.Piloting.CancelMoveTo",
+    "ardrone3.Piloting.StartPilotedPOI",
+    "ardrone3.Piloting.StopPilotedPOI",
+    "ardrone3.Piloting.CancelMoveBy",
+    "ardrone3.Piloting.StartPilotedPOIV2",
+    "ardrone3.Animations.Flip",
+    # Running a stored flight plan moves the whole mission.
+    "common.Mavlink.Start",
+    "common.Mavlink.Pause",
+    "common.Mavlink.Stop",
+    # Calibration asks a powered aircraft to rotate.
+    "common.Calibration.MagnetoCalibration",
+    "common.Calibration.PitotCalibration",
+    "common.Animations.StartAnimation",
+    "common.Animations.StopAnimation",
+    "common.Animations.StopAllAnimations",
+}
+
+
+def test_motion_set_is_exactly_these():
+    tiered = {c.full_name for c in xml_index.all_commands() if c.tier is Tier.MOTION}
+    assert tiered == MOTION
+
+
+def test_every_motion_command_is_sendable():
+    for name in MOTION:
+        command = xml_index.get(name)
+        assert command is not None, name
+        assert command.direction is Direction.TO_DRONE, name
+
+
+def test_observe_is_the_two_dump_everything_commands():
+    sendable_observe = {
+        c.full_name
+        for c in xml_index.all_commands()
+        if c.tier is Tier.OBSERVE and c.direction is Direction.TO_DRONE
+    }
+    assert sendable_observe == {"common.Common.AllStates", "common.Settings.AllSettings"}
+
+
+def test_envelope_covers_the_limit_and_radio_classes():
+    envelope = {c.full_name for c in xml_index.all_commands() if c.tier is Tier.ENVELOPE}
+    for name in (
+        "ardrone3.PilotingSettings.MaxAltitude",
+        "ardrone3.PilotingSettings.NoFlyOverMaxDistance",
+        "ardrone3.SpeedSettings.MaxVerticalSpeed",
+        "ardrone3.GPSSettings.SetHome",
+        "ardrone3.Network.WifiScan",
+        "ardrone3.NetworkSettings.WifiSelection",
+        "common.Network.Disconnect",
+        "common.WifiSettings.OutdoorSetting",
+        "common.FlightPlanSettings.ReturnHomeOnDisconnect",
+        "common.Settings.Country",
+        "common.Settings.AutoCountry",
+        "common.Settings.Reset",
+        "common.Common.Reboot",
+        "common.Factory.Reset",
+        "common.OverHeat.SwitchOff",
+        "common.Charger.SetMaxChargeRate",
+        "common.Controller.isPiloting",
+    ):
+        assert name in envelope, name
+    assert len(envelope) == 44
+
+
+def test_config_is_the_harmless_remainder():
+    config = {c.full_name for c in xml_index.all_commands() if c.tier is Tier.CONFIG}
+    assert len(config) == 31
+    for name in (
+        "ardrone3.MediaRecord.Picture",
+        "ardrone3.MediaStreaming.VideoEnable",
+        "ardrone3.PictureSettings.VideoFramerate",
+        "ardrone3.Camera.Orientation",
+        "common.Common.CurrentDate",
+        "common.Settings.ProductName",
+        "common.Headlights.intensity",
+    ):
+        assert name in config, name
+    assert not any(
+        c.direction is Direction.FROM_DRONE for c in xml_index.all_commands() if c.tier is Tier.CONFIG
+    )
+
+
+def test_sendable_tiers_account_for_every_to_drone_command():
+    counts = collections.Counter(
+        c.tier for c in xml_index.all_commands() if c.direction is Direction.TO_DRONE
+    )
+    assert counts[Tier.MOTION] == 25
+    assert counts[Tier.ENVELOPE] == 44
+    assert counts[Tier.OBSERVE] == 2
+    assert counts[Tier.CONFIG] == 31
+    assert sum(counts.values()) == 102  # 101 in the XML plus the injected FlatTrim
+
+
+def test_events_are_observe():
+    for command in xml_index.all_commands():
+        if command.direction is Direction.FROM_DRONE:
+            assert command.tier is Tier.OBSERVE, command.full_name
+
+
+def test_always_allowed_are_the_two_that_end_a_flight():
+    assert frozenset({"ardrone3.Piloting.Landing", "ardrone3.Piloting.Emergency"}) == ALWAYS_ALLOWED
+    for name in ALWAYS_ALLOWED:
+        command = xml_index.get(name)
+        assert command.tier is Tier.MOTION  # locked by tier, exempted by name
+
+
+def test_tier_for_is_what_the_index_stamped():
+    for command in xml_index.all_commands():
+        assert tier_for(command) is command.tier