Add software watchdog and timeout protection for all I2C/USB paths (firmware v3.05.0)
Motor watchdog: asyncio background task auto-halts motor after 30s of
continuous drive, fires even if LLM client disconnects. Integrated into
move_dish (start on continuous, cancel on halt/goto/gotox) and lifespan
teardown.
Test suite: 64 tests covering all 17 MCP tools — device status, spectrum
sweep validation, tune/blind-scan boundary checks, motor safety (stepped,
continuous opt-in, watchdog lifecycle), jog/store limits, LNB/I2C, TS
capture, frequency identification, and path traversal protection. Uses
MockSkyWalker1 + MockContext for direct async function testing without
USB hardware.
Fixes: FastMCP 2.x description→instructions constructor change,
parents[4] path resolution for tools directory import.
2026-02-17 15:08:52 -07:00
|
|
|
"""Shared fixtures for skywalker-mcp tests."""
|
|
|
|
|
|
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
|
|
import skywalker_mcp.server as srv
|
2026-02-17 15:23:05 -07:00
|
|
|
from skywalker_mcp.mock_device import MockSkyWalker1
|
Add software watchdog and timeout protection for all I2C/USB paths (firmware v3.05.0)
Motor watchdog: asyncio background task auto-halts motor after 30s of
continuous drive, fires even if LLM client disconnects. Integrated into
move_dish (start on continuous, cancel on halt/goto/gotox) and lifespan
teardown.
Test suite: 64 tests covering all 17 MCP tools — device status, spectrum
sweep validation, tune/blind-scan boundary checks, motor safety (stepped,
continuous opt-in, watchdog lifecycle), jog/store limits, LNB/I2C, TS
capture, frequency identification, and path traversal protection. Uses
MockSkyWalker1 + MockContext for direct async function testing without
USB hardware.
Fixes: FastMCP 2.x description→instructions constructor change,
parents[4] path resolution for tools directory import.
2026-02-17 15:08:52 -07:00
|
|
|
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)
|