75 lines
1.9 KiB
Python
75 lines
1.9 KiB
Python
|
"""Shared fixtures for birdcage-mcp tests.
|
||
|
|
|
||
|
|
Uses FastMCP's run_server_async to spin up a real MCP server backed by
|
||
|
|
DemoDevice + DemoCraftClient. No serial hardware, no subprocesses.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import json
|
||
|
|
from contextlib import asynccontextmanager
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from birdcage.demo import DemoCraftClient, DemoDevice
|
||
|
|
from fastmcp import Client, FastMCP
|
||
|
|
from fastmcp.client.transports import StreamableHttpTransport
|
||
|
|
from fastmcp.utilities.tests import run_server_async
|
||
|
|
|
||
|
|
from birdcage_mcp.state import BirdcageState
|
||
|
|
|
||
|
|
|
||
|
|
@asynccontextmanager
|
||
|
|
async def _test_lifespan(server: FastMCP):
|
||
|
|
"""Test lifespan that pre-connects a DemoDevice."""
|
||
|
|
device = DemoDevice()
|
||
|
|
device.connect()
|
||
|
|
device.initialize()
|
||
|
|
|
||
|
|
state = BirdcageState(
|
||
|
|
device=device,
|
||
|
|
craft_client=DemoCraftClient(),
|
||
|
|
demo_mode=True,
|
||
|
|
serial_port="/dev/demo",
|
||
|
|
firmware_name="g2",
|
||
|
|
connected=True,
|
||
|
|
)
|
||
|
|
yield state
|
||
|
|
|
||
|
|
|
||
|
|
def _build_server() -> FastMCP:
|
||
|
|
"""Build a FastMCP server with all tools registered."""
|
||
|
|
from birdcage_mcp import prompts, resources
|
||
|
|
from birdcage_mcp.tools import (
|
||
|
|
connection,
|
||
|
|
console,
|
||
|
|
movement,
|
||
|
|
satellite,
|
||
|
|
signal,
|
||
|
|
system,
|
||
|
|
)
|
||
|
|
|
||
|
|
mcp = FastMCP("birdcage-test", lifespan=_test_lifespan)
|
||
|
|
connection.register(mcp)
|
||
|
|
movement.register(mcp)
|
||
|
|
signal.register(mcp)
|
||
|
|
system.register(mcp)
|
||
|
|
satellite.register(mcp)
|
||
|
|
console.register(mcp)
|
||
|
|
resources.register(mcp)
|
||
|
|
prompts.register(mcp)
|
||
|
|
return mcp
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
async def mcp_client():
|
||
|
|
"""Yield a connected MCP client backed by DemoDevice."""
|
||
|
|
server = _build_server()
|
||
|
|
async with (
|
||
|
|
run_server_async(server) as url,
|
||
|
|
Client(StreamableHttpTransport(url)) as client,
|
||
|
|
):
|
||
|
|
yield client
|
||
|
|
|
||
|
|
|
||
|
|
def parse_result(result) -> dict:
|
||
|
|
"""Extract the dict from a tool call result."""
|
||
|
|
return json.loads(result.content[0].text)
|