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