rsp2k
6f2c36936e
17 events are MAP_ITEM or LIST_ITEM: they arrive once per element, all carrying the same argument names, so a flat store kept only whichever landed last. SensorsStatesListChanged reports six sensors that way, and the simulator's deliberate magnetometer fault was invisible through state() as a result, which is exactly the fact preflight_check needs to see. Entries are now keyed Command[element]_arg. Parrot marks these events but never names the key; it is the first argument by convention.
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
"""List-shaped events must not overwrite each other.
|
|
|
|
MAP_ITEM and LIST_ITEM events arrive once per element with identical argument
|
|
names. A flat store keeps only the last, which silently hides exactly the kind
|
|
of fact a preflight check looks for.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from mcbebop.arsdk.session import DroneSession
|
|
from mcbebop.protocol import xml_index
|
|
from mcbebop.sim import FakeBebop
|
|
|
|
SENSORS = "common.CommonState.SensorsStatesListChanged"
|
|
|
|
|
|
def test_list_events_are_indexed():
|
|
listed = [c for c in xml_index.all_commands() if c.list_key]
|
|
assert len(listed) == 17, "both XML files together define 17 MAP_ITEM/LIST_ITEM events"
|
|
assert xml_index.get(SENSORS).list_key == "sensorName"
|
|
# A scalar event must not be treated as a list.
|
|
assert xml_index.get("common.CommonState.BatteryStateChanged").list_key is None
|
|
|
|
|
|
def test_every_sensor_survives_and_a_fault_is_visible():
|
|
async def run():
|
|
with FakeBebop() as sim:
|
|
session = DroneSession(ip=sim.host, discovery_port=sim.discovery_port)
|
|
await session.connect()
|
|
await session.request_full_state()
|
|
await asyncio.sleep(2.0)
|
|
values = session.values()
|
|
await session.disconnect()
|
|
return values
|
|
|
|
values = asyncio.run(run())
|
|
sensors = {
|
|
key.split("[", 1)[1].split("]", 1)[0]: value
|
|
for key, value in values.items()
|
|
if key.startswith("SensorsStatesListChanged[") and key.endswith("_sensorState")
|
|
}
|
|
assert len(sensors) == 6, f"expected all six self-tests, got {sorted(sensors)}"
|
|
faults = sorted(name for name, ok in sensors.items() if not ok)
|
|
assert faults == ["magnetometer"], "the simulator's deliberate fault must be reachable"
|
|
|
|
|
|
@pytest.mark.parametrize("name", [SENSORS, "common.CommonState.MassStorageInfoStateListChanged"])
|
|
def test_list_key_is_the_first_argument(name):
|
|
spec = xml_index.get(name)
|
|
assert spec.list_key == spec.args[0].name
|