music-assistant-server

1"""
2Tests for the enqueue-option handling in the player-queues controller.
3
4These exercise ``QueueLoaderMixin._enqueue_with_option`` against a bare controller
5instance, mirroring ``test_user_initiated_plays``. The real ``load`` and
6``update_items`` run so the insert position is verified end-to-end; only the
7side-effecting ``play_index`` and ``signal_update`` are stubbed out.
8"""
9
10from __future__ import annotations
11
12from unittest.mock import AsyncMock, Mock
13
14from music_assistant_models.enums import MediaType, PlaybackState, QueueOption
15from music_assistant_models.media_items import (
16    Album,
17    ItemMapping,
18    MediaItemType,
19    Podcast,
20    ProviderMapping,
21    Radio,
22    Track,
23)
24from music_assistant_models.player_queue import PlayerQueue
25from music_assistant_models.queue_item import QueueItem
26from music_assistant_models.unique_list import UniqueList
27
28from music_assistant.controllers.player_queues import PlayerQueuesController
29from music_assistant.controllers.player_queues.state import PlayerQueueData
30
31
32def _controller() -> PlayerQueuesController:
33    """Create a bare controller instance with the noisy ``signal_update`` stubbed out."""
34    ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
35    ctrl.signal_update = Mock()  # type: ignore[method-assign]
36    return ctrl
37
38
39def _items(queue_id: str, names: list[str]) -> list[QueueItem]:
40    """Build a list of simple queue items with the given names."""
41    return [
42        QueueItem(queue_id=queue_id, queue_item_id=name, name=name, duration=60) for name in names
43    ]
44
45
46def _track(item_id: str) -> Track:
47    """Build a playable Track on the 'test' provider (mirrors the managed-pool test helper)."""
48    return Track(
49        item_id=item_id,
50        provider="test",
51        name=f"Track {item_id}",
52        duration=60,
53        artists=UniqueList(
54            [ItemMapping(item_id="a", provider="test", name="A", media_type=MediaType.ARTIST)]
55        ),
56        provider_mappings={
57            ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
58        },
59    )
60
61
62def _album(item_id: str) -> Album:
63    """Build an Album on the 'test' provider (a container source, kept in the wire `sources`)."""
64    return Album(
65        item_id=item_id,
66        provider="test",
67        name=f"Album {item_id}",
68        provider_mappings={
69            ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
70        },
71    )
72
73
74def _podcast(item_id: str) -> Podcast:
75    """Build a Podcast on the 'test' provider (a container source, kept in the wire `sources`)."""
76    return Podcast(
77        item_id=item_id,
78        provider="test",
79        name=f"Podcast {item_id}",
80        provider_mappings={
81            ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
82        },
83    )
84
85
86def _radio(item_id: str) -> Radio:
87    """Build a Radio on the 'test' provider (an individual item, omitted from the wire `sources`)."""
88    return Radio(
89        item_id=item_id,
90        provider="test",
91        name=f"Radio {item_id}",
92        provider_mappings={
93            ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
94        },
95    )
96
97
98async def test_play_on_idle_queue_starts_at_first_item() -> None:
99    """PLAY with multiple items onto an idle/empty queue plays the first item, not the second."""
100    ctrl = _controller()
101    play_index = AsyncMock()
102    ctrl.play_index = play_index  # type: ignore[method-assign]
103    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
104    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
105    items = _items("q1", ["a", "b", "c"])
106
107    await ctrl._enqueue_with_option("q1", items, QueueOption.PLAY)
108
109    # the items are loaded from the very start and playback begins at the first one
110    assert ctrl._queue_data["q1"].items[0] is items[0]
111    assert len(ctrl._queue_data["q1"].items) == 3
112    play_index.assert_awaited_once_with("q1", 0)
113
114
115async def test_play_on_active_queue_inserts_after_current() -> None:
116    """PLAY on an active queue keeps inserting right after the current index and jumps to it."""
117    ctrl = _controller()
118    play_index = AsyncMock()
119    ctrl.play_index = play_index  # type: ignore[method-assign]
120    ctrl.get_next_item = Mock(return_value=None)  # type: ignore[method-assign]
121    existing = _items("q1", ["e0", "e1", "e2"])
122    queue = PlayerQueue(
123        queue_id="q1",
124        active=True,
125        display_name="Q1",
126        available=True,
127        items=len(existing),
128        state=PlaybackState.PLAYING,
129        current_index=2,
130        index_in_buffer=2,
131    )
132    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue, items=list(existing))}
133    new_items = _items("q1", ["n0", "n1"])
134
135    await ctrl._enqueue_with_option("q1", new_items, QueueOption.PLAY)
136
137    # inserted right after the current/buffered index (2) and playback jumps there
138    assert ctrl._queue_data["q1"].items[3] is new_items[0]
139    play_index.assert_awaited_once_with("q1", 3)
140
141
142async def test_next_shuffle_pins_first_item_and_shuffles_rest() -> None:
143    """NEXT with shuffle on pins the first new item after the buffered index, shuffles the rest."""
144    ctrl = _controller()
145    ctrl.play_index = AsyncMock()  # type: ignore[method-assign]
146    ctrl.get_next_item = Mock(return_value=None)  # type: ignore[method-assign]
147    # keep the shuffle deterministic-enough: pure random of the "rest", first item stays pinned
148    ctrl._smart_shuffle = Mock()
149    ctrl._smart_shuffle.is_enabled = Mock(return_value=False)
150    existing = _items("q1", ["e0", "e1", "e2"])
151    queue = PlayerQueue(
152        queue_id="q1",
153        active=True,
154        display_name="Q1",
155        available=True,
156        items=len(existing),
157        state=PlaybackState.PLAYING,
158        current_index=0,
159        index_in_buffer=0,
160        shuffle_enabled=True,
161    )
162    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue, items=list(existing))}
163    new_items = _items("q1", ["n0", "n1"])
164
165    await ctrl._enqueue_with_option("q1", new_items, QueueOption.NEXT)
166
167    items = ctrl._queue_data["q1"].items
168    # the first new item is pinned right after the buffered index (0) so it plays next
169    assert items[0] is existing[0]
170    assert items[1] is new_items[0]
171    # the rest of the batch and the existing unplayed tail are shuffled together behind it
172    assert {item.queue_item_id for item in items[2:]} == {"n1", "e1", "e2"}
173    ctrl.play_index.assert_not_awaited()
174
175
176async def test_next_shuffle_single_item_keeps_tail_order() -> None:
177    """A single-item NEXT inserts at the buffered index+1 without re-arranging the tail."""
178    ctrl = _controller()
179    ctrl.play_index = AsyncMock()  # type: ignore[method-assign]
180    ctrl.get_next_item = Mock(return_value=None)  # type: ignore[method-assign]
181    existing = _items("q1", ["e0", "e1", "e2"])
182    queue = PlayerQueue(
183        queue_id="q1",
184        active=True,
185        display_name="Q1",
186        available=True,
187        items=len(existing),
188        state=PlaybackState.PLAYING,
189        current_index=0,
190        index_in_buffer=0,
191        shuffle_enabled=True,
192    )
193    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue, items=list(existing))}
194    new_items = _items("q1", ["n0"])
195
196    await ctrl._enqueue_with_option("q1", new_items, QueueOption.NEXT)
197
198    # inserted right after the buffered index; the existing tail keeps its order (no reshuffle)
199    assert [item.queue_item_id for item in ctrl._queue_data["q1"].items] == ["e0", "n0", "e1", "e2"]
200    ctrl.play_index.assert_not_awaited()
201
202
203async def test_next_on_empty_queue_sets_current_index_without_playing() -> None:
204    """NEXT onto an empty queue stages the items and points the current index at the first one."""
205    ctrl = _controller()
206    ctrl.play_index = AsyncMock()  # type: ignore[method-assign]
207    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
208    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
209    items = _items("q1", ["a", "b", "c"])
210
211    await ctrl._enqueue_with_option("q1", items, QueueOption.NEXT)
212
213    assert [item.queue_item_id for item in ctrl._queue_data["q1"].items] == ["a", "b", "c"]
214    assert queue.current_index == 0
215    assert queue.current_item is items[0]
216    ctrl.play_index.assert_not_awaited()
217
218
219async def test_replace_next_on_empty_queue_sets_current_index_without_playing() -> None:
220    """REPLACE_NEXT onto an empty queue stages the items and sets the current index."""
221    ctrl = _controller()
222    ctrl.play_index = AsyncMock()  # type: ignore[method-assign]
223    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
224    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
225    items = _items("q1", ["a", "b"])
226
227    await ctrl._enqueue_with_option("q1", items, QueueOption.REPLACE_NEXT)
228
229    assert [item.queue_item_id for item in ctrl._queue_data["q1"].items] == ["a", "b"]
230    assert queue.current_index == 0
231    assert queue.current_item is items[0]
232    ctrl.play_index.assert_not_awaited()
233
234
235async def test_next_on_idle_queue_with_content_keeps_current_index() -> None:
236    """NEXT on an idle queue that has content leaves the current index and inserts after it."""
237    ctrl = _controller()
238    ctrl.play_index = AsyncMock()  # type: ignore[method-assign]
239    existing = _items("q1", ["e0", "e1", "e2"])
240    queue = PlayerQueue(
241        queue_id="q1",
242        active=True,
243        display_name="Q1",
244        available=True,
245        items=len(existing),
246        state=PlaybackState.IDLE,
247        current_index=1,
248    )
249    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue, items=list(existing))}
250    new_items = _items("q1", ["n0"])
251
252    await ctrl._enqueue_with_option("q1", new_items, QueueOption.NEXT)
253
254    items = ctrl._queue_data["q1"].items
255    # current index is untouched; the new item is inserted right after it
256    assert queue.current_index == 1
257    assert [item.queue_item_id for item in items] == ["e0", "e1", "n0", "e2"]
258    ctrl.play_index.assert_not_awaited()
259
260
261def _dynamic_controller() -> PlayerQueuesController:
262    """Build a bare controller wired to drive ``_enter_dynamic_mode`` with a stubbed managed pool."""
263    ctrl = _controller()
264    ctrl.get_next_item = Mock(return_value=None)  # type: ignore[method-assign]
265    ctrl.is_smart_shuffle_active = Mock(return_value=True)  # type: ignore[method-assign]
266    ctrl._managed_pool = Mock()
267    ctrl._managed_pool.fill = AsyncMock(return_value=[_track("p0"), _track("p1")])
268    return ctrl
269
270
271async def test_enter_dynamic_mode_add_on_idle_does_not_start_playback() -> None:
272    """ADD of a dynamic source onto an idle/empty queue stages the pool but does not start playing."""
273    ctrl = _dynamic_controller()
274    play_index = AsyncMock()
275    ctrl.play_index = play_index  # type: ignore[method-assign]
276    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
277    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
278
279    await ctrl._enter_dynamic_mode("q1", QueueOption.ADD)
280
281    # the pool is loaded and the current item is set, but playback is not started
282    items = ctrl._queue_data["q1"].items
283    assert {item.media_item.item_id for item in items if item.media_item is not None} == {
284        "p0",
285        "p1",
286    }
287    assert queue.current_index == 0
288    assert queue.current_item is not None
289    play_index.assert_not_awaited()
290
291
292async def test_enter_dynamic_mode_play_on_idle_starts_playback() -> None:
293    """PLAY of a dynamic source onto an idle/empty queue starts playback on the rebuilt pool."""
294    ctrl = _dynamic_controller()
295    play_index = AsyncMock()
296    ctrl.play_index = play_index  # type: ignore[method-assign]
297    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
298    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
299
300    await ctrl._enter_dynamic_mode("q1", QueueOption.PLAY)
301
302    assert len(ctrl._queue_data["q1"].items) == 2
303    play_index.assert_awaited_once_with("q1", 0)
304
305
306async def test_enter_dynamic_mode_add_on_active_keeps_current_and_rebuilds_tail() -> None:
307    """ADD of a dynamic source onto a playing queue rebuilds the tail without interrupting it."""
308    ctrl = _dynamic_controller()
309    play_index = AsyncMock()
310    ctrl.play_index = play_index  # type: ignore[method-assign]
311    existing = _items("q1", ["e0", "e1", "e2"])
312    queue = PlayerQueue(
313        queue_id="q1",
314        active=True,
315        display_name="Q1",
316        available=True,
317        items=len(existing),
318        state=PlaybackState.PLAYING,
319        current_index=1,
320        index_in_buffer=1,
321    )
322    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue, items=list(existing))}
323
324    await ctrl._enter_dynamic_mode("q1", QueueOption.ADD)
325
326    items = ctrl._queue_data["q1"].items
327    # the current track and history are kept; the finite tail behind it is replaced by the pool
328    assert [item.queue_item_id for item in items[:2]] == ["e0", "e1"]
329    assert {item.media_item.item_id for item in items[2:] if item.media_item is not None} == {
330        "p0",
331        "p1",
332    }
333    assert queue.current_index == 1
334    play_index.assert_not_awaited()
335
336
337async def test_enter_dynamic_mode_replaces_old_pool_tail_stays_bounded() -> None:
338    """Re-building on an already-dynamic queue drops the whole old pool tail, keeping it bounded."""
339    ctrl = _dynamic_controller()
340    play_index = AsyncMock()
341    ctrl.play_index = play_index  # type: ignore[method-assign]
342    # current + buffered (e0, e1) followed by a large existing pool tail (o0..o9)
343    existing = _items("q1", ["e0", "e1", *[f"o{i}" for i in range(10)]])
344    queue = PlayerQueue(
345        queue_id="q1",
346        active=True,
347        display_name="Q1",
348        available=True,
349        items=len(existing),
350        state=PlaybackState.PLAYING,
351        current_index=1,
352        index_in_buffer=1,
353    )
354    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue, items=list(existing))}
355
356    await ctrl._enter_dynamic_mode("q1", QueueOption.ADD)
357
358    items = ctrl._queue_data["q1"].items
359    # old 10-track tail is dropped and replaced by the fresh bounded pool (2 + 2, not 2 + 10 + 2)
360    assert [item.queue_item_id for item in items[:2]] == ["e0", "e1"]
361    assert {item.media_item.item_id for item in items[2:] if item.media_item is not None} == {
362        "p0",
363        "p1",
364    }
365    assert len(items) == 4
366    play_index.assert_not_awaited()
367
368
369async def test_enter_dynamic_mode_rebuilds_from_buffer_index() -> None:
370    """The rebuild keeps the already-buffered next track and only replaces what is behind it."""
371    ctrl = _dynamic_controller()
372    play_index = AsyncMock()
373    ctrl.play_index = play_index  # type: ignore[method-assign]
374    # current at 1, but the player has already buffered index 2, so it must be kept
375    existing = _items("q1", ["e0", "e1", "e2", "o0", "o1"])
376    queue = PlayerQueue(
377        queue_id="q1",
378        active=True,
379        display_name="Q1",
380        available=True,
381        items=len(existing),
382        state=PlaybackState.PLAYING,
383        current_index=1,
384        index_in_buffer=2,
385    )
386    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue, items=list(existing))}
387
388    await ctrl._enter_dynamic_mode("q1", QueueOption.ADD)
389
390    items = ctrl._queue_data["q1"].items
391    # kept through the buffered index (e0, e1, e2); everything after it rebuilt from the pool
392    assert [item.queue_item_id for item in items[:3]] == ["e0", "e1", "e2"]
393    assert {item.media_item.item_id for item in items[3:] if item.media_item is not None} == {
394        "p0",
395        "p1",
396    }
397    play_index.assert_not_awaited()
398
399
400def test_store_sources_dedupes_wire_sources_keeps_internal_multiplicity() -> None:
401    """The wire `sources` list is deduped per source; the server keeps every occurrence."""
402    ctrl = _controller()
403    ctrl._managed_pool = Mock()
404    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
405    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
406    a, b = _album("a"), _album("b")
407
408    # "a" added twice (multiplicity 2), "b" once
409    ctrl.store_sources(queue, [a, b, a])
410
411    # server-side list keeps every occurrence (drives the managed-pool weighting)
412    assert ctrl._queue_data["q1"].source_items == [a, b, a]
413    # wire list clients see is deduped to the distinct sources, order preserved
414    assert [mapping.uri for mapping in queue.sources] == [a.uri, b.uri]
415
416
417def test_store_sources_keeps_only_container_types_on_wire() -> None:
418    """Container sources reach the wire `sources`; individual items are omitted (kept server-side)."""
419    ctrl = _controller()
420    ctrl._managed_pool = Mock()
421    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
422    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
423    album, podcast = _album("al"), _podcast("po")
424    track, radio = _track("t"), _radio("r")
425    items: list[MediaItemType] = [album, track, podcast, radio]
426
427    ctrl.store_sources(queue, items)
428
429    # the full set (incl. the individual items) is retained server-side for pool weighting / seeds
430    assert ctrl._queue_data["q1"].source_items == items
431    # but only the container sources are shown to clients, in original order
432    assert [mapping.uri for mapping in queue.sources] == [album.uri, podcast.uri]
433
434
435def _shuffled_queue(ctrl: PlayerQueuesController) -> PlayerQueue:
436    """Set up an idle queue with shuffle on and a deterministic (reversing) shuffle."""
437    ctrl._smart_shuffle = Mock()
438    ctrl._smart_shuffle.is_enabled = Mock(return_value=True)
439    ctrl._smart_shuffle.arrange = AsyncMock(
440        side_effect=lambda _queue, items: list(items)[::-1],
441    )
442    queue = PlayerQueue(
443        queue_id="q1",
444        active=True,
445        display_name="Q1",
446        available=True,
447        items=0,
448        shuffle_enabled=True,
449    )
450    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
451    return queue
452
453
454async def test_play_with_start_item_pins_chosen_track_under_shuffle() -> None:
455    """PLAY from a chosen track keeps that track first when shuffle is on."""
456    ctrl = _controller()
457    play_index = AsyncMock()
458    ctrl.play_index = play_index  # type: ignore[method-assign]
459    _shuffled_queue(ctrl)
460    items = _items("q1", ["chosen", "b", "c", "d"])
461
462    await ctrl._enqueue_with_option("q1", items, QueueOption.PLAY, pin_first=True)
463
464    queue_items = ctrl._queue_data["q1"].items
465    # the track the user picked starts playing; the rest is shuffled behind it
466    assert queue_items[0] is items[0]
467    assert {item.queue_item_id for item in queue_items[1:]} == {"b", "c", "d"}
468    play_index.assert_awaited_once_with("q1", 0)
469
470
471async def test_replace_with_start_item_pins_chosen_track_under_shuffle() -> None:
472    """REPLACE from a chosen track keeps that track first when shuffle is on."""
473    ctrl = _controller()
474    play_index = AsyncMock()
475    ctrl.play_index = play_index  # type: ignore[method-assign]
476    _shuffled_queue(ctrl)
477    items = _items("q1", ["chosen", "b", "c", "d"])
478
479    await ctrl._enqueue_with_option("q1", items, QueueOption.REPLACE, pin_first=True)
480
481    queue_items = ctrl._queue_data["q1"].items
482    assert queue_items[0] is items[0]
483    assert {item.queue_item_id for item in queue_items[1:]} == {"b", "c", "d"}
484    play_index.assert_awaited_once_with("q1", 0)
485
486
487async def test_play_without_start_item_shuffles_the_whole_batch() -> None:
488    """PLAY of a whole playlist under shuffle still randomises which track starts."""
489    ctrl = _controller()
490    play_index = AsyncMock()
491    ctrl.play_index = play_index  # type: ignore[method-assign]
492    _shuffled_queue(ctrl)
493    items = _items("q1", ["a", "b", "c", "d"])
494
495    await ctrl._enqueue_with_option("q1", items, QueueOption.PLAY)
496
497    # nothing is pinned: the deterministic reversal moves the last item to the front
498    assert ctrl._queue_data["q1"].items[0] is items[-1]
499    play_index.assert_awaited_once_with("q1", 0)
500
501
502async def test_play_with_start_item_keeps_order_when_shuffle_is_off() -> None:
503    """With shuffle off, PLAY from a chosen track plays it and keeps the rest in order."""
504    ctrl = _controller()
505    play_index = AsyncMock()
506    ctrl.play_index = play_index  # type: ignore[method-assign]
507    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
508    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
509    items = _items("q1", ["chosen", "b", "c"])
510
511    await ctrl._enqueue_with_option("q1", items, QueueOption.PLAY, pin_first=True)
512
513    assert [item.queue_item_id for item in ctrl._queue_data["q1"].items] == ["chosen", "b", "c"]
514    play_index.assert_awaited_once_with("q1", 0)
515