Spaces:
Running
Running
File size: 6,255 Bytes
93a887a fd44cb0 52c1e0f 93a887a 52c1e0f fd44cb0 93a887a fd44cb0 93a887a fd44cb0 93a887a fd44cb0 93a887a fd44cb0 52c1e0f 93a887a effc0fc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | """Protocol-level discovery, schema, dispatch, and transport tests."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, cast
import pytest
from mcp import Client
import spotify_mcp_server.server as server_module
from spotify_mcp_server.server import create_server, main, mcp
from spotify_mcp_server.tools.common import ToolResponse
from spotify_mcp_server.tools.service import SpotifyService
pytestmark = pytest.mark.anyio
TOOL_NAMES = [
"search_catalog",
"get_item",
"player_status",
"player_control",
"playlist_read",
"playlist_modify",
"library_read",
"library_modify",
"listening_activity",
]
def load_snapshot() -> object:
path = Path(__file__).resolve().parents[1] / "schemas" / "tools.json"
return json.loads(path.read_text(encoding="utf-8"))
async def test_server_discovers_current_protocol_and_exact_tool_catalog() -> None:
async with Client(mcp) as client:
assert client.protocol_version == "2026-07-28"
assert client.server_info is not None
assert client.server_info.name == "Spotify MCP Server"
tools = await client.list_tools()
assert [tool.name for tool in tools.tools] == TOOL_NAMES
async def test_discovery_exposes_strict_structured_contracts_and_annotations() -> None:
tools = await mcp.list_tools()
for tool in tools:
assert tool.description
request_name = tool.input_schema["properties"]["request"]["$ref"].rsplit("/", 1)[-1]
request_schema = tool.input_schema["$defs"][request_name]
assert request_schema["additionalProperties"] is False
assert tool.output_schema is not None
assert tool.output_schema["additionalProperties"] is False
assert tool.annotations and tool.annotations.open_world_hint is True
async def test_committed_schema_snapshot_matches_discovery() -> None:
actual = [
tool.model_dump(by_alias=True, exclude_none=True, mode="json")
for tool in await mcp.list_tools()
]
assert actual == load_snapshot()
class RecordingService:
def __init__(self) -> None:
self.called: list[str] = []
def __getattr__(self, name: str) -> Any:
async def call(_: object) -> ToolResponse:
self.called.append(name)
return ToolResponse(status="ok", data={"tool": name})
return call
async def test_every_registered_handler_validates_and_dispatches() -> None:
service = RecordingService()
server = create_server(cast(SpotifyService, service))
payloads = {
"search_catalog": {"query": "focus", "types": ["track"]},
"get_item": {"items": [{"value": "spotify:track:t"}]},
"player_status": {},
"player_control": {"actions": [{"action": "pause_playback"}]},
"playlist_read": {"requests": [{"operation": "list_current_playlists"}]},
"playlist_modify": {"actions": [{"action": "create_playlist", "name": "x"}]},
"library_read": {"requests": [{"operation": "list_saved", "type": "track"}]},
"library_modify": {
"actions": [
{
"action": "save",
"items": [{"value": "spotify:track:t"}],
}
]
},
"listening_activity": {},
}
async with Client(server) as client:
for name, request in payloads.items():
response = await client.call_tool(name, {"request": request})
assert response.is_error is False
assert response.structured_content is not None
assert response.structured_content["status"] == "ok"
assert service.called == TOOL_NAMES
def test_main_rejects_non_loopback_host(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("MCP_HOST", "0.0.0.0")
with pytest.raises(ValueError, match="loopback"):
main()
def test_main_runs_stateless_streamable_http(monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, object] = {}
runtime_service = object()
class RuntimeServer:
def run(self, *, transport: str, **kwargs: object) -> None:
captured.update(transport=transport, **kwargs)
def build_service(settings: object) -> object:
captured["settings"] = settings
return runtime_service
def create_server(service: object) -> RuntimeServer:
assert service is runtime_service
return RuntimeServer()
monkeypatch.setenv("MCP_HOST", "127.0.0.1")
monkeypatch.setenv("MCP_PORT", "8123")
monkeypatch.setattr(server_module, "build_service", build_service)
monkeypatch.setattr(server_module, "create_server", create_server)
main()
settings = captured.pop("settings")
assert isinstance(settings, server_module.Settings)
assert settings.host == "127.0.0.1"
assert settings.port == 8123
assert captured == {
"transport": "streamable-http",
"host": "127.0.0.1",
"port": 8123,
"stateless_http": True,
"json_response": True,
}
def test_main_runs_authenticated_hosted_transport(monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, object] = {}
transport_security = object()
class HostedRuntimeSettings:
host = "0.0.0.0"
port = 7860
@classmethod
def from_env(cls) -> HostedRuntimeSettings:
return cls()
def transport_security(self) -> object:
return transport_security
class RuntimeServer:
def run(self, *, transport: str, **kwargs: object) -> None:
captured.update(transport=transport, **kwargs)
def create_hosted_server(settings: object) -> RuntimeServer:
assert isinstance(settings, HostedRuntimeSettings)
return RuntimeServer()
monkeypatch.setenv("MCP_DEPLOYMENT_MODE", "hosted")
monkeypatch.setattr(server_module, "HostedSettings", HostedRuntimeSettings)
monkeypatch.setattr(server_module, "create_hosted_server", create_hosted_server)
main()
assert captured == {
"transport": "streamable-http",
"host": "0.0.0.0",
"port": 7860,
"stateless_http": True,
"json_response": True,
"transport_security": transport_security,
}
|