music-assistant-server

1"""
2Tests for queue-command delegation to a queue-capable AudioSource.
3
4While the queue's current item is a live (playing/paused) AudioSource declaring
5``queue_capabilities``, the external session owns the queue: shuffle/repeat/next/previous/
6seek are forwarded to the owning plugin (the mirrored options event updates the queue
7state afterwards) and ``queue_owner`` tells clients who owns the ordering. MA-owned tail
8items behind the source stay editable. A transport-only AudioSource (no
9``queue_capabilities``) keeps the exact pre-delegation behavior; playback state is
10deliberately not gated (a paused session may read as a stopped player).
11"""
12
13from __future__ import annotations
14
15import asyncio
16from typing import Any
17from unittest.mock import AsyncMock, MagicMock, Mock
18
19import pytest
20from music_assistant_models.enums import (
21    EventType,
22    PlaybackState,
23    ProviderFeature,
24    QueueOption,
25    RepeatMode,
26    SourceControl,
27)
28from music_assistant_models.errors import InvalidCommand
29from music_assistant_models.media_items import (
30    AudioSource,
31    ProviderMapping,
32    SourceQueueCapabilities,
33)
34from music_assistant_models.player_queue import PlayerQueue
35from music_assistant_models.queue_item import QueueItem
36
37from music_assistant.controllers.player_queues import PlayerQueuesController
38from music_assistant.controllers.player_queues.state import PlayerQueueData
39from music_assistant.models.plugin import PluginProvider
40
41QUEUE_ID = "q1"
42SOURCE_ID = "main"
43INSTANCE_ID = "spotify_connect--test"
44
45
46def _capabilities(**overrides: Any) -> SourceQueueCapabilities:
47    """Build a queue-capability declaration, with optional field overrides."""
48    kwargs: dict[str, Any] = {
49        "provider_domain": "spotify",
50        "can_shuffle": True,
51        "can_repeat": True,
52    }
53    kwargs.update(overrides)
54    return SourceQueueCapabilities(**kwargs)
55
56
57def _audio_source(
58    caps: SourceQueueCapabilities | None,
59    *,
60    can_seek: bool = True,
61    can_next_previous: bool = True,
62) -> AudioSource:
63    """Build the live AudioSource queue item payload with the given queue capabilities."""
64    return AudioSource(
65        item_id=SOURCE_ID,
66        provider=INSTANCE_ID,
67        name="Spotify Connect",
68        provider_mappings={
69            ProviderMapping(
70                item_id=SOURCE_ID,
71                provider_domain="spotify_connect",
72                provider_instance=INSTANCE_ID,
73            )
74        },
75        can_play_pause=True,
76        can_seek=can_seek,
77        can_next_previous=can_next_previous,
78        queue_capabilities=caps,
79    )
80
81
82def _controller(
83    current_item: AudioSource | None, **queue_kwargs: Any
84) -> tuple[PlayerQueuesController, MagicMock]:
85    """
86    Build a bare controller with a single queue playing the given AudioSource.
87
88    Returns the controller and the (spec'd) PluginProvider mock owning the source, so
89    tests can assert what was (not) forwarded through ``on_source_control``.
90    """
91    ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
92    ctrl.logger = MagicMock()
93    ctrl.mass = MagicMock()
94    ctrl.mass.create_task = Mock(
95        side_effect=lambda coro, **_kw: coro.close() if asyncio.iscoroutine(coro) else MagicMock()
96    )
97    lock_cm = MagicMock()
98    lock_cm.__aenter__ = AsyncMock(return_value=None)
99    lock_cm.__aexit__ = AsyncMock(return_value=None)
100    ctrl.mass.players.get_player_lock = Mock(return_value=lock_cm)
101    ctrl.mass.players.get_player = Mock(return_value=MagicMock(extra_data={}))
102    ctrl.on_player_update = Mock()  # type: ignore[method-assign]
103    ctrl.play_index = AsyncMock()  # type: ignore[method-assign]
104    ctrl.load = AsyncMock()  # type: ignore[method-assign]
105    ctrl._smart_shuffle = Mock()
106    ctrl._smart_shuffle.is_enabled = Mock(return_value=False)
107    ctrl._managed_pool = MagicMock()
108    queue = PlayerQueue(
109        queue_id=QUEUE_ID, active=True, display_name="Q1", available=True, items=0, **queue_kwargs
110    )
111    queue_data = PlayerQueueData(queue=queue)
112    ctrl._queue_data = {QUEUE_ID: queue_data}
113    if current_item is not None:
114        item = QueueItem.from_media_item(QUEUE_ID, current_item)
115        queue_data.items = [item]
116        queue.items = 1
117        queue.current_index = 0
118        queue.current_item = item
119        queue.state = PlaybackState.PLAYING
120    provider = MagicMock(spec=PluginProvider)
121    provider.supported_features = {ProviderFeature.AUDIO_SOURCE}
122    ctrl.mass.get_provider = Mock(return_value=provider)
123    return ctrl, provider
124
125
126def _queue(ctrl: PlayerQueuesController) -> PlayerQueue:
127    """Return the controller's queue."""
128    return ctrl._queue_data[QUEUE_ID].queue
129
130
131def _tail_item(item_id: str) -> QueueItem:
132    """Build a plain MA-owned queue item to place behind the AudioSource."""
133    return QueueItem(queue_id=QUEUE_ID, queue_item_id=item_id, name=item_id, duration=100)
134
135
136def _add_tail_items(ctrl: PlayerQueuesController, *item_ids: str) -> None:
137    """Append MA-owned items behind the current AudioSource item."""
138    queue_data = ctrl._queue_data[QUEUE_ID]
139    queue_data.items = [*queue_data.items, *(_tail_item(item_id) for item_id in item_ids)]
140    queue_data.queue.items = len(queue_data.items)
141
142
143async def test_delegated_shuffle_forwards_without_touching_the_queue() -> None:
144    """Shuffle on a delegated queue reaches the plugin; the MA queue state is left alone."""
145    ctrl, provider = _controller(_audio_source(_capabilities()))
146
147    await ctrl.set_shuffle(QUEUE_ID, True)
148
149    provider.on_source_control.assert_awaited_once_with(SOURCE_ID, SourceControl.SHUFFLE, True)
150    # the mirrored options event updates the state, not the command itself
151    assert _queue(ctrl).shuffle_enabled is False
152    ctrl.load.assert_not_awaited()  # type: ignore[attr-defined]
153
154
155async def test_delegated_shuffle_with_the_mirrored_value_still_forwards() -> None:
156    """
157    Setting shuffle to the already-mirrored value still forwards to the session.
158
159    The mirrored state lags the session's options echo, so an equality no-op check
160    would drop a quick second toggle; the session itself is the deduplicating end.
161    """
162    ctrl, provider = _controller(_audio_source(_capabilities()))
163    _queue(ctrl).shuffle_enabled = True
164
165    await ctrl.set_shuffle(QUEUE_ID, True)
166
167    provider.on_source_control.assert_awaited_once_with(SOURCE_ID, SourceControl.SHUFFLE, True)
168    ctrl.load.assert_not_awaited()  # type: ignore[attr-defined]
169
170
171async def test_delegated_repeat_forwards_the_repeat_mode() -> None:
172    """Repeat on a delegated queue reaches the plugin with the RepeatMode as payload."""
173    ctrl, provider = _controller(_audio_source(_capabilities()))
174
175    await ctrl.set_repeat(QUEUE_ID, RepeatMode.ALL)
176
177    provider.on_source_control.assert_awaited_once_with(
178        SOURCE_ID, SourceControl.REPEAT, RepeatMode.ALL
179    )
180    assert _queue(ctrl).repeat_mode == RepeatMode.OFF
181
182
183async def test_delegated_repeat_with_the_mirrored_value_still_forwards() -> None:
184    """
185    Setting repeat to the already-mirrored mode still forwards to the session.
186
187    The mirrored state lags the session's options echo, so an equality no-op check
188    would drop a quick second change; the session itself is the deduplicating end.
189    """
190    ctrl, provider = _controller(_audio_source(_capabilities()))
191    _queue(ctrl).repeat_mode = RepeatMode.ALL
192
193    await ctrl.set_repeat(QUEUE_ID, RepeatMode.ALL)
194
195    provider.on_source_control.assert_awaited_once_with(
196        SOURCE_ID, SourceControl.REPEAT, RepeatMode.ALL
197    )
198
199
200async def test_delegated_repeat_unknown_is_refused() -> None:
201    """An UNKNOWN repeat mode is rejected before anything reaches the session."""
202    ctrl, provider = _controller(_audio_source(_capabilities()))
203
204    with pytest.raises(InvalidCommand):
205        await ctrl.set_repeat(QUEUE_ID, RepeatMode.UNKNOWN)
206
207    provider.on_source_control.assert_not_awaited()
208
209
210async def test_shuffle_refused_when_the_session_cannot_shuffle() -> None:
211    """A session without shuffle support refuses the toggle instead of mutating the queue."""
212    ctrl, provider = _controller(_audio_source(_capabilities(can_shuffle=False)))
213
214    with pytest.raises(InvalidCommand):
215        await ctrl.set_shuffle(QUEUE_ID, True)
216
217    provider.on_source_control.assert_not_awaited()
218    assert _queue(ctrl).shuffle_enabled is False
219
220
221async def test_repeat_refused_when_the_session_cannot_repeat() -> None:
222    """A session without repeat support refuses the command instead of mutating the queue."""
223    ctrl, provider = _controller(_audio_source(_capabilities(can_repeat=False)))
224
225    with pytest.raises(InvalidCommand):
226        await ctrl.set_repeat(QUEUE_ID, RepeatMode.ALL)
227
228    provider.on_source_control.assert_not_awaited()
229    assert _queue(ctrl).repeat_mode == RepeatMode.OFF
230
231
232async def test_idle_queue_stays_delegated() -> None:
233    """
234    An IDLE queue with the session item current still forwards to the plugin.
235
236    Spotify Connect maps a paused session onto a stopped player, so the queue may
237    read IDLE while the session is alive and resumable; a genuinely dead session is
238    the plugin's call (it raises its localized not-active error) — same behavior as
239    the player-layer transport proxy.
240    """
241    ctrl, provider = _controller(_audio_source(_capabilities()))
242    _queue(ctrl).state = PlaybackState.IDLE
243
244    await ctrl.set_shuffle(QUEUE_ID, True)
245    ctrl.signal_update(QUEUE_ID)
246
247    provider.on_source_control.assert_awaited_once_with(SOURCE_ID, SourceControl.SHUFFLE, True)
248    assert _queue(ctrl).shuffle_enabled is False
249    assert _queue(ctrl).queue_owner is not None
250
251
252async def test_delegated_next_forwards_instead_of_walking_the_queue() -> None:
253    """player_queues/next on a delegated queue skips within the session."""
254    ctrl, provider = _controller(_audio_source(_capabilities()))
255
256    await ctrl.next(QUEUE_ID)
257
258    provider.on_source_control.assert_awaited_once_with(SOURCE_ID, SourceControl.NEXT)
259    # no MA index walk: the session is the only item and stays current
260    assert _queue(ctrl).current_index == 0
261    ctrl.play_index.assert_not_awaited()  # type: ignore[attr-defined]
262
263
264async def test_delegated_previous_forwards_instead_of_walking_the_queue() -> None:
265    """player_queues/previous on a delegated queue skips within the session."""
266    ctrl, provider = _controller(_audio_source(_capabilities()))
267
268    await ctrl.previous(QUEUE_ID)
269
270    provider.on_source_control.assert_awaited_once_with(SOURCE_ID, SourceControl.PREVIOUS)
271    assert _queue(ctrl).current_index == 0
272    ctrl.play_index.assert_not_awaited()  # type: ignore[attr-defined]
273
274
275async def test_next_and_previous_refused_when_the_session_cannot_skip() -> None:
276    """A session without skip support refuses next/previous instead of restarting the stream."""
277    ctrl, provider = _controller(_audio_source(_capabilities(), can_next_previous=False))
278
279    with pytest.raises(InvalidCommand):
280        await ctrl.next(QUEUE_ID)
281    with pytest.raises(InvalidCommand):
282        await ctrl.previous(QUEUE_ID)
283
284    provider.on_source_control.assert_not_awaited()
285    ctrl.play_index.assert_not_awaited()  # type: ignore[attr-defined]
286
287
288async def test_delegated_seek_forwards_without_requiring_a_duration() -> None:
289    """Seek forwards the absolute position even though the live source has no duration."""
290    ctrl, provider = _controller(_audio_source(_capabilities()))
291
292    await ctrl.seek(QUEUE_ID, 42)
293
294    provider.on_source_control.assert_awaited_once_with(SOURCE_ID, SourceControl.SEEK, 42)
295    ctrl.play_index.assert_not_awaited()  # type: ignore[attr-defined]
296
297
298async def test_delegated_seek_publishes_the_seek_target() -> None:
299    """The forwarded seek also publishes the target position so progress doesn't snap back."""
300    ctrl, _provider = _controller(_audio_source(_capabilities()))
301
302    await ctrl.seek(QUEUE_ID, 42)
303
304    queue = _queue(ctrl)
305    assert queue.elapsed_time == 42
306    assert queue.elapsed_time_last_updated > 0
307    assert any(
308        call.args and call.args[0] == EventType.QUEUE_UPDATED
309        for call in ctrl.mass.signal_event.call_args_list  # type: ignore[attr-defined]
310    )
311
312
313async def test_seek_refused_when_the_session_cannot_seek() -> None:
314    """A session without seek support refuses the command instead of forwarding it."""
315    ctrl, provider = _controller(_audio_source(_capabilities(), can_seek=False))
316
317    with pytest.raises(InvalidCommand):
318        await ctrl.seek(QUEUE_ID, 42)
319
320    provider.on_source_control.assert_not_awaited()
321
322
323async def test_delegated_skip_forwards_the_absolute_position() -> None:
324    """Skip translates its relative offset to an absolute in-session seek."""
325    ctrl, provider = _controller(_audio_source(_capabilities()))
326    _queue(ctrl).elapsed_time = 5
327
328    await ctrl.skip(QUEUE_ID, 10)
329
330    provider.on_source_control.assert_awaited_once_with(SOURCE_ID, SourceControl.SEEK, 15)
331
332
333async def test_tail_items_stay_editable_while_delegated() -> None:
334    """MA-owned items behind the AudioSource can still be moved and deleted."""
335    ctrl, _provider = _controller(_audio_source(_capabilities()))
336    _add_tail_items(ctrl, "t1", "t2")
337
338    ctrl.move_item(QUEUE_ID, "t1", 1)
339    assert [item.queue_item_id for item in ctrl._queue_data[QUEUE_ID].items[1:]] == ["t2", "t1"]
340
341    ctrl.move_item_end(QUEUE_ID, "t2")
342    assert ctrl._queue_data[QUEUE_ID].items[-1].queue_item_id == "t2"
343
344    ctrl.delete_item(QUEUE_ID, "t1")
345    assert [item.queue_item_id for item in ctrl._queue_data[QUEUE_ID].items[1:]] == ["t2"]
346
347
348async def test_queue_owner_set_while_delegated() -> None:
349    """The emitted queue carries the owning AudioSource uri while delegated."""
350    source = _audio_source(_capabilities())
351    ctrl, _provider = _controller(source)
352
353    ctrl.signal_update(QUEUE_ID)
354
355    assert _queue(ctrl).queue_owner == source.uri
356    # the QUEUE_UPDATED event carries the owner for clients
357    event_call = next(
358        call
359        for call in ctrl.mass.signal_event.call_args_list  # type: ignore[attr-defined]
360        if call.args and call.args[0] == EventType.QUEUE_UPDATED
361    )
362    assert event_call.kwargs["data"].queue_owner == source.uri
363
364
365async def test_queue_owner_cleared_for_a_transport_only_source() -> None:
366    """A transport-only AudioSource does not delegate, so the queue stays MA-owned."""
367    ctrl, _provider = _controller(_audio_source(None))
368
369    ctrl.signal_update(QUEUE_ID)
370
371    assert _queue(ctrl).queue_owner is None
372
373
374async def test_queue_owner_cleared_when_the_provider_is_gone() -> None:
375    """A vanished plugin provider ends the delegation, whatever the item declares."""
376    ctrl, _provider = _controller(_audio_source(_capabilities()))
377    ctrl.mass.get_provider = Mock(return_value=None)  # type: ignore[method-assign]
378
379    ctrl.signal_update(QUEUE_ID)
380
381    assert _queue(ctrl).queue_owner is None
382
383
384async def test_signal_update_refreshes_the_derived_shuffle_flag() -> None:
385    """A stale smart_shuffle_active flag is corrected by the next signal_update."""
386    ctrl, _provider = _controller(_audio_source(_capabilities()))
387    ctrl._smart_shuffle.is_enabled = Mock(return_value=True)  # type: ignore[method-assign]
388    queue = _queue(ctrl)
389    # simulate the streams controller mirroring a session-side shuffle enable: the raw
390    # flag flips directly, leaving the derived flag stale until the next signal
391    queue.shuffle_enabled = True
392    assert queue.smart_shuffle_active is False
393
394    ctrl.signal_update(QUEUE_ID)
395
396    assert queue.smart_shuffle_active is True
397
398
399async def test_transport_only_source_keeps_normal_queue_behavior() -> None:
400    """Regression: without queue_capabilities nothing is forwarded to the plugin."""
401    ctrl, provider = _controller(_audio_source(None))
402
403    await ctrl.set_shuffle(QUEUE_ID, True)
404    await ctrl.set_repeat(QUEUE_ID, RepeatMode.ALL)
405    await ctrl.next(QUEUE_ID)
406    ctrl.delete_item(QUEUE_ID, 0)
407
408    provider.on_source_control.assert_not_awaited()
409    # the commands applied to the MA queue itself
410    assert _queue(ctrl).shuffle_enabled is True
411    assert _queue(ctrl).repeat_mode == RepeatMode.ALL
412    assert ctrl._queue_data[QUEUE_ID].items == []
413
414
415async def test_ordered_play_reorders_the_tail_locally_while_delegated() -> None:
416    """An ordered play on a delegated queue unshuffles the MA tail without forwarding."""
417    ctrl, provider = _controller(_audio_source(_capabilities()))
418    _add_tail_items(ctrl, "t1", "t2")
419    queue_data = ctrl._queue_data[QUEUE_ID]
420    # the tail sits in scattered order (mirrored session shuffle was on during an ADD)
421    queue_data.items[1].sort_index = 5
422    queue_data.items[2].sort_index = 4
423    queue = _queue(ctrl)
424    queue.shuffle_enabled = True
425
426    await ctrl._apply_shuffle(QUEUE_ID, QueueOption.PLAY, False)
427
428    # the session's own shuffle is not touched; the MA-owned tail is restored locally
429    provider.on_source_control.assert_not_awaited()
430    assert queue.shuffle_enabled is False
431    ctrl.load.assert_awaited_once()  # type: ignore[attr-defined]
432    kwargs = ctrl.load.await_args.kwargs  # type: ignore[attr-defined]
433    assert [item.queue_item_id for item in kwargs["queue_items"]] == ["t2", "t1"]
434    assert kwargs["shuffle"] is False
435
436
437async def test_delegated_seek_skips_the_progress_publish_when_the_item_changed() -> None:
438    """A concurrent item change during the seek forward must not get a stale progress stamp."""
439    ctrl, provider = _controller(_audio_source(_capabilities()))
440    queue = _queue(ctrl)
441
442    async def _swap_item(*_args: Any, **_kwargs: Any) -> None:
443        queue.current_item = _tail_item("other")
444
445    provider.on_source_control.side_effect = _swap_item
446    await ctrl.seek(QUEUE_ID, 42)
447
448    assert queue.elapsed_time == 0
449
450
451async def test_mirrored_shuffle_applies_under_revalidated_delegation() -> None:
452    """The scheduled mirror task re-validates the delegation and applies the tail shuffle."""
453    ctrl, _provider = _controller(_audio_source(_capabilities()))
454    _add_tail_items(ctrl, "t1", "t2")
455
456    await ctrl._apply_mirrored_shuffle(QUEUE_ID, SOURCE_ID, INSTANCE_ID, True)
457
458    assert _queue(ctrl).shuffle_enabled is True
459    ctrl.load.assert_awaited_once()  # type: ignore[attr-defined]
460    assert ctrl.load.await_args.kwargs["shuffle"] is True  # type: ignore[attr-defined]
461
462
463async def test_mirrored_shuffle_skipped_when_the_session_changed() -> None:
464    """A mirror task for a source that is no longer current must not touch the queue."""
465    ctrl, _provider = _controller(_audio_source(_capabilities()))
466
467    await ctrl._apply_mirrored_shuffle(QUEUE_ID, "other_source", INSTANCE_ID, True)
468
469    assert _queue(ctrl).shuffle_enabled is False
470    ctrl.load.assert_not_awaited()  # type: ignore[attr-defined]
471
472
473async def test_mirrored_shuffle_skipped_when_already_mirrored() -> None:
474    """A mirror task carrying the already-applied value is a no-op."""
475    ctrl, _provider = _controller(_audio_source(_capabilities()))
476    _queue(ctrl).shuffle_enabled = True
477
478    await ctrl._apply_mirrored_shuffle(QUEUE_ID, SOURCE_ID, INSTANCE_ID, True)
479
480    ctrl.load.assert_not_awaited()  # type: ignore[attr-defined]
481