Ryan Malloy
5a26dce075
Core library published as winegard-birdcage (import stays `birdcage`). MCP server renamed birdcage-mcp → mcbirdcage, matching mcserial pattern. Both packages live on PyPI — `uvx mcbirdcage` works out of the box.
75 lines
1.9 KiB
Python
75 lines
1.9 KiB
Python
"""Shared fixtures for mcbirdcage 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 mcbirdcage.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 mcbirdcage import prompts, resources
|
|
from mcbirdcage.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)
|