127 lines
5.3 KiB
Python
127 lines
5.3 KiB
Python
"""Tests for emby-mcp (run with emby-mcp/.venv: needs mcp 1.x FastMCP).
|
|
|
|
emby-mcp\\.venv\\Scripts\\python.exe -m pytest tests/test_emby_mcp.py -q
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import types
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("mcp.server.fastmcp")
|
|
pytest.importorskip("emby_mcp")
|
|
|
|
import httpx # noqa: E402
|
|
|
|
from emby_mcp import server as srv # noqa: E402
|
|
from emby_mcp.client import EmbyClient # noqa: E402
|
|
from emby_mcp.config import Settings # noqa: E402
|
|
|
|
SESSIONS = [
|
|
{"Id": "s-tv", "DeviceName": "Living Room TV", "Client": "Emby Theater", "SupportsRemoteControl": True},
|
|
{"Id": "s-srv", "DeviceName": "Tracearr Server", "Client": "Tracearr", "SupportsRemoteControl": False},
|
|
]
|
|
|
|
|
|
def _client(handler) -> tuple[EmbyClient, list[httpx.Request]]:
|
|
seen: list[httpx.Request] = []
|
|
|
|
def _record(request: httpx.Request) -> httpx.Response:
|
|
seen.append(request)
|
|
return handler(request)
|
|
|
|
client = EmbyClient(Settings(host="emby.local:8096", api_key="k"))
|
|
client._http = httpx.AsyncClient(base_url=client.settings.base_url,
|
|
transport=httpx.MockTransport(_record))
|
|
return client, seen
|
|
|
|
|
|
def _ctx(client: EmbyClient):
|
|
return types.SimpleNamespace(
|
|
request_context=types.SimpleNamespace(lifespan_context=srv.AppContext(emby=client)))
|
|
|
|
|
|
def test_base_url_defaults_to_http_and_accepts_ports_and_urls():
|
|
assert Settings(host="10.0.0.5:9096", api_key="k").base_url == "http://10.0.0.5:9096"
|
|
assert Settings(host="10.0.0.5", port=8096, api_key="k").base_url == "http://10.0.0.5:8096"
|
|
assert Settings(host="10.0.0.5:9096", port=443, api_key="k").base_url == "http://10.0.0.5:9096"
|
|
assert Settings(host="https://emby.example.org/", api_key="k").base_url == "https://emby.example.org"
|
|
assert Settings(host="emby.example.org", scheme="https", api_key="k").base_url == "https://emby.example.org"
|
|
|
|
|
|
def test_ids_cannot_reach_other_endpoints():
|
|
client, seen = _client(lambda r: httpx.Response(200, json={}))
|
|
|
|
async def run():
|
|
await client.get_user("../System/Restart")
|
|
await client.get_playback_info("123/Delete")
|
|
await client.play_on_session("s/../../System/Shutdown", ["1"])
|
|
await client.aclose()
|
|
|
|
asyncio.run(run())
|
|
paths = [r.url.raw_path.decode().split("?")[0] for r in seen]
|
|
assert paths == ["/Users/..%2FSystem%2FRestart", "/Items/123%2FDelete/PlaybackInfo",
|
|
"/Sessions/s%2F..%2F..%2FSystem%2FShutdown/Playing"]
|
|
|
|
|
|
def test_search_is_server_side():
|
|
client, seen = _client(lambda r: httpx.Response(200, json={"Items": [
|
|
{"Id": "1", "Name": "Stranger Things", "Type": "Series", "ProductionYear": 2016,
|
|
"RunTimeTicks": 30_000_000_000}]}))
|
|
hits = asyncio.run(srv.search_items(_ctx(client), "Stranger Things"))
|
|
params = seen[0].url.params
|
|
assert params["SearchTerm"] == "Stranger Things"
|
|
assert "Movie" in params["IncludeItemTypes"] and "Person" not in params["IncludeItemTypes"]
|
|
assert hits[0].name == "Stranger Things" and hits[0].year == 2016 and hits[0].runtime_minutes == 50
|
|
|
|
|
|
def test_get_item_uses_items_query():
|
|
client, seen = _client(lambda r: httpx.Response(200, json={"Items": [{"Id": "42", "Name": "X"}]}))
|
|
item = asyncio.run(srv.get_item(_ctx(client), "42"))
|
|
assert item["Name"] == "X"
|
|
assert seen[0].url.path == "/Items" and seen[0].url.params["Ids"] == "42"
|
|
empty, _ = _client(lambda r: httpx.Response(200, json={"Items": []}))
|
|
with pytest.raises(LookupError):
|
|
asyncio.run(srv.get_item(_ctx(empty), "nope"))
|
|
|
|
|
|
def test_play_item_targets_a_controllable_session():
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path == "/Sessions":
|
|
return httpx.Response(200, json=SESSIONS)
|
|
return httpx.Response(204)
|
|
|
|
client, seen = _client(handler)
|
|
result = asyncio.run(srv.play_item(_ctx(client), "42"))
|
|
assert result["session_id"] == "s-tv"
|
|
play = seen[-1]
|
|
assert play.method == "POST" and play.url.path == "/Sessions/s-tv/Playing"
|
|
assert play.url.params["ItemIds"] == "42" and play.url.params["PlayCommand"] == "PlayNow"
|
|
|
|
with pytest.raises(ValueError, match="No matching"):
|
|
asyncio.run(srv.play_item(_ctx(client), "42", device="Kitchen"))
|
|
with pytest.raises(ValueError, match="No matching"):
|
|
asyncio.run(srv.play_item(_ctx(client), "42", session_id="s-srv"))
|
|
|
|
|
|
def test_list_devices_unwraps_items():
|
|
client, _ = _client(lambda r: httpx.Response(200, json={"Items": [{"Id": "d", "Name": "Phone"}]}))
|
|
assert asyncio.run(srv.list_devices(_ctx(client))) == [
|
|
{"id": "d", "name": "Phone", "app": None, "last_user": None, "last_active": None}]
|
|
|
|
|
|
def test_playback_url_has_no_credentials():
|
|
client, _ = _client(lambda r: httpx.Response(200, json={"Items": [
|
|
{"Id": "7", "Name": "Song", "MediaType": "Audio"}]}))
|
|
out = asyncio.run(srv.playback_url(_ctx(client), "7"))
|
|
assert out["url"] == "http://emby.local:8096/Audio/7/stream?static=true"
|
|
assert "api_key" not in out["url"] and "k" not in out["url"].split("/")[-1]
|
|
|
|
|
|
def test_tool_annotations():
|
|
tools = {t.name: t for t in asyncio.run(srv.mcp.list_tools())}
|
|
assert tools["play_item"].annotations.readOnlyHint is False
|
|
for name in ("search_items", "get_item", "list_sessions", "playback_url"):
|
|
assert tools[name].annotations.readOnlyHint is True, name
|