rsp2k
ad81cc6e38
read_log fetches the live log over FTP and filters by tag, severity and message before it returns anything, because half an hour of uptime is five thousand entries. It excludes shp_usbmode by default, a five-second USB poll that can be a third of the log, and says in the result that it did so. Only the current file is reachable: the archives the drone rotates into are a previous owner's sessions and carry network and location data, which is a decision for a person, not a tool. bebop://commands is the resource worth having, because it needs no drone: the whole catalogue is readable with the aircraft switched off. The ones that do need a link return an explanation rather than raising, since a resource that errors looks broken while one that explains itself is empty for a reason. bebop://state also reports telemetry age, because the aircraft accepts a second controller and silently redirects telemetry to it, so a connected flag proves nothing. The log says nothing resolvable about ARSDK, measured rather than assumed: no command ids appear anywhere in it, and the COMMANDS function names match arsdk-xml for 4 of 11 symbols, 3.5% of the tag. So nothing here resolves them, and read_log says where to look instead. _host moves into tools/_common as drone_host now that three tool modules need it, and head/tail in the shell allow-list gains the reason they must take a file argument: ld.so.preload makes SIGPIPE fatal, so a pipe into head writes a crash report to the aircraft's flash.
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", "read_log", "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"})
|