rsp2k
607d14979d
connect/disconnect/status, list_commands/command_info, the generic send_command, get_state/watch_state/preflight_check, camera snapshot and record, FTP listing and fetch, the read-only shell, and arm/disarm. The safety gate is tested exhaustively rather than by sample: every motion and envelope command is asserted to refuse while unarmed, and Landing and Emergency are asserted to work without arming, because refusing to bring down an airborne aircraft is the more dangerous answer. preflight_check reads the per-element sensor keys, so a single failing self-test is visible instead of being hidden by whichever arrived last.
132 lines
5.3 KiB
Python
132 lines
5.3 KiB
Python
"""The tools layer, driven through a real MCP client against the simulator."""
|
|
|
|
import pytest
|
|
from fastmcp import Client
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
from mcbebop.protocol import xml_index
|
|
from mcbebop.protocol.safety import ALWAYS_ALLOWED
|
|
from mcbebop.protocol.types import Direction, Tier
|
|
from mcbebop.server import build_server
|
|
|
|
|
|
def data(result):
|
|
return result.structured_content if result.structured_content is not None else result.data
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
async with Client(build_server()) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
async def connected(client):
|
|
await client.call_tool("connect", {"target": "sim"})
|
|
yield client
|
|
await client.call_tool("disconnect", {})
|
|
|
|
|
|
async def test_every_tool_is_registered(client):
|
|
names = {t.name for t in await client.list_tools()}
|
|
assert names == {
|
|
"connect", "disconnect", "connection_status", "list_commands", "command_info",
|
|
"send_command", "get_state", "watch_state", "preflight_check", "camera_snapshot",
|
|
"camera_record", "list_files", "fetch_file", "shell_read", "arm", "disarm",
|
|
} # fmt: skip
|
|
|
|
|
|
async def test_tools_refuse_before_connecting(client):
|
|
calls = (
|
|
("get_state", {}),
|
|
("preflight_check", {}),
|
|
("send_command", {"name": "common.Common.AllStates"}),
|
|
)
|
|
for tool, args in calls:
|
|
with pytest.raises(ToolError, match=r"[Nn]ot connected"):
|
|
await client.call_tool(tool, args)
|
|
|
|
|
|
async def test_connect_reports_live_telemetry(connected):
|
|
info = data(await connected.call_tool("connection_status", {}))
|
|
assert info["connected"] and info["target"] == "sim"
|
|
assert info["telemetry_keys"] > 20, "the drone's state burst should have arrived"
|
|
assert info["last_update_age"] is not None and info["last_update_age"] < 10
|
|
assert info["undecodable_events"] == 0, "every event the simulator sends should decode"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"name",
|
|
[c.full_name for c in xml_index.all_commands()
|
|
if c.direction is Direction.TO_DRONE and c.tier in (Tier.MOTION, Tier.ENVELOPE)
|
|
and c.full_name not in ALWAYS_ALLOWED],
|
|
) # fmt: skip
|
|
async def test_every_locked_command_is_refused_unarmed(connected, name):
|
|
"""The gate must hold for all of them, not just the obvious ones."""
|
|
with pytest.raises(ToolError, match="locked"):
|
|
await connected.call_tool("send_command", {"name": name})
|
|
|
|
|
|
@pytest.mark.parametrize("name", sorted(ALWAYS_ALLOWED))
|
|
async def test_landing_and_emergency_are_never_locked(connected, name):
|
|
"""Refusing to bring down an airborne aircraft is the more dangerous answer."""
|
|
result = data(await connected.call_tool("send_command", {"name": name}))
|
|
assert result["acked"]
|
|
|
|
|
|
async def test_arming_unlocks_and_disarming_relocks(connected):
|
|
takeoff = "ardrone3.Piloting.TakeOff"
|
|
with pytest.raises(ToolError, match="locked"):
|
|
await connected.call_tool("send_command", {"name": takeoff})
|
|
|
|
armed = data(await connected.call_tool("arm", {"reason": "test", "minutes": 1}))
|
|
assert armed["armed"] and armed["expires_at"]
|
|
assert data(await connected.call_tool("send_command", {"name": takeoff}))["acked"]
|
|
|
|
await connected.call_tool("disarm", {})
|
|
with pytest.raises(ToolError, match="locked"):
|
|
await connected.call_tool("send_command", {"name": takeoff})
|
|
|
|
|
|
async def test_events_cannot_be_sent(connected):
|
|
with pytest.raises(ToolError, match="event the drone emits"):
|
|
await connected.call_tool("send_command", {"name": "common.CommonState.BatteryStateChanged"})
|
|
|
|
|
|
async def test_unknown_command_suggests_alternatives(connected):
|
|
with pytest.raises(ToolError, match="No command named"):
|
|
await connected.call_tool("send_command", {"name": "ardrone3.Piloting.TakeOf"})
|
|
|
|
|
|
async def test_preflight_sees_a_failing_sensor(connected):
|
|
"""The simulator fails its magnetometer; a flat telemetry store would hide it."""
|
|
report = data(await connected.call_tool("preflight_check", {}))
|
|
sensors = next(c for c in report["checks"] if c["name"] == "sensor self-tests")
|
|
assert sensors["ok"] is False and "magnetometer" in sensors["detail"]
|
|
assert report["ready"] is False
|
|
assert any("magnetometer" in b for b in report["blocking"])
|
|
|
|
|
|
async def test_command_info_carries_the_confirming_event(connected):
|
|
info = data(await connected.call_tool("command_info", {"name": "ardrone3.MediaStreaming.VideoEnable"}))
|
|
assert info["ids"] == [1, 21, 0]
|
|
assert info["args"][0]["name"] == "enable"
|
|
assert any("VideoEnableChanged" in c for c in info["confirmed_by"])
|
|
|
|
|
|
async def test_list_commands_filters(connected):
|
|
args = {"direction": "to_drone", "bebop2_only": False}
|
|
sendable = data(await connected.call_tool("list_commands", args))
|
|
sendable = sendable.get("result", sendable)
|
|
# 101 in Parrot's XML, plus FlatTrim which the firmware accepts but the XML omits.
|
|
assert len(sendable) == 102
|
|
motion = data(await connected.call_tool("list_commands", {"tier": "motion", "bebop2_only": False}))
|
|
motion = motion.get("result", motion)
|
|
assert all(c["tier"] == "motion" for c in motion)
|
|
|
|
|
|
async def test_file_tools_reject_the_simulator(connected):
|
|
"""The simulator has no filesystem; saying so beats a confusing timeout."""
|
|
with pytest.raises(ToolError, match="simulator has no filesystem"):
|
|
await connected.call_tool("list_files", {"area": "media"})
|