music-assistant-server

1"""Tests for the opt-in simplified tool discovery (meta-tool) mode."""
2# mypy: disable-error-code="arg-type, no-untyped-def, type-arg, assignment, operator, misc"
3
4from __future__ import annotations
5
6from typing import Any
7from unittest.mock import MagicMock
8
9import pytest
10from fastmcp import Client, FastMCP
11from fastmcp.client.elicitation import ElicitResult
12from fastmcp.exceptions import ToolError
13from mcp.shared.exceptions import McpError
14
15from music_assistant.providers.fastmcp_server.config import build_config_entries
16from music_assistant.providers.fastmcp_server.constants import (
17    CONF_META_TOOL_DISCOVERY,
18    DEFAULT_MOUNT_PATH,
19    HOT_SWAPPABLE_KEYS,
20)
21from music_assistant.providers.fastmcp_server.meta_discovery import register_meta_discovery
22from music_assistant.providers.fastmcp_server.middleware import TagFilterMiddleware
23from music_assistant.providers.fastmcp_server.server import build_tag_lookup
24from music_assistant.providers.fastmcp_server.tools import build_queue_server
25
26_ALL_TAGS = {"query:queue", "edit:queue", "delete:queue", "control:playback"}
27_META_NAMES = {"search_tools", "call_tool", "get_tool_schema"}
28
29
30def _server(
31    mock_mass: MagicMock,
32    *,
33    require_confirmation: bool = False,
34) -> tuple[FastMCP, dict[str, Any]]:
35    """
36    Build a root FastMCP mirroring the runtime wiring for meta-discovery tests.
37
38    Returns the server plus a mutable state dict: ``state["enabled"]`` gates
39    the meta mode and ``state["allowed"]`` is the live allowed-tag set, both
40    read through closures exactly like the runtime's hot-swap path.
41    """
42    state: dict[str, Any] = {"enabled": True, "allowed": set(_ALL_TAGS)}
43    mcp: FastMCP = FastMCP(name="test")
44    mcp.mount(
45        build_queue_server(mock_mass, require_confirmation=require_confirmation),
46        namespace="queue",
47    )
48    mcp.add_middleware(TagFilterMiddleware(lambda: state["allowed"], build_tag_lookup(mcp)))
49    register_meta_discovery(
50        mcp,
51        enabled=lambda: bool(state["enabled"]),
52        allowed_tags_provider=lambda: state["allowed"],
53        lookup_component_tags=build_tag_lookup(mcp),
54    )
55    return mcp, state
56
57
58async def test_meta_off_keeps_full_catalog(mock_mass: MagicMock) -> None:
59    """With the toggle off the normal catalog is listed and no meta tool leaks."""
60    mcp, state = _server(mock_mass)
61    state["enabled"] = False
62    async with Client(mcp) as client:
63        names = {t.name for t in await client.list_tools()}
64    assert "queue_get_active_queue" in names
65    assert not (_META_NAMES & names)
66
67
68async def test_meta_on_lists_exactly_three_tools(mock_mass: MagicMock) -> None:
69    """With the toggle on the listing collapses to the three meta tools."""
70    mcp, _state = _server(mock_mass)
71    async with Client(mcp) as client:
72        names = {t.name for t in await client.list_tools()}
73    assert names == _META_NAMES
74
75
76async def test_toggle_hot_swaps_listing_without_rebuild(mock_mass: MagicMock) -> None:
77    """Flipping the flag changes the listing on the next request, same server."""
78    mcp, state = _server(mock_mass)
79    async with Client(mcp) as client:
80        assert {t.name for t in await client.list_tools()} == _META_NAMES
81        state["enabled"] = False
82        names = {t.name for t in await client.list_tools()}
83        assert "queue_set_shuffle" in names
84        assert not (_META_NAMES & names)
85        state["enabled"] = True
86        assert {t.name for t in await client.list_tools()} == _META_NAMES
87
88
89async def test_search_tools_returns_lightweight_results(mock_mass: MagicMock) -> None:
90    """search_tools ranks the catalog and never inlines schemas."""
91    mcp, _state = _server(mock_mass)
92    async with Client(mcp) as client:
93        result = await client.call_tool("search_tools", {"query": "shuffle queue"})
94    entries = result.data
95    names = [e["name"] for e in entries]
96    assert "queue_set_shuffle" in names
97    for entry in entries:
98        assert set(entry) == {"name", "description"}, (
99            f"search result must stay lightweight, got keys {set(entry)}"
100        )
101
102
103async def test_search_tools_respects_rbac(mock_mass: MagicMock) -> None:
104    """A tag-disabled tool never surfaces in search results."""
105    mcp, state = _server(mock_mass)
106    state["allowed"] = {"query:queue"}
107    async with Client(mcp) as client:
108        result = await client.call_tool("search_tools", {"query": "shuffle queue"})
109    names = [e["name"] for e in result.data]
110    assert "queue_set_shuffle" not in names, "edit:queue is disabled — must not surface"
111
112
113async def test_call_tool_proxies_execution(mock_mass: MagicMock) -> None:
114    """call_tool executes a permitted catalogued tool with the given arguments."""
115    mcp, _state = _server(mock_mass)
116    async with Client(mcp) as client:
117        await client.call_tool(
118            "call_tool",
119            {"name": "queue_set_shuffle", "arguments": {"queue_id": "q1", "enabled": True}},
120        )
121    mock_mass.player_queues.set_shuffle.assert_awaited_once_with("q1", True)
122
123
124async def test_call_tool_blocked_for_disabled_tag(mock_mass: MagicMock) -> None:
125    """The proxy re-enters the middleware chain, so RBAC still blocks the call."""
126    mcp, state = _server(mock_mass)
127    state["allowed"] = {"query:queue"}
128    async with Client(mcp) as client:
129        with pytest.raises((ToolError, McpError), match=r"disabled|not found"):
130            await client.call_tool(
131                "call_tool",
132                {"name": "queue_set_shuffle", "arguments": {"queue_id": "q1", "enabled": True}},
133            )
134    mock_mass.player_queues.set_shuffle.assert_not_awaited()
135
136
137async def test_get_tool_schema_returns_full_schema(mock_mass: MagicMock) -> None:
138    """get_tool_schema returns the input schema for one permitted tool."""
139    mcp, _state = _server(mock_mass)
140    async with Client(mcp) as client:
141        result = await client.call_tool("get_tool_schema", {"tool_name": "queue_set_shuffle"})
142    schema = result.data
143    assert schema["name"] == "queue_set_shuffle"
144    assert "queue_id" in schema["inputSchema"]["properties"]
145    assert "enabled" in schema["inputSchema"]["properties"]
146
147
148async def test_get_tool_schema_hides_disabled_tool(mock_mass: MagicMock) -> None:
149    """A tag-disabled tool's schema is not disclosed."""
150    mcp, state = _server(mock_mass)
151    state["allowed"] = {"query:queue"}
152    async with Client(mcp) as client:
153        with pytest.raises((ToolError, McpError), match="not found"):
154            await client.call_tool("get_tool_schema", {"tool_name": "queue_set_shuffle"})
155
156
157async def test_get_tool_schema_unknown_tool(mock_mass: MagicMock) -> None:
158    """An unknown tool name reports not-found."""
159    mcp, _state = _server(mock_mass)
160    async with Client(mcp) as client:
161        with pytest.raises((ToolError, McpError), match="not found"):
162            await client.call_tool("get_tool_schema", {"tool_name": "no_such_tool"})
163
164
165async def test_elicitation_fires_through_call_tool_proxy(mock_mass: MagicMock) -> None:
166    """Destructive-op confirmation still gates tools invoked via the proxy."""
167    prompts: list[str] = []
168
169    async def handler(message, response_type, params, context):  # noqa: ARG001
170        prompts.append(message)
171        return True
172
173    mcp, _state = _server(mock_mass, require_confirmation=True)
174    async with Client(mcp, elicitation_handler=handler) as client:
175        await client.call_tool(
176            "call_tool",
177            {"name": "queue_clear_queue", "arguments": {"queue_id": "q1"}},
178        )
179    assert prompts, "elicitation prompt must reach the client through the proxy"
180    mock_mass.player_queues.clear.assert_called_once_with("q1")
181
182
183async def test_elicitation_decline_blocks_through_proxy(mock_mass: MagicMock) -> None:
184    """Declining the confirmation through the proxy leaves the queue untouched."""
185
186    async def decliner(message, response_type, params, context):  # noqa: ARG001
187        return ElicitResult(action="decline", content=None)
188
189    mcp, _state = _server(mock_mass, require_confirmation=True)
190    async with Client(mcp, elicitation_handler=decliner) as client:
191        with pytest.raises((ToolError, McpError)):
192            await client.call_tool(
193                "call_tool",
194                {"name": "queue_clear_queue", "arguments": {"queue_id": "q1"}},
195            )
196    mock_mass.player_queues.clear.assert_not_called()
197
198
199async def test_direct_call_still_works_in_meta_mode(mock_mass: MagicMock) -> None:
200    """
201    Catalogued tools stay callable by name in meta mode (hidden, not blocked).
202
203    FastMCP's search transforms hide tools from the listing but keep
204    ``get_tool`` delegating, so a stale client that cached tool names keeps
205    working; RBAC still applies via the middleware.
206    """
207    mcp, _state = _server(mock_mass)
208    async with Client(mcp) as client:
209        await client.call_tool("queue_set_shuffle", {"queue_id": "q1", "enabled": False})
210    mock_mass.player_queues.set_shuffle.assert_awaited_once_with("q1", False)
211
212
213async def test_config_entry_registered_default_off(mock_mass: MagicMock) -> None:
214    """The meta toggle ships as a Server-category boolean, default off."""
215    entries = {e.key: e for e in build_config_entries(mock_mass, DEFAULT_MOUNT_PATH)}
216    entry = entries[CONF_META_TOOL_DISCOVERY]
217    assert entry.default_value is False
218    assert entry.category == "server"
219    assert CONF_META_TOOL_DISCOVERY in HOT_SWAPPABLE_KEYS
220