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