music-assistant-server

1"""
2Tests for routing player commands to whatever is producing the audio.
3
4A player command applies to what the player is actually playing: a live external
5source handles it in its own session, and Music Assistant's queue handles it for
6its own items. This replaces the queue-delegation tests, which covered the same
7forwarding while a live source was a queue item.
8"""
9
10from typing import Any
11from unittest.mock import AsyncMock, MagicMock
12
13import pytest
14from music_assistant_models.enums import ProviderFeature, RepeatMode, SourceControl
15from music_assistant_models.errors import InvalidCommand, PlayerCommandFailed
16from music_assistant_models.media_items import AudioSource
17from music_assistant_models.media_items.provider_mapping import ProviderMapping
18
19from music_assistant.controllers.players import PlayerController
20from music_assistant.controllers.players.audio_sources import AudioSourceSession
21from music_assistant.models.plugin import PluginProvider
22
23PLAYER_ID = "player_1"
24PROVIDER_INSTANCE = "spotify_connect--abc"
25
26
27def _source(
28    *,
29    can_play_pause: bool = False,
30    can_seek: bool = False,
31    can_next_previous: bool = False,
32) -> AudioSource:
33    return AudioSource(
34        item_id="main",
35        provider=PROVIDER_INSTANCE,
36        name="Spotify Connect",
37        provider_mappings={
38            ProviderMapping(
39                item_id="main",
40                provider_domain="spotify_connect",
41                provider_instance=PROVIDER_INSTANCE,
42            )
43        },
44        can_play_pause=can_play_pause,
45        can_seek=can_seek,
46        can_next_previous=can_next_previous,
47    )
48
49
50def _controller(source: AudioSource | None) -> tuple[Any, MagicMock]:
51    """Build a controller with (or without) a live source on PLAYER_ID."""
52    mass = MagicMock()
53    mass.config.get_raw_core_config_value.return_value = "INFO"
54    controller = PlayerController(mass)
55    provider = MagicMock(spec=PluginProvider)
56    provider.instance_id = PROVIDER_INSTANCE
57    provider.supported_features = {ProviderFeature.AUDIO_SOURCE}
58    provider.on_source_control = AsyncMock()
59    mass.get_provider.return_value = provider
60    if source is not None:
61        controller._source_sessions[PLAYER_ID] = AudioSourceSession(
62            player_id=PLAYER_ID,
63            source=source,
64            provider_instance_id=PROVIDER_INSTANCE,
65        )
66    player = MagicMock()
67    player.player_id = PLAYER_ID
68    player.display_name = "Player 1"
69    player.available = True
70    player.state.synced_to = None
71    player.state.active_group = None
72    player.protocol_parent_id = None
73    controller.get_player = MagicMock(return_value=player)  # type: ignore[method-assign]
74    # the command decorator looks the player up in the registry, not via get_player
75    controller._players[PLAYER_ID] = player
76    return controller, provider
77
78
79async def test_a_capable_source_takes_the_command() -> None:
80    """The source is asked to seek within its own session."""
81    controller, provider = _controller(_source(can_seek=True))
82
83    handled = await controller._forward_to_external_source(
84        controller.get_player(PLAYER_ID), SourceControl.SEEK, 42
85    )
86
87    assert handled is True
88    provider.on_source_control.assert_awaited_once_with("main", SourceControl.SEEK, 42)
89
90
91async def test_a_source_that_cannot_do_it_refuses_rather_than_forwarding() -> None:
92    """A client is told no instead of being left waiting on a command nothing handles."""
93    controller, provider = _controller(_source(can_seek=False))
94
95    with pytest.raises(PlayerCommandFailed, match="does not support this action"):
96        await controller._forward_to_external_source(
97            controller.get_player(PLAYER_ID), SourceControl.SEEK, 42
98        )
99
100    provider.on_source_control.assert_not_awaited()
101
102
103@pytest.mark.parametrize(
104    ("action", "flag"),
105    [
106        (SourceControl.PLAY, "can_play_pause"),
107        (SourceControl.PAUSE, "can_play_pause"),
108        (SourceControl.NEXT, "can_next_previous"),
109        (SourceControl.PREVIOUS, "can_next_previous"),
110        (SourceControl.SEEK, "can_seek"),
111    ],
112)
113async def test_each_transport_action_is_gated_on_its_own_flag(
114    action: SourceControl, flag: str
115) -> None:
116    """Every transport action is gated by the capability that describes it."""
117    controller, _provider = _controller(_source(**{flag: True}))
118    assert await controller._forward_to_external_source(controller.get_player(PLAYER_ID), action)
119
120    controller, _provider = _controller(_source(**{flag: False}))
121    with pytest.raises(PlayerCommandFailed):
122        await controller._forward_to_external_source(controller.get_player(PLAYER_ID), action)
123
124
125@pytest.mark.parametrize("action", [SourceControl.SHUFFLE, SourceControl.REPEAT])
126async def test_ordering_is_not_gated_by_a_capability_flag(action: SourceControl) -> None:
127    """
128    Shuffle and repeat go to the session unconditionally.
129
130    Only the session knows whether its content can be reordered, and it refuses in
131    its own words — a flag here would second-guess it.
132    """
133    controller, provider = _controller(_source())
134
135    assert await controller._forward_to_external_source(
136        controller.get_player(PLAYER_ID), action, True
137    )
138
139    provider.on_source_control.assert_awaited_once_with("main", action, True)
140
141
142async def test_nothing_is_forwarded_when_no_source_is_playing() -> None:
143    """With no live source the caller is told to look elsewhere, not refused."""
144    controller, provider = _controller(None)
145
146    handled = await controller._forward_to_external_source(
147        controller.get_player(PLAYER_ID), SourceControl.SEEK, 42
148    )
149
150    assert handled is False
151    provider.on_source_control.assert_not_awaited()
152
153
154async def test_a_gone_provider_is_not_forwarded_to() -> None:
155    """A session whose plugin has unloaded forwards nothing rather than raising."""
156    controller, _provider = _controller(_source(can_seek=True))
157    controller.mass.get_provider.return_value = None
158
159    assert (
160        await controller._forward_to_external_source(
161            controller.get_player(PLAYER_ID), SourceControl.SEEK, 42
162        )
163        is False
164    )
165
166
167async def test_a_group_member_is_playing_its_groups_source() -> None:
168    """A member hearing its group's audio resolves to the group's source, not its own."""
169    controller, provider = _controller(None)
170    group_source = _source(can_seek=True)
171    controller._source_sessions["group_1"] = AudioSourceSession(
172        player_id="group_1",
173        source=group_source,
174        provider_instance_id=PROVIDER_INSTANCE,
175    )
176    member = MagicMock()
177    member.player_id = "member_1"
178    member.display_name = "Member"
179    member.state.synced_to = None
180    member.state.active_group = "group_1"
181    member.protocol_parent_id = None
182    group = MagicMock()
183    group.player_id = "group_1"
184    group.state.synced_to = None
185    group.state.active_group = None
186    group.protocol_parent_id = None
187    controller.get_player = MagicMock(
188        side_effect=lambda pid, *_a, **_k: {"member_1": member, "group_1": group}.get(pid)
189    )
190
191    assert await controller._forward_to_external_source(member, SourceControl.SEEK, 7)
192
193    provider.on_source_control.assert_awaited_once_with("main", SourceControl.SEEK, 7)
194
195
196async def test_shuffle_falls_through_to_the_queue_when_no_source_is_playing() -> None:
197    """Without a live source, shuffle is the Music Assistant queue's business."""
198    controller, provider = _controller(None)
199    controller._get_player_with_redirect = MagicMock(return_value=controller.get_player(PLAYER_ID))
200    queue = MagicMock()
201    queue.queue_id = PLAYER_ID
202    controller.get_active_queue = MagicMock(return_value=queue)
203    controller.mass.player_queues.set_shuffle = AsyncMock()
204
205    await controller.cmd_shuffle(PLAYER_ID, shuffle_enabled=True)
206
207    controller.mass.player_queues.set_shuffle.assert_awaited_once_with(PLAYER_ID, True)
208    provider.on_source_control.assert_not_awaited()
209
210
211async def test_repeat_reaches_the_live_source_before_the_queue() -> None:
212    """A live source is what is playing, so it gets the command and the queue does not."""
213    controller, provider = _controller(_source())
214    controller._get_player_with_redirect = MagicMock(return_value=controller.get_player(PLAYER_ID))
215    controller.get_active_queue = MagicMock(return_value=MagicMock())
216    controller.mass.player_queues.set_repeat = AsyncMock()
217
218    await controller.cmd_repeat(PLAYER_ID, RepeatMode.ALL)
219
220    provider.on_source_control.assert_awaited_once_with(
221        "main", SourceControl.REPEAT, RepeatMode.ALL
222    )
223    controller.mass.player_queues.set_repeat.assert_not_awaited()
224
225
226async def test_a_source_with_no_control_surface_refuses_cleanly() -> None:
227    """
228    A source that implements no controls at all is a refusal, not a server error.
229
230    vban_receiver has no on_source_control, so the call reaches the base
231    implementation and raises NotImplementedError. Ordering is not gated here, so
232    that path is reachable and a caller should get a refusal it can render.
233    """
234    controller, provider = _controller(_source())
235    provider.on_source_control = AsyncMock(side_effect=NotImplementedError)
236
237    with pytest.raises(PlayerCommandFailed, match="can not be controlled"):
238        await controller._forward_to_external_source(
239            controller.get_player(PLAYER_ID), SourceControl.SHUFFLE, True
240        )
241
242
243async def test_an_unknown_repeat_mode_is_refused_before_it_reaches_a_source() -> None:
244    """
245    UNKNOWN is what a source reports when it cannot say, not a mode to set.
246
247    Forwarding it asks a plugin to apply a non-mode: soloist raises a bare ValueError
248    on it, and other providers would silently accept and do nothing.
249    """
250    controller, provider = _controller(_source())
251    controller._get_player_with_redirect = MagicMock(return_value=controller.get_player(PLAYER_ID))
252
253    with pytest.raises(InvalidCommand, match="unknown repeat mode"):
254        await controller.cmd_repeat(PLAYER_ID, RepeatMode.UNKNOWN)
255
256    provider.on_source_control.assert_not_awaited()
257