Extract MockSkyWalker1 to shared mock_device.py (used by both unit tests and server lifespan). Server checks SKYWALKER_MOCK env var at startup — when set, uses mock device instead of USB hardware, enabling full MCP transport testing via claude -p without the dongle connected. Verified: 64 unit tests pass, claude -p integration tests exercise identify_frequency, get_device_status, and sweep_spectrum through the complete JSON-RPC pipeline.
48 lines
1.2 KiB
Python
48 lines
1.2 KiB
Python
"""Shared fixtures for skywalker-mcp tests."""
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
import skywalker_mcp.server as srv
|
|
from skywalker_mcp.mock_device import MockSkyWalker1
|
|
from skywalker_mcp.server import DeviceBridge
|
|
|
|
|
|
class MockContext:
|
|
"""Minimal mock of FastMCP Context for direct tool function calls.
|
|
|
|
Provides the bridge via request_context.lifespan_context["bridge"],
|
|
matching what _get_bridge(ctx) expects.
|
|
"""
|
|
|
|
def __init__(self, bridge: DeviceBridge):
|
|
self.request_context = MagicMock()
|
|
self.request_context.lifespan_context = {"bridge": bridge}
|
|
self._progress = []
|
|
|
|
async def report_progress(self, current, total):
|
|
self._progress.append((current, total))
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_device():
|
|
"""Provide a fresh MockSkyWalker1 instance."""
|
|
return MockSkyWalker1()
|
|
|
|
|
|
@pytest.fixture
|
|
def bridge(mock_device):
|
|
"""Provide a DeviceBridge wrapping the mock device."""
|
|
b = DeviceBridge(mock_device)
|
|
srv._bridge = b
|
|
yield b
|
|
b.cancel_motor_watchdog()
|
|
srv._bridge = None
|
|
|
|
|
|
@pytest.fixture
|
|
def ctx(bridge):
|
|
"""Provide a MockContext wired to the bridge."""
|
|
return MockContext(bridge)
|