music-assistant-server

1"""
2Tests for the ``players`` sub-server tools.
3
4End-to-end through a FastMCP ``Client`` so the parameter signature and
5filtering behaviour exposed to MCP clients are pinned, not just the
6in-process helpers.
7"""
8
9from __future__ import annotations
10
11import contextlib
12import json
13from collections.abc import Iterator
14from pathlib import Path
15from types import SimpleNamespace
16from typing import Any
17
18import pytest
19from fastmcp import Client, FastMCP
20from fastmcp.exceptions import ToolError
21
22from music_assistant.providers.fastmcp_server.tools import build_players_server, build_queue_server
23
24
25def _player(
26    *,
27    player_id: str,
28    name: str,
29    available: bool = True,
30    enabled: bool = True,
31    state: str = "idle",
32    needs_setup: bool = False,
33    active_group: str | None = None,
34    synced_to: str | None = None,
35) -> SimpleNamespace:
36    """Build a minimal player stub that satisfies ``to_brief_player``."""
37    return SimpleNamespace(
38        player_id=player_id,
39        name=name,
40        playback_state=SimpleNamespace(value=state),
41        volume_level=None,
42        powered=True,
43        current_media=None,
44        available=available,
45        enabled=enabled,
46        needs_setup=needs_setup,
47        active_group=active_group,
48        synced_to=synced_to,
49    )
50
51
52def _make_all_players_mock(roster: list[SimpleNamespace]) -> Any:
53    """
54    Side effect mirroring MA's ``all_players(return_unavailable, return_disabled)`` filter.
55
56    Lets the tests pin the contract — "we forward the flag to MA" — without
57    coupling to a re-implementation in Python.
58    """
59
60    def side_effect(
61        *,
62        return_unavailable: bool = True,
63        return_disabled: bool = False,
64        **_kwargs: Any,
65    ) -> list[Any]:
66        result = list(roster)
67        if not return_unavailable:
68            result = [p for p in result if p.available]
69        if not return_disabled:
70            result = [p for p in result if p.enabled]
71        return result
72
73    return side_effect
74
75
76@pytest.fixture
77def mounted_players(mock_mass: Any) -> Iterator[FastMCP]:
78    """
79    Build a root FastMCP with only the players sub-server mounted.
80
81    Yields rather than returns so future FastMCP lifecycle methods (e.g.
82    a ``close()`` / ``shutdown()`` once upstream adds one) can be wired
83    into the ``finally`` block without per-test churn. ``contextlib.suppress``
84    keeps the teardown idempotent against an environment where ``close``
85    is later renamed or removed.
86    """
87    mcp: FastMCP = FastMCP(name="test")
88    mcp.mount(build_players_server(mock_mass), namespace="players")
89    try:
90        yield mcp
91    finally:
92        close = getattr(mcp, "close", None) or getattr(mcp, "shutdown", None)
93        if callable(close):
94            with contextlib.suppress(Exception):
95                close()
96
97
98async def test_list_players_hides_unavailable_by_default(
99    mock_mass: Any, mounted_players: FastMCP
100) -> None:
101    """
102    Default ``list_players`` call asks MA to omit unavailable and disabled players.
103
104    Matches the spec: a model asked to pick a speaker should not see
105    devices MA can no longer reach or that the admin has disabled.
106    The filter lives in MA's controller, so the contract this test
107    pins is that the tool forwards both flags to ``all_players``.
108    """
109    roster = [
110        _player(player_id="ok", name="Kitchen", available=True),
111        _player(player_id="gone", name="Bedroom", available=False),
112    ]
113    mock_mass.players.all_players.side_effect = _make_all_players_mock(roster)
114    async with Client(mounted_players) as client:
115        result = await client.call_tool("players_list_players", {})
116    ids = {p.player_id for p in result.data}
117    assert ids == {"ok"}, "unavailable players must be filtered out by default"
118    mock_mass.players.all_players.assert_called_with(
119        return_unavailable=False, return_disabled=False
120    )
121
122
123async def test_list_players_include_unavailable_returns_all(
124    mock_mass: Any, mounted_players: FastMCP
125) -> None:
126    """
127    With ``include_unavailable=True`` every player comes through.
128
129    The unavailable one is still tagged ``state="unavailable"`` so the
130    caller can act on it.
131    """
132    roster = [
133        _player(player_id="ok", name="Kitchen", available=True),
134        _player(player_id="gone", name="Bedroom", available=False),
135    ]
136    mock_mass.players.all_players.side_effect = _make_all_players_mock(roster)
137    async with Client(mounted_players) as client:
138        result = await client.call_tool("players_list_players", {"include_unavailable": True})
139    by_id = {p.player_id: p for p in result.data}
140    assert set(by_id) == {"ok", "gone"}
141    assert by_id["gone"].state == "unavailable"
142    assert by_id["ok"].state == "idle"
143    mock_mass.players.all_players.assert_called_with(return_unavailable=True, return_disabled=False)
144
145
146async def test_list_players_include_disabled_returns_disabled(
147    mock_mass: Any, mounted_players: FastMCP
148) -> None:
149    """
150    With ``include_disabled=True`` admin-disabled players surface as ``state="disabled"``.
151
152    Without the flag MA filters them out before they reach the brief,
153    so the ``enabled`` field on the response is always ``True``.
154    """
155    roster = [
156        _player(player_id="ok", name="Kitchen", available=True, enabled=True),
157        _player(player_id="off", name="Closet", available=True, enabled=False),
158    ]
159    mock_mass.players.all_players.side_effect = _make_all_players_mock(roster)
160    async with Client(mounted_players) as client:
161        result = await client.call_tool("players_list_players", {"include_disabled": True})
162    by_id = {p.player_id: p for p in result.data}
163    assert set(by_id) == {"ok", "off"}
164    assert by_id["off"].enabled is False
165    assert by_id["off"].state == "disabled"
166    mock_mass.players.all_players.assert_called_with(return_unavailable=False, return_disabled=True)
167
168
169async def test_list_players_synced_player_reports_synced_state(
170    mock_mass: Any, mounted_players: FastMCP
171) -> None:
172    """
173    A player belonging to an active sync group reports ``state="synced"``.
174
175    The cached ``playback_state`` on a sync follower stays ``idle``
176    because the device doesn't have its own queue — the group plays
177    through it. Without this synthesis an LLM would treat a sync
178    follower as a quiet, idle speaker.
179    """
180    roster = [
181        _player(
182            player_id="follower",
183            name="Lenco LS-500",
184            available=True,
185            active_group="syncgroup_x",
186        ),
187    ]
188    mock_mass.players.all_players.side_effect = _make_all_players_mock(roster)
189    async with Client(mounted_players) as client:
190        result = await client.call_tool("players_list_players", {})
191    by_id = {p.player_id: p for p in result.data}
192    assert by_id["follower"].state == "synced"
193    assert by_id["follower"].active_group == "syncgroup_x"
194
195
196async def test_list_players_synced_reads_from_state_object(
197    mock_mass: Any, mounted_players: FastMCP
198) -> None:
199    """
200    ``state.active_group`` (canonical) wins over the raw dataclass attr.
201
202    Mirrors the live MA shape — ``Player.state.active_group`` is set
203    by ``__final_active_group`` while the raw ``Player.active_group``
204    field stays ``None`` for SyncGroupPlayer followers. The brief must
205    read the canonical signal end-to-end through the tool.
206    """
207    follower = SimpleNamespace(
208        player_id="follower",
209        name="Kitchen",
210        playback_state=SimpleNamespace(value="idle"),
211        volume_level=None,
212        powered=True,
213        current_media=None,
214        available=True,
215        enabled=True,
216        needs_setup=False,
217        # Raw attrs are None — the bug condition we shipped in 0.3.32.
218        active_group=None,
219        synced_to=None,
220        # Canonical view carries the resolved group id.
221        state=SimpleNamespace(
222            powered=True,
223            current_media=None,
224            active_group="syncgroup_x",
225            synced_to=None,
226        ),
227    )
228    mock_mass.players.all_players.side_effect = _make_all_players_mock([follower])
229    async with Client(mounted_players) as client:
230        result = await client.call_tool("players_list_players", {})
231    by_id = {p.player_id: p for p in result.data}
232    assert by_id["follower"].state == "synced"
233    assert by_id["follower"].active_group == "syncgroup_x"
234
235
236async def test_list_players_syncgroup_reports_group_volume(
237    mock_mass: Any, mounted_players: FastMCP
238) -> None:
239    """
240    A SyncGroupPlayer surfaces its ``group_volume`` round-tripped through MCP.
241
242    Per-player ``volume_level`` is ``None`` on a sync group; the
243    canonical volume signal lives on ``state.group_volume``. The brief
244    must surface this so a caller can answer "what volume is the
245    group at?" without a separate query.
246    """
247    syncgroup = SimpleNamespace(
248        player_id="syncgroup_x",
249        name="Test Group",
250        playback_state=SimpleNamespace(value="playing"),
251        volume_level=None,
252        powered=True,
253        current_media=None,
254        available=True,
255        enabled=True,
256        needs_setup=False,
257        active_group=None,
258        synced_to=None,
259        state=SimpleNamespace(
260            powered=True,
261            current_media=None,
262            active_group=None,
263            synced_to=None,
264            volume_muted=False,
265            group_volume=60,
266            group_volume_muted=False,
267        ),
268    )
269    mock_mass.players.all_players.side_effect = _make_all_players_mock([syncgroup])
270    async with Client(mounted_players) as client:
271        result = await client.call_tool("players_list_players", {})
272    by_id = {p.player_id: p for p in result.data}
273    assert by_id["syncgroup_x"].group_volume == 60
274    assert by_id["syncgroup_x"].group_volume_muted is False
275    assert by_id["syncgroup_x"].volume_muted is False
276    assert by_id["syncgroup_x"].volume_level is None
277
278
279async def test_list_players_needs_setup_reports_needs_setup_state(
280    mock_mass: Any, mounted_players: FastMCP
281) -> None:
282    """A player still awaiting first-run setup reports ``state="needs_setup"``."""
283    roster = [
284        _player(
285            player_id="raw",
286            name="Unboxed Speaker",
287            available=True,
288            needs_setup=True,
289        ),
290    ]
291    mock_mass.players.all_players.side_effect = _make_all_players_mock(roster)
292    async with Client(mounted_players) as client:
293        result = await client.call_tool("players_list_players", {})
294    by_id = {p.player_id: p for p in result.data}
295    assert by_id["raw"].state == "needs_setup"
296    assert by_id["raw"].needs_setup is True
297
298
299async def test_get_player_returns_unavailable_player(
300    mock_mass: Any, mounted_players: FastMCP
301) -> None:
302    """Direct id lookup ignores availability — the caller already has the id."""
303    mock_mass.players.get_player.return_value = _player(
304        player_id="gone", name="Bedroom", available=False
305    )
306    async with Client(mounted_players) as client:
307        result = await client.call_tool("players_get_player", {"player_id": "gone"})
308    assert result.data.player_id == "gone"
309    assert result.data.available is False
310    assert result.data.state == "unavailable"
311
312
313async def test_group_player_calls_cmd_group(mock_mass: Any, mounted_players: FastMCP) -> None:
314    """``players_group_player`` forwards to ``mass.players.cmd_group``."""
315    async with Client(mounted_players) as client:
316        await client.call_tool(
317            "players_group_player",
318            {"player_id": "follower", "target_player_id": "leader"},
319        )
320    mock_mass.players.cmd_group.assert_awaited_once_with("follower", "leader")
321
322
323async def test_ungroup_player_calls_cmd_ungroup(mock_mass: Any, mounted_players: FastMCP) -> None:
324    """``players_ungroup_player`` forwards to ``mass.players.cmd_ungroup``."""
325    async with Client(mounted_players) as client:
326        await client.call_tool("players_ungroup_player", {"player_id": "follower"})
327    mock_mass.players.cmd_ungroup.assert_awaited_once_with("follower")
328
329
330async def test_get_player_reports_external_source(mock_mass: Any, mounted_players: FastMCP) -> None:
331    """An idle player driven by a Connect source reports playing + provider."""
332    player = _player(player_id="lenco", name="Lenco LS-500", state="idle")
333    mock_mass.players.get_player.return_value = player
334    queue = SimpleNamespace(
335        state=SimpleNamespace(value="playing"),
336        current_item=SimpleNamespace(
337            name="Yandex Music Connect (Ynison)",
338            streamdetails=SimpleNamespace(
339                media_type=SimpleNamespace(value="audio_source"),
340                provider="yandex_ynison--PL8BnL7a",
341                stream_metadata=SimpleNamespace(title="Behind Your Walls"),
342            ),
343        ),
344    )
345    mock_mass.player_queues.get_active_queue.return_value = queue
346    async with Client(mounted_players) as client:
347        result = await client.call_tool("players_get_player", {"player_id": "lenco"})
348    assert result.data.state == "playing"
349    assert result.data.external_source == "yandex_ynison--PL8BnL7a"
350    assert result.data.current_item == "Behind Your Walls"
351
352
353def _ns(obj: Any) -> Any:
354    """Recursively turn dicts/lists into attribute-accessible namespaces."""
355    if isinstance(obj, dict):
356        return SimpleNamespace(**{k: _ns(v) for k, v in obj.items()})
357    if isinstance(obj, list):
358        return [_ns(v) for v in obj]
359    return obj
360
361
362@pytest.mark.parametrize("call_args", [{"player_id": "lenco"}, {"queue_id": "lenco"}])
363async def test_queue_get_active_queue_by_player_or_queue_id(
364    mock_mass: Any, call_args: dict[str, str]
365) -> None:
366    """queue_get_active_queue accepts player_id or queue_id and surfaces AUDIO_SOURCE titles."""
367    raw = json.loads(
368        Path(__file__).parent.joinpath("fixtures/queue_external_audio_source.json").read_text()
369    )
370    queue = _ns(raw)
371    mock_mass.player_queues.get_active_queue.return_value = queue
372    mock_mass.player_queues.items.return_value = [queue.current_item]
373
374    mcp = FastMCP(name="test")
375    mcp.mount(build_queue_server(mock_mass), namespace="queue")
376    async with Client(mcp) as client:
377        result = await client.call_tool("queue_get_active_queue", call_args)
378    assert result.data.items[0].name == "Behind Your Walls"
379    mock_mass.player_queues.get_active_queue.assert_called_with("lenco")
380    mock_mass.player_queues.items.assert_called_with("lenco", limit=25, offset=0)
381
382
383async def test_queue_get_active_queue_requires_an_identifier(mock_mass: Any) -> None:
384    """queue_get_active_queue raises a clear error when neither id is provided."""
385    mcp = FastMCP(name="test")
386    mcp.mount(build_queue_server(mock_mass), namespace="queue")
387    async with Client(mcp) as client:
388        with pytest.raises(ToolError, match=r"player_id.*queue_id"):
389            await client.call_tool("queue_get_active_queue", {})
390