/
/
/
1"""Tests for the ``playback`` sub-server tools."""
2
3from __future__ import annotations
4
5import contextlib
6from collections.abc import Iterator
7from typing import Any
8from unittest.mock import MagicMock
9
10import pytest
11from fastmcp import Client, FastMCP
12from fastmcp.exceptions import ToolError
13from music_assistant_models.errors import MediaNotFoundError
14from music_assistant_models.media_items import BrowseFolder
15
16from music_assistant.providers.fastmcp_server.tools import build_playback_server
17
18
19@pytest.fixture
20def mounted_playback(mock_mass: Any) -> Iterator[FastMCP]:
21 """Build a root FastMCP with only the playback sub-server mounted."""
22 mcp: FastMCP = FastMCP(name="test")
23 mcp.mount(build_playback_server(mock_mass), namespace="playback")
24 try:
25 yield mcp
26 finally:
27 close = getattr(mcp, "close", None) or getattr(mcp, "shutdown", None)
28 if callable(close):
29 with contextlib.suppress(Exception):
30 close()
31
32
33async def test_pause_calls_player_queues_pause(mock_mass: Any, mounted_playback: FastMCP) -> None:
34 """``playback_pause`` always pauses via the non-toggling queue API."""
35 async with Client(mounted_playback) as client:
36 await client.call_tool("playback_pause", {"queue_id": "player1"})
37 mock_mass.player_queues.pause.assert_awaited_once_with("player1")
38
39
40async def test_resume_calls_player_queues_resume(mock_mass: Any, mounted_playback: FastMCP) -> None:
41 """``playback_resume`` always resumes via the non-toggling queue API."""
42 async with Client(mounted_playback) as client:
43 await client.call_tool("playback_resume", {"queue_id": "player1"})
44 mock_mass.player_queues.resume.assert_awaited_once_with("player1")
45
46
47async def test_play_pause_calls_player_queues_play_pause(
48 mock_mass: Any, mounted_playback: FastMCP
49) -> None:
50 """``playback_play_pause`` forwards to the toggling queue API."""
51 async with Client(mounted_playback) as client:
52 await client.call_tool("playback_play_pause", {"queue_id": "player1"})
53 mock_mass.player_queues.play_pause.assert_awaited_once_with("player1")
54
55
56async def test_play_media_enqueues_uri_directly(mock_mass: Any, mounted_playback: FastMCP) -> None:
57 """``playback_play_media`` plays the given URI as-is (no radio)."""
58 async with Client(mounted_playback) as client:
59 await client.call_tool(
60 "playback_play_media",
61 {"queue_id": "player1", "uri": "spotify://track/abc"},
62 )
63 mock_mass.music.get_item_by_uri.assert_not_awaited()
64 mock_mass.player_queues.play_media.assert_awaited_once_with("player1", "spotify://track/abc")
65
66
67async def test_play_media_radio_enqueues_radio_playlist_uri(
68 mock_mass: Any, mounted_playback: FastMCP
69) -> None:
70 """``radio=True`` resolves the seed and enqueues its radio_playlist:// dynamic playlist."""
71 seed = MagicMock()
72 seed.uri = "spotify://artist/abc"
73 mock_mass.music.get_item_by_uri.return_value = seed
74 async with Client(mounted_playback) as client:
75 await client.call_tool(
76 "playback_play_media",
77 {"queue_id": "player1", "uri": "spotify://artist/abc", "radio": True},
78 )
79 mock_mass.music.get_item_by_uri.assert_awaited_once_with("spotify://artist/abc")
80 mock_mass.player_queues.play_media.assert_awaited_once_with(
81 "player1", "radio_playlist://playlist/spotify://artist/abc"
82 )
83
84
85async def test_play_media_radio_rejects_browse_folder(
86 mock_mass: Any, mounted_playback: FastMCP
87) -> None:
88 """A browse folder is not a playable radio seed; the tool rejects it cleanly."""
89 mock_mass.music.get_item_by_uri.return_value = MagicMock(spec=BrowseFolder)
90 async with Client(mounted_playback) as client:
91 with pytest.raises(ToolError, match="browse folder"):
92 await client.call_tool(
93 "playback_play_media",
94 {"queue_id": "player1", "uri": "library://browse/x", "radio": True},
95 )
96 mock_mass.player_queues.play_media.assert_not_awaited()
97
98
99async def test_play_media_radio_resolution_failure_raises_tool_error(
100 mock_mass: Any, mounted_playback: FastMCP
101) -> None:
102 """A failed seed lookup surfaces as a clean ToolError, not a raw MA error."""
103 mock_mass.music.get_item_by_uri.side_effect = MediaNotFoundError("nope")
104 async with Client(mounted_playback) as client:
105 with pytest.raises(ToolError, match="Could not resolve URI for radio"):
106 await client.call_tool(
107 "playback_play_media",
108 {"queue_id": "player1", "uri": "spotify://track/missing", "radio": True},
109 )
110 mock_mass.player_queues.play_media.assert_not_awaited()
111