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