Keep every element of a list-shaped event
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.
This commit is contained in:
@@ -0,0 +1,52 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user