94 lines
3.9 KiB
Python
94 lines
3.9 KiB
Python
"""Tests for javis-slave-mcp (needs mcp 1.x with FastMCP — run with javis-slave-mcp/.venv).
|
|
|
|
javis-slave-mcp\\.venv\\Scripts\\python.exe -m pytest tests/test_slave_mcp.py -q
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("mcp.server.fastmcp")
|
|
|
|
import httpx # noqa: E402
|
|
|
|
MCP_SRC = Path(__file__).resolve().parents[1] / "javis-slave-mcp" / "src"
|
|
if str(MCP_SRC) not in sys.path:
|
|
sys.path.insert(0, str(MCP_SRC))
|
|
|
|
from mcp.server.fastmcp.exceptions import ToolError # noqa: E402
|
|
|
|
from javis_slave_mcp import server as srv # noqa: E402
|
|
|
|
REAL_ID = "3f6c1a7e-0000-4000-8000-000000000001"
|
|
|
|
|
|
def test_env_parsing_skips_bad_entries():
|
|
slaves = srv._parse_env_devices(
|
|
" a = 10.0.0.1:8766 , bad, b=tok-123@host:9000, c=host2,,d=:1", "default-tok")
|
|
assert set(slaves) == {"a", "b", "c"}
|
|
assert (slaves["a"].host, slaves["a"].port, slaves["a"].token) == ("10.0.0.1", 8766, "default-tok")
|
|
assert (slaves["b"].host, slaves["b"].port, slaves["b"].token) == ("host", 9000, "tok-123")
|
|
assert (slaves["c"].host, slaves["c"].port) == ("host2", srv.DEFAULT_PORT)
|
|
assert "tok-123" not in repr(slaves["b"])
|
|
|
|
|
|
def _install(handler, token: str = "") -> list[httpx.Request]:
|
|
seen: list[httpx.Request] = []
|
|
|
|
def _record(request: httpx.Request) -> httpx.Response:
|
|
seen.append(request)
|
|
return handler(request)
|
|
|
|
client = srv.SlaveClient(device_id="localhost", host="127.0.0.1", token=token)
|
|
client._http = httpx.AsyncClient(transport=httpx.MockTransport(_record),
|
|
base_url=client.base_url(), headers=client.headers())
|
|
srv._slaves = {"localhost": client}
|
|
return seen
|
|
|
|
|
|
def _status_json() -> dict:
|
|
return {"device_id": REAL_ID, "device_name": "Kueche", "paired": True,
|
|
"engine_running": True, "audio_state": "idle"}
|
|
|
|
|
|
def test_tools_run_inside_the_server_loop_and_resolve_real_id():
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/mcp/v1/status":
|
|
return httpx.Response(200, json=_status_json())
|
|
if request.url.path == f"/mcp/v1/master/slaves/{REAL_ID}/ip":
|
|
return httpx.Response(200, json={"device_id": REAL_ID, "ip_address": "10.0.0.7"})
|
|
return httpx.Response(404, json={"detail": "nope"})
|
|
|
|
seen = _install(handler, token="secret-token")
|
|
# FastMCP awaits the tool on its running loop — the old asyncio.run() failed here.
|
|
result = asyncio.run(srv.mcp.call_tool("slave_get_ip_address", {"device_id": "localhost"}))
|
|
assert "10.0.0.7" in str(result)
|
|
assert [r.url.path for r in seen] == ["/mcp/v1/status", f"/mcp/v1/master/slaves/{REAL_ID}/ip"]
|
|
assert all(r.headers["X-Javis-Control"] == "1" for r in seen)
|
|
assert all(r.headers["Authorization"] == "Bearer secret-token" for r in seen)
|
|
|
|
|
|
def test_unauthorized_becomes_tool_error():
|
|
_install(lambda request: httpx.Response(401, json={"detail": "unauthorized"}))
|
|
with pytest.raises(ToolError, match="control token"):
|
|
asyncio.run(srv.mcp.call_tool("slave_get_status", {"device_id": "localhost"}))
|
|
|
|
|
|
def test_unknown_device_is_tool_error():
|
|
_install(lambda request: httpx.Response(200, json=_status_json()))
|
|
with pytest.raises(ToolError, match="not found"):
|
|
asyncio.run(srv.mcp.call_tool("slave_get_status", {"device_id": "nope"}))
|
|
|
|
|
|
def test_tool_annotations():
|
|
tools = {t.name: t for t in asyncio.run(srv.mcp.list_tools())}
|
|
for name in ("slave_shutdown", "slave_restart_system", "slave_sleep"):
|
|
assert tools[name].annotations.destructiveHint is True, name
|
|
for name in ("slave_get_status", "slave_get_volume", "slave_list_devices"):
|
|
assert tools[name].annotations.readOnlyHint is True, name
|
|
assert tools["slave_set_volume"].annotations.idempotentHint is True
|
|
assert tools["slave_increase_volume"].annotations.idempotentHint is False
|