213 lines
7.6 KiB
Python
213 lines
7.6 KiB
Python
"""Tests for MCP servers — tools, resources, prompts via the MCP protocol."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from fastmcp import Client
|
|
|
|
|
|
def _client(module: str) -> Client:
|
|
return Client({"mcpServers": {"default": {
|
|
"command": "uv",
|
|
"args": ["run", "python", "-m", module],
|
|
}}})
|
|
|
|
|
|
def _parse_result(result) -> any:
|
|
"""Parse a fastmcp CallToolResult into Python data."""
|
|
if hasattr(result, "content"):
|
|
texts = [c.text for c in result.content if hasattr(c, "text")]
|
|
elif isinstance(result, list):
|
|
texts = [c.text for c in result if hasattr(c, "text")]
|
|
else:
|
|
return result
|
|
|
|
if not texts:
|
|
return None
|
|
combined = texts[0] if len(texts) == 1 else "[" + ",".join(texts) + "]"
|
|
try:
|
|
return json.loads(combined)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return combined
|
|
|
|
|
|
class TestSharedServer:
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools(self):
|
|
async with _client("mcp_servers.shared") as c:
|
|
tools = await c.list_tools()
|
|
names = [t.name for t in tools]
|
|
assert "get_flight_status" in names
|
|
assert "get_flight_details" in names
|
|
assert "get_irregular_ops" in names
|
|
assert "get_route_weather" in names
|
|
assert "get_hub_forecasts" in names
|
|
assert "get_airport_status" in names
|
|
assert "get_airport_congestion" in names
|
|
assert "get_maintenance_flags" in names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flight_status_found(self):
|
|
async with _client("mcp_servers.shared") as c:
|
|
result = await c.call_tool("get_flight_status", {"flight_id": "UA432"})
|
|
data = _parse_result(result)
|
|
assert data["flight_id"] == "UA432"
|
|
assert data["status"] == "DELAYED"
|
|
assert data["delay_minutes"] == 55
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_flight_status_not_found(self):
|
|
async with _client("mcp_servers.shared") as c:
|
|
result = await c.call_tool("get_flight_status", {"flight_id": "FAKE999"})
|
|
data = _parse_result(result)
|
|
assert "error" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_irregular_ops(self):
|
|
async with _client("mcp_servers.shared") as c:
|
|
result = await c.call_tool("get_irregular_ops", {"hub": "ORD"})
|
|
data = _parse_result(result)
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 4
|
|
for f in data:
|
|
assert f["irrop_type"] in ("DELAYED", "CANCELLED", "DIVERTED")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_maintenance_flags(self):
|
|
async with _client("mcp_servers.shared") as c:
|
|
result = await c.call_tool("get_maintenance_flags", {"aircraft_tail": "N78501"})
|
|
data = _parse_result(result)
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 1
|
|
assert "system" in data[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_maintenance_flags_empty(self):
|
|
async with _client("mcp_servers.shared") as c:
|
|
result = await c.call_tool("get_maintenance_flags", {"aircraft_tail": "FAKE"})
|
|
data = _parse_result(result)
|
|
# Empty list may come back as [] or as "[]" string
|
|
if isinstance(data, list):
|
|
assert len(data) == 0
|
|
else:
|
|
assert data == "[]" or data is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_prompts(self):
|
|
async with _client("mcp_servers.shared") as c:
|
|
prompts = await c.list_prompts()
|
|
names = [p.name for p in prompts]
|
|
assert "delay_explainer" in names
|
|
|
|
|
|
class TestOpsServer:
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools(self):
|
|
async with _client("mcp_servers.ops") as c:
|
|
tools = await c.list_tools()
|
|
names = [t.name for t in tools]
|
|
assert "get_crew_notes" in names
|
|
assert "get_crew_duty_status" in names
|
|
assert "get_pending_rebookings" in names
|
|
assert "generate_narrative" in names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crew_notes(self):
|
|
async with _client("mcp_servers.ops") as c:
|
|
result = await c.call_tool("get_crew_notes", {"flight_id": "UA432"})
|
|
data = _parse_result(result)
|
|
assert isinstance(data, list)
|
|
assert len(data) > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crew_duty_status(self):
|
|
async with _client("mcp_servers.ops") as c:
|
|
result = await c.call_tool(
|
|
"get_crew_duty_status",
|
|
{"crew_ids": ["CR-1001", "CR-1003"]}
|
|
)
|
|
data = _parse_result(result)
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
for c_data in data:
|
|
assert "hours_until_limit" in c_data
|
|
assert "at_risk" in c_data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crew_at_risk_detection(self):
|
|
async with _client("mcp_servers.ops") as c:
|
|
result = await c.call_tool(
|
|
"get_crew_duty_status",
|
|
{"crew_ids": ["CR-1003"]}
|
|
)
|
|
data = _parse_result(result)
|
|
crew = data[0] if isinstance(data, list) else data
|
|
assert crew["at_risk"] is True
|
|
assert crew["hours_until_limit"] <= 2.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_rebookings(self):
|
|
async with _client("mcp_servers.ops") as c:
|
|
result = await c.call_tool("get_pending_rebookings", {"hub": "ORD"})
|
|
data = _parse_result(result)
|
|
assert isinstance(data, list)
|
|
assert len(data) > 0
|
|
urgencies = [r["urgency"] for r in data]
|
|
assert urgencies[0] == "HIGH"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_resources(self):
|
|
async with _client("mcp_servers.ops") as c:
|
|
resources = await c.list_resources()
|
|
uris = [str(r.uri) for r in resources]
|
|
assert "ops://crew/roster" in uris
|
|
assert "ops://handover/latest" in uris
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_prompts(self):
|
|
async with _client("mcp_servers.ops") as c:
|
|
prompts = await c.list_prompts()
|
|
names = [p.name for p in prompts]
|
|
assert "handover_brief" in names
|
|
|
|
|
|
class TestPassengerServer:
|
|
@pytest.mark.asyncio
|
|
async def test_list_tools(self):
|
|
async with _client("mcp_servers.passenger") as c:
|
|
tools = await c.list_tools()
|
|
names = [t.name for t in tools]
|
|
assert "generate_notification" in names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_notification(self):
|
|
async with _client("mcp_servers.passenger") as c:
|
|
result = await c.call_tool("generate_notification", {
|
|
"context": {
|
|
"flight_id": "UA432",
|
|
"origin": "ORD",
|
|
"destination": "SFO",
|
|
"status": "DELAYED",
|
|
"delay_minutes": 55,
|
|
"delay_cause": "WEATHER",
|
|
"gate": "H14",
|
|
}
|
|
})
|
|
data = _parse_result(result)
|
|
# Response is JSON with text + provider
|
|
if isinstance(data, dict):
|
|
assert "text" in data
|
|
assert "provider" in data
|
|
text = data["text"]
|
|
else:
|
|
text = str(data)
|
|
assert "UA432" in text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_prompts(self):
|
|
async with _client("mcp_servers.passenger") as c:
|
|
prompts = await c.list_prompts()
|
|
names = [p.name for p in prompts]
|
|
assert "passenger_notification" in names
|