music-assistant-server

1"""Tests for elicitation on destructive operations (C8)."""
2# mypy: disable-error-code="arg-type, no-untyped-def, type-arg, assignment, operator, misc"
3
4from __future__ import annotations
5
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from fastmcp import Client, FastMCP
10from fastmcp.exceptions import ToolError
11from mcp.shared.exceptions import McpError
12from mcp.types import INTERNAL_ERROR, INVALID_REQUEST, ErrorData
13
14from music_assistant.providers.fastmcp_server.tools import build_media_server, build_queue_server
15from music_assistant.providers.fastmcp_server.tools._common import confirm_or_raise
16
17
18def _server(mass: MagicMock, *, require_confirmation: bool) -> FastMCP:
19    """Build a small root server mounting only queue + media for elicitation tests."""
20    mcp: FastMCP = FastMCP(name="t")
21    mcp.mount(
22        build_queue_server(mass, require_confirmation=require_confirmation),
23        namespace="queue",
24    )
25    mcp.mount(
26        build_media_server(mass, require_confirmation=require_confirmation),
27        namespace="media",
28    )
29    return mcp
30
31
32def _accepter() -> object:
33    """Build an elicitation handler that always accepts with True."""
34
35    async def handler(message, response_type, params, context):  # noqa: ARG001
36        return True
37
38    return handler
39
40
41def _decliner() -> object:
42    """Build an elicitation handler that always declines."""
43    from fastmcp.client.elicitation import ElicitResult  # noqa: PLC0415
44
45    async def handler(message, response_type, params, context):  # noqa: ARG001
46        return ElicitResult(action="decline", content=None)
47
48    return handler
49
50
51async def test_clear_queue_runs_when_user_accepts(mock_mass: MagicMock) -> None:
52    """User accepts the elicitation prompt → clear_queue dispatches to MA."""
53    mock_mass.player_queues.clear = MagicMock()
54    mcp = _server(mock_mass, require_confirmation=True)
55
56    async with Client(mcp, elicitation_handler=_accepter()) as client:
57        await client.call_tool("queue_clear_queue", {"queue_id": "q1"})
58    mock_mass.player_queues.clear.assert_called_once_with("q1")
59
60
61async def test_clear_queue_blocked_when_user_declines(mock_mass: MagicMock) -> None:
62    """User declines → tool raises ToolError, no MA call is made."""
63    mock_mass.player_queues.clear = MagicMock()
64    mcp = _server(mock_mass, require_confirmation=True)
65
66    async with Client(mcp, elicitation_handler=_decliner()) as client:
67        with pytest.raises(ToolError):
68            await client.call_tool("queue_clear_queue", {"queue_id": "q1"})
69    mock_mass.player_queues.clear.assert_not_called()
70
71
72async def test_remove_item_runs_when_user_accepts(mock_mass: MagicMock) -> None:
73    """User accepts the elicitation prompt → remove_item dispatches to MA."""
74    queue = MagicMock(queue_id="q1", current_index=0, index_in_buffer=0)
75    mock_mass.player_queues.get = MagicMock(return_value=queue)
76    mock_mass.player_queues.index_by_id = MagicMock(side_effect=[2, None])
77    mcp = _server(mock_mass, require_confirmation=True)
78
79    async with Client(mcp, elicitation_handler=_accepter()) as client:
80        result = await client.call_tool(
81            "queue_remove_item",
82            {"queue_id": "q1", "item_ids": ["item-1"]},
83        )
84    mock_mass.player_queues.delete_item.assert_called_once_with("q1", "item-1")
85    assert result.data.removed == ["item-1"]
86
87
88async def test_remove_item_blocked_when_user_declines(mock_mass: MagicMock) -> None:
89    """User declines → tool raises ToolError, no MA call is made."""
90    mock_mass.player_queues.get = MagicMock(return_value=MagicMock(queue_id="q1"))
91    mcp = _server(mock_mass, require_confirmation=True)
92
93    async with Client(mcp, elicitation_handler=_decliner()) as client:
94        with pytest.raises(ToolError):
95            await client.call_tool(
96                "queue_remove_item",
97                {"queue_id": "q1", "item_ids": ["item-1"]},
98            )
99    mock_mass.player_queues.delete_item.assert_not_called()
100
101
102async def test_no_confirmation_when_disabled(mock_mass: MagicMock) -> None:
103    """With require_confirmation=False, elicitation is skipped entirely."""
104    mock_mass.player_queues.clear = MagicMock()
105    mcp = _server(mock_mass, require_confirmation=False)
106
107    elicit_called = False
108
109    async def handler(message, response_type, params, context):  # noqa: ARG001
110        nonlocal elicit_called
111        elicit_called = True
112        return True
113
114    async with Client(mcp, elicitation_handler=handler) as client:
115        await client.call_tool("queue_clear_queue", {"queue_id": "q1"})
116    assert elicit_called is False
117    mock_mass.player_queues.clear.assert_called_once_with("q1")
118
119
120async def test_get_active_queue_clamps_include_items(mock_mass: MagicMock) -> None:
121    """
122    A client-supplied ``include_items`` is clamped to 500 to bound memory.
123
124    Without the clamp a hostile or sloppy caller could pass ``include_items=10**6``
125    and force MA to materialise the entire queue per request.
126    """
127    queue = MagicMock(queue_id="q1")
128    mock_mass.player_queues.get_active_queue = MagicMock(return_value=queue)
129    mock_mass.player_queues.items = MagicMock(return_value=[])
130    mcp = _server(mock_mass, require_confirmation=False)
131
132    async with Client(mcp) as client:
133        await client.call_tool(
134            "queue_get_active_queue",
135            {"player_id": "p1", "include_items": 10_000},
136        )
137    mock_mass.player_queues.items.assert_called_once_with("q1", limit=500, offset=0)
138
139
140async def test_get_active_queue_passes_small_limit_through(mock_mass: MagicMock) -> None:
141    """A reasonable ``include_items`` is forwarded verbatim — no over-cap."""
142    queue = MagicMock(queue_id="q1")
143    mock_mass.player_queues.get_active_queue = MagicMock(return_value=queue)
144    mock_mass.player_queues.items = MagicMock(return_value=[])
145    mcp = _server(mock_mass, require_confirmation=False)
146
147    async with Client(mcp) as client:
148        await client.call_tool(
149            "queue_get_active_queue",
150            {"player_id": "p1", "include_items": 10},
151        )
152    mock_mass.player_queues.items.assert_called_once_with("q1", limit=10, offset=0)
153
154
155async def test_remove_from_library_confirms(mock_mass: MagicMock) -> None:
156    """media.remove_from_library also triggers elicitation."""
157    # MA's MusicController takes (media_type, library_item_id), not a URI —
158    # the tool resolves the URI via get_item_by_uri first.
159    resolved = MagicMock(media_type=MagicMock(), item_id="42", provider="library")
160    mock_mass.music.get_item_by_uri = AsyncMock(return_value=resolved)
161    mock_mass.music.remove_item_from_library = AsyncMock()
162    mcp = _server(mock_mass, require_confirmation=True)
163
164    async with Client(mcp, elicitation_handler=_accepter()) as client:
165        await client.call_tool("media_remove_from_library", {"uri": "lib://t/42"})
166    mock_mass.music.get_item_by_uri.assert_awaited_once_with("lib://t/42")
167    mock_mass.music.remove_item_from_library.assert_awaited_once_with(resolved.media_type, "42")
168
169
170async def test_remove_from_favorites_resolves_provider_uri_to_library(
171    mock_mass: MagicMock,
172) -> None:
173    """
174    A provider URI is resolved to the matching library item before removal.
175
176    ``MusicController.remove_item_from_*`` expects a library item id; passing the
177    provider's native item id silently targets the wrong item (or raises on a
178    non-numeric ``int()`` cast). The tool now looks up the library counterpart via
179    ``get_library_item_by_prov_id``.
180    """
181    provider_item = MagicMock(media_type=MagicMock(), item_id="prov-abc", provider="yandex_music")
182    library_item = MagicMock(media_type=provider_item.media_type, item_id="99")
183    mock_mass.music.get_item_by_uri = AsyncMock(return_value=provider_item)
184    mock_mass.music.get_library_item_by_prov_id = AsyncMock(return_value=library_item)
185    mock_mass.music.remove_item_from_favorites = AsyncMock()
186    mcp = _server(mock_mass, require_confirmation=False)
187
188    async with Client(mcp) as client:
189        await client.call_tool(
190            "media_remove_from_favorites", {"uri": "yandex_music://track/prov-abc"}
191        )
192    mock_mass.music.get_library_item_by_prov_id.assert_awaited_once_with(
193        provider_item.media_type, "prov-abc", "yandex_music"
194    )
195    mock_mass.music.remove_item_from_favorites.assert_awaited_once_with(
196        library_item.media_type, "99"
197    )
198
199
200async def test_remove_from_library_raises_when_not_in_library(
201    mock_mass: MagicMock,
202) -> None:
203    """
204    When the URI's library counterpart cannot be resolved, the tool raises.
205
206    Without this, the tool would silently call ``remove_item_from_library`` with
207    a provider-native item id, which either fails on ``int()`` cast or targets
208    the wrong item.
209    """
210    provider_item = MagicMock(media_type=MagicMock(), item_id="prov-abc", provider="yandex_music")
211    mock_mass.music.get_item_by_uri = AsyncMock(return_value=provider_item)
212    mock_mass.music.get_library_item_by_prov_id = AsyncMock(return_value=None)
213    mock_mass.music.remove_item_from_library = AsyncMock()
214    mcp = _server(mock_mass, require_confirmation=False)
215
216    async with Client(mcp) as client:
217        with pytest.raises(ToolError):
218            await client.call_tool(
219                "media_remove_from_library", {"uri": "yandex_music://track/prov-abc"}
220            )
221    mock_mass.music.remove_item_from_library.assert_not_awaited()
222
223
224async def test_confirm_passes_through_when_elicitation_unsupported() -> None:
225    """McpError(INVALID_REQUEST) = no capability → pass through (no raise)."""
226    ctx = MagicMock()
227    ctx.elicit = AsyncMock(
228        side_effect=McpError(ErrorData(code=INVALID_REQUEST, message="Elicitation not supported"))
229    )
230    # Must NOT raise — pass-through to permission flag.
231    await confirm_or_raise(ctx, "confirm?", enabled=True)
232
233
234async def test_confirm_fails_closed_on_unexpected_mcp_error() -> None:
235    """A non-capability McpError must re-raise (fail closed), not bypass confirmation."""
236    ctx = MagicMock()
237    ctx.elicit = AsyncMock(
238        side_effect=McpError(ErrorData(code=INTERNAL_ERROR, message="handler blew up"))
239    )
240    with pytest.raises(McpError):
241        await confirm_or_raise(ctx, "confirm?", enabled=True)
242