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 (
15    PlaybackState,
16    ProviderFeature,
17    RepeatMode,
18    SourceControl,
19)
20from music_assistant_models.errors import InvalidCommand, PlayerCommandFailed
21from music_assistant_models.media_items import AudioSource
22from music_assistant_models.media_items.provider_mapping import ProviderMapping
23from music_assistant_models.player import PlayerSource
24
25from music_assistant.controllers.players import PlayerController
26from music_assistant.controllers.players.audio_sources import AudioSourceSession
27from music_assistant.models.plugin import PluginProvider
28from tests.common import MockPlayer, MockProvider
29
30PLAYER_ID = "player_1"
31PROVIDER_INSTANCE = "spotify_connect--abc"
32# a source the player's own device runs, so Music Assistant has no session for it
33NATIVE_SOURCE_ID = "spotify"
34
35
36def _source(
37    *,
38    can_play_pause: bool = False,
39    can_seek: bool = False,
40    can_next_previous: bool = False,
41) -> AudioSource:
42    return AudioSource(
43        item_id="main",
44        provider=PROVIDER_INSTANCE,
45        name="Spotify Connect",
46        provider_mappings={
47            ProviderMapping(
48                item_id="main",
49                provider_domain="spotify_connect",
50                provider_instance=PROVIDER_INSTANCE,
51            )
52        },
53        can_play_pause=can_play_pause,
54        can_seek=can_seek,
55        can_next_previous=can_next_previous,
56    )
57
58
59def _controller(source: AudioSource | None) -> tuple[Any, MagicMock]:
60    """Build a controller with (or without) a live source on PLAYER_ID."""
61    mass = MagicMock()
62    mass.config.get_raw_core_config_value.return_value = "INFO"
63    controller = PlayerController(mass)
64    provider = MagicMock(spec=PluginProvider)
65    provider.instance_id = PROVIDER_INSTANCE
66    provider.supported_features = {ProviderFeature.AUDIO_SOURCE}
67    provider.on_source_control = AsyncMock()
68    mass.get_provider.return_value = provider
69    if source is not None:
70        controller._source_sessions[PLAYER_ID] = AudioSourceSession(
71            player_id=PLAYER_ID,
72            source=source,
73            provider_instance_id=PROVIDER_INSTANCE,
74        )
75    player = MagicMock()
76    player.player_id = PLAYER_ID
77    player.display_name = "Player 1"
78    player.available = True
79    player.state.synced_to = None
80    player.state.active_group = None
81    player.protocol_parent_id = None
82    controller.get_player = MagicMock(return_value=player)  # type: ignore[method-assign]
83    # the command decorator looks the player up in the registry, not via get_player
84    controller._players[PLAYER_ID] = player
85    return controller, provider
86
87
88async def test_a_capable_source_takes_the_command() -> None:
89    """The source is asked to seek within its own session."""
90    controller, provider = _controller(_source(can_seek=True))
91
92    handled = await controller._forward_to_external_source(
93        controller.get_player(PLAYER_ID), SourceControl.SEEK, 42
94    )
95
96    assert handled is True
97    provider.on_source_control.assert_awaited_once_with("main", SourceControl.SEEK, 42)
98
99
100async def test_a_source_that_cannot_do_it_refuses_rather_than_forwarding() -> None:
101    """A client is told no instead of being left waiting on a command nothing handles."""
102    controller, provider = _controller(_source(can_seek=False))
103
104    with pytest.raises(PlayerCommandFailed, match="does not support this action"):
105        await controller._forward_to_external_source(
106            controller.get_player(PLAYER_ID), SourceControl.SEEK, 42
107        )
108
109    provider.on_source_control.assert_not_awaited()
110
111
112@pytest.mark.parametrize(
113    ("action", "flag"),
114    [
115        (SourceControl.PLAY, "can_play_pause"),
116        (SourceControl.PAUSE, "can_play_pause"),
117        (SourceControl.NEXT, "can_next_previous"),
118        (SourceControl.PREVIOUS, "can_next_previous"),
119        (SourceControl.SEEK, "can_seek"),
120    ],
121)
122async def test_each_transport_action_is_gated_on_its_own_flag(
123    action: SourceControl, flag: str
124) -> None:
125    """Every transport action is gated by the capability that describes it."""
126    controller, _provider = _controller(_source(**{flag: True}))
127    assert await controller._forward_to_external_source(controller.get_player(PLAYER_ID), action)
128
129    controller, _provider = _controller(_source(**{flag: False}))
130    with pytest.raises(PlayerCommandFailed):
131        await controller._forward_to_external_source(controller.get_player(PLAYER_ID), action)
132
133
134@pytest.mark.parametrize("action", [SourceControl.SHUFFLE, SourceControl.REPEAT])
135async def test_ordering_is_not_gated_by_a_capability_flag(action: SourceControl) -> None:
136    """
137    Shuffle and repeat go to the session unconditionally.
138
139    Only the session knows whether its content can be reordered, and it refuses in
140    its own words — a flag here would second-guess it.
141    """
142    controller, provider = _controller(_source())
143
144    assert await controller._forward_to_external_source(
145        controller.get_player(PLAYER_ID), action, True
146    )
147
148    provider.on_source_control.assert_awaited_once_with("main", action, True)
149
150
151async def test_nothing_is_forwarded_when_no_source_is_playing() -> None:
152    """With no live source the caller is told to look elsewhere, not refused."""
153    controller, provider = _controller(None)
154
155    handled = await controller._forward_to_external_source(
156        controller.get_player(PLAYER_ID), SourceControl.SEEK, 42
157    )
158
159    assert handled is False
160    provider.on_source_control.assert_not_awaited()
161
162
163async def test_a_gone_provider_is_not_forwarded_to() -> None:
164    """A session whose plugin has unloaded forwards nothing rather than raising."""
165    controller, _provider = _controller(_source(can_seek=True))
166    controller.mass.get_provider.return_value = None
167
168    assert (
169        await controller._forward_to_external_source(
170            controller.get_player(PLAYER_ID), SourceControl.SEEK, 42
171        )
172        is False
173    )
174
175
176async def test_a_group_member_is_playing_its_groups_source() -> None:
177    """A member hearing its group's audio resolves to the group's source, not its own."""
178    controller, provider = _controller(None)
179    group_source = _source(can_seek=True)
180    controller._source_sessions["group_1"] = AudioSourceSession(
181        player_id="group_1",
182        source=group_source,
183        provider_instance_id=PROVIDER_INSTANCE,
184    )
185    member = MagicMock()
186    member.player_id = "member_1"
187    member.display_name = "Member"
188    member.state.synced_to = None
189    member.state.active_group = "group_1"
190    member.protocol_parent_id = None
191    group = MagicMock()
192    group.player_id = "group_1"
193    group.state.synced_to = None
194    group.state.active_group = None
195    group.protocol_parent_id = None
196    controller.get_player = MagicMock(
197        side_effect=lambda pid, *_a, **_k: {"member_1": member, "group_1": group}.get(pid)
198    )
199
200    assert await controller._forward_to_external_source(member, SourceControl.SEEK, 7)
201
202    provider.on_source_control.assert_awaited_once_with("main", SourceControl.SEEK, 7)
203
204
205async def test_shuffle_falls_through_to_the_queue_when_no_source_is_playing() -> None:
206    """Without a live source, shuffle is the Music Assistant queue's business."""
207    controller, provider = _controller(None)
208    controller._get_player_with_redirect = MagicMock(return_value=controller.get_player(PLAYER_ID))
209    queue = MagicMock()
210    queue.queue_id = PLAYER_ID
211    controller.get_active_queue = MagicMock(return_value=queue)
212    controller.mass.player_queues.set_shuffle = AsyncMock()
213
214    await controller.cmd_shuffle(PLAYER_ID, shuffle_enabled=True)
215
216    controller.mass.player_queues.set_shuffle.assert_awaited_once_with(PLAYER_ID, True)
217    provider.on_source_control.assert_not_awaited()
218
219
220async def test_repeat_reaches_the_live_source_before_the_queue() -> None:
221    """A live source is what is playing, so it gets the command and the queue does not."""
222    controller, provider = _controller(_source())
223    controller._get_player_with_redirect = MagicMock(return_value=controller.get_player(PLAYER_ID))
224    controller.get_active_queue = MagicMock(return_value=MagicMock())
225    controller.mass.player_queues.set_repeat = AsyncMock()
226
227    await controller.cmd_repeat(PLAYER_ID, RepeatMode.ALL)
228
229    provider.on_source_control.assert_awaited_once_with(
230        "main", SourceControl.REPEAT, RepeatMode.ALL
231    )
232    controller.mass.player_queues.set_repeat.assert_not_awaited()
233
234
235async def test_a_source_with_no_control_surface_refuses_cleanly() -> None:
236    """
237    A source that implements no controls at all is a refusal, not a server error.
238
239    vban_receiver has no on_source_control, so the call reaches the base
240    implementation and raises NotImplementedError. Ordering is not gated here, so
241    that path is reachable and a caller should get a refusal it can render.
242    """
243    controller, provider = _controller(_source())
244    provider.on_source_control = AsyncMock(side_effect=NotImplementedError)
245
246    with pytest.raises(PlayerCommandFailed, match="can not be controlled"):
247        await controller._forward_to_external_source(
248            controller.get_player(PLAYER_ID), SourceControl.SHUFFLE, True
249        )
250
251
252async def test_an_unknown_repeat_mode_is_refused_before_it_reaches_a_source() -> None:
253    """
254    UNKNOWN is what a source reports when it cannot say, not a mode to set.
255
256    Forwarding it asks a plugin to apply a non-mode: soloist raises a bare ValueError
257    on it, and other providers would silently accept and do nothing.
258    """
259    controller, provider = _controller(_source())
260    controller._get_player_with_redirect = MagicMock(return_value=controller.get_player(PLAYER_ID))
261
262    with pytest.raises(InvalidCommand, match="unknown repeat mode"):
263        await controller.cmd_repeat(PLAYER_ID, RepeatMode.UNKNOWN)
264
265    provider.on_source_control.assert_not_awaited()
266
267
268class _NativeSourcePlayer(MockPlayer):
269    """A player whose device runs a source of its own and orders that source itself."""
270
271    def __init__(self, provider: MockProvider, player_id: str, name: str) -> None:
272        super().__init__(provider, player_id, name)
273        self.shuffle_calls: list[bool] = []
274        self.repeat_calls: list[RepeatMode] = []
275
276    async def set_shuffle(self, shuffle_enabled: bool) -> None:
277        self.shuffle_calls.append(shuffle_enabled)
278
279    async def set_repeat(self, repeat_mode: RepeatMode) -> None:
280        self.repeat_calls.append(repeat_mode)
281
282
283def _native_source_controller(
284    *,
285    can_shuffle: bool = False,
286    can_repeat: bool = False,
287    active_source: str = NATIVE_SOURCE_ID,
288    queue: MagicMock | None = None,
289) -> tuple[PlayerController, _NativeSourcePlayer]:
290    """Build a controller whose player is playing a source its own device runs."""
291    mass = MagicMock()
292    mass.closing = False
293    mass.config.get_raw_core_config_value.return_value = "GLOBAL"
294    mass.config.get = MagicMock(return_value=[])
295    mass.signal_event = MagicMock()
296    controller = PlayerController(mass)
297    mass.players = controller
298    mass.player_queues = MagicMock()
299    mass.player_queues.get = MagicMock(return_value=queue)
300    provider = MockProvider("test_provider", instance_id="test", mass=mass)
301    player = _NativeSourcePlayer(provider, PLAYER_ID, "Player 1")
302    player._attr_source_list = [
303        PlayerSource(
304            id=NATIVE_SOURCE_ID,
305            name="Spotify",
306            can_shuffle=can_shuffle,
307            can_repeat=can_repeat,
308        )
309    ]
310    player._attr_active_source = active_source
311    # a device only counts as playing its own source while it is not idle
312    player._attr_playback_state = PlaybackState.PLAYING
313    player._cache.clear()
314    controller._players[PLAYER_ID] = player
315    player.update_state(signal_event=False)
316    return controller, player
317
318
319async def test_a_device_native_source_orders_its_own_content() -> None:
320    """A source the device runs itself has no session, so the player is asked directly."""
321    controller, player = _native_source_controller(can_shuffle=True, can_repeat=True)
322
323    await controller.cmd_shuffle(PLAYER_ID, shuffle_enabled=True)
324    await controller.cmd_repeat(PLAYER_ID, RepeatMode.ONE)
325
326    assert player.shuffle_calls == [True]
327    assert player.repeat_calls == [RepeatMode.ONE]
328
329
330@pytest.mark.parametrize(
331    ("command", "kwargs"),
332    [("cmd_shuffle", {"shuffle_enabled": True}), ("cmd_repeat", {"repeat_mode": RepeatMode.ALL})],
333)
334async def test_a_device_native_source_without_the_capability_is_refused(
335    command: str, kwargs: dict[str, Any]
336) -> None:
337    """A source that does not claim to order its own content is told no, not left waiting."""
338    controller, player = _native_source_controller()
339
340    with pytest.raises(PlayerCommandFailed, match="unavailable for this source"):
341        await getattr(controller, command)(PLAYER_ID, **kwargs)
342
343    assert not player.shuffle_calls
344    assert not player.repeat_calls
345
346
347async def test_a_command_aimed_at_a_source_that_stopped_is_refused() -> None:
348    """
349    Naming the source keeps a command off whatever took the player since.
350
351    A client builds its shuffle control against the source it is showing. If that
352    source ends before the click, Music Assistant's queue takes the player back -
353    and the setting would surprise the user whenever that queue next resumes.
354    """
355    queue = MagicMock()
356    queue.queue_id = PLAYER_ID
357    controller, player = _native_source_controller(
358        can_shuffle=True, active_source=PLAYER_ID, queue=queue
359    )
360    controller.mass.player_queues.set_shuffle = AsyncMock()  # type: ignore[method-assign]
361
362    with pytest.raises(PlayerCommandFailed, match="no longer playing"):
363        await controller.cmd_shuffle(PLAYER_ID, shuffle_enabled=True, source_id=NATIVE_SOURCE_ID)
364
365    controller.mass.player_queues.set_shuffle.assert_not_awaited()
366    assert not player.shuffle_calls
367
368
369async def test_a_command_aimed_at_the_queue_still_reaches_it() -> None:
370    """Naming the source it is showing does not stand in a client's way."""
371    queue = MagicMock()
372    queue.queue_id = PLAYER_ID
373    controller, _player = _native_source_controller(active_source=PLAYER_ID, queue=queue)
374    controller.mass.player_queues.set_shuffle = AsyncMock()  # type: ignore[method-assign]
375
376    await controller.cmd_shuffle(PLAYER_ID, shuffle_enabled=True, source_id=PLAYER_ID)
377
378    controller.mass.player_queues.set_shuffle.assert_awaited_once_with(PLAYER_ID, True)
379
380
381async def test_a_command_aimed_at_the_source_still_playing_is_delivered() -> None:
382    """The source named is the one playing, so the command goes through as usual."""
383    controller, player = _native_source_controller(can_repeat=True)
384
385    await controller.cmd_repeat(PLAYER_ID, RepeatMode.ALL, source_id=NATIVE_SOURCE_ID)
386
387    assert player.repeat_calls == [RepeatMode.ALL]
388
389
390async def test_a_command_aimed_at_a_live_source_reaches_its_session() -> None:
391    """
392    Naming a live source does not stand in the way of reaching its session.
393
394    The target is checked before anything is dispatched, so what a client reads as
395    the player's active source has to be the same string the session publishes.
396    """
397    source = _source()
398    controller, provider = _controller(source)
399    player = controller.get_player(PLAYER_ID)
400    player.state.active_source = source.uri
401    controller._get_player_with_redirect = MagicMock(return_value=player)
402
403    await controller.cmd_shuffle(PLAYER_ID, shuffle_enabled=True, source_id=source.uri)
404
405    provider.on_source_control.assert_awaited_once_with("main", SourceControl.SHUFFLE, True)
406
407
408async def test_a_command_aimed_at_a_live_source_that_ended_is_refused() -> None:
409    """A session that is gone gets nothing, and neither does the queue that took over."""
410    controller, provider = _controller(None)
411    player = controller.get_player(PLAYER_ID)
412    player.state.active_source = PLAYER_ID
413    controller._get_player_with_redirect = MagicMock(return_value=player)
414    queue = MagicMock()
415    queue.queue_id = PLAYER_ID
416    controller.get_active_queue = MagicMock(return_value=queue)
417    controller.mass.player_queues.set_shuffle = AsyncMock()
418
419    with pytest.raises(PlayerCommandFailed, match="no longer playing"):
420        await controller.cmd_shuffle(
421            PLAYER_ID, shuffle_enabled=True, source_id="spotify_connect--abc://audio_source/main"
422        )
423
424    provider.on_source_control.assert_not_awaited()
425    controller.mass.player_queues.set_shuffle.assert_not_awaited()
426