/
/
1"""
2Tests for the shuffle state a newly started media item ends up with.
3
4A shuffle left switched on by an earlier listening session must not silently reorder the album the
5user just picked, while a shuffle the user switched on moments before pressing play is a deliberate
6"shuffle this" gesture and has to be honoured. Only the options that start the media right away
7(play and replace) begin such a new listening session; the options that stage items onto the queue
8keep its shuffle state. These drive the real ``play_media`` path against a bare controller instance,
9mirroring ``test_user_initiated_plays`` and ``test_enqueue_options``: resolution and playback are
10stubbed, but the enqueue/load path runs for real so the resulting item order is verified
11end-to-end.
12"""
13
14from __future__ import annotations
15
16import time
17from typing import Any, cast
18from unittest.mock import AsyncMock, MagicMock, Mock
19
20import pytest
21from music_assistant_models.enums import MediaType, QueueOption
22from music_assistant_models.media_items import (
23 Album,
24 ItemMapping,
25 Playlist,
26 ProviderMapping,
27 Track,
28)
29from music_assistant_models.player_queue import PlayerQueue
30from music_assistant_models.queue_item import QueueItem
31from music_assistant_models.unique_list import UniqueList
32
33from music_assistant.controllers.player_queues import PlayerQueuesController
34from music_assistant.controllers.player_queues.constants import SHUFFLE_INTENT_WINDOW
35from music_assistant.controllers.player_queues.state import PlayerQueueData
36
37# the album the user starts, in its own track order
38ALBUM_TRACKS = ["t1", "t2", "t3", "t4"]
39
40# the options that start the media right away, i.e. begin a new listening session
41START_OPTIONS = [QueueOption.PLAY, QueueOption.REPLACE]
42
43# the options that stage media onto the queue instead of starting it
44STAGE_OPTIONS = [QueueOption.ADD, QueueOption.NEXT, QueueOption.REPLACE_NEXT]
45
46# a queue as an earlier shuffle left it: the list order deliberately disagrees with the sort order
47SHUFFLED_QUEUE = [("e1", 0), ("e4", 3), ("e2", 1), ("e5", 4), ("e3", 2)]
48
49
50def _track(item_id: str) -> Track:
51 """Build a playable Track on the 'test' provider."""
52 return Track(
53 item_id=item_id,
54 provider="test",
55 name=f"Track {item_id}",
56 duration=60,
57 artists=UniqueList(
58 [ItemMapping(item_id="a", provider="test", name="A", media_type=MediaType.ARTIST)]
59 ),
60 provider_mappings={
61 ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
62 },
63 )
64
65
66def _album() -> Album:
67 """Build the Album the user presses play on (its configured enqueue default is 'replace')."""
68 return Album(
69 item_id="al1",
70 provider="test",
71 name="Album al1",
72 provider_mappings={
73 ProviderMapping(item_id="al1", provider_domain="test", provider_instance="test")
74 },
75 )
76
77
78def _dynamic_playlist() -> Playlist:
79 """Build a dynamic playlist: a source that supplies its own tracks and is always a smart mix."""
80 playlist = Playlist(
81 item_id="dyn1",
82 provider="test",
83 name="Dynamic",
84 provider_mappings={
85 ProviderMapping(item_id="dyn1", provider_domain="test", provider_instance="test")
86 },
87 )
88 playlist.is_dynamic = True
89 return playlist
90
91
92def _controller(**queue_kwargs: Any) -> Any:
93 """
94 Build a bare controller driving ``play_media`` on a single queue "q1".
95
96 The album's tracks come from a stubbed media resolver and the shuffle is made deterministic
97 (it reverses the batch), so the resulting queue order tells shuffle-on from shuffle-off.
98
99 :param queue_kwargs: Overrides for the queue this controller is set up with.
100 """
101 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
102 ctrl.logger = Mock()
103 ctrl.mass = MagicMock()
104 ctrl.mass.players.get_player = Mock(return_value=Mock(extra_data={}))
105 lock_cm = MagicMock()
106 lock_cm.__aenter__ = AsyncMock(return_value=None)
107 lock_cm.__aexit__ = AsyncMock(return_value=None)
108 ctrl.mass.players.get_player_lock = Mock(return_value=lock_cm)
109 ctrl.signal_update = Mock() # type: ignore[method-assign]
110 ctrl.on_player_update = Mock() # type: ignore[method-assign]
111 ctrl.play_index = AsyncMock() # type: ignore[method-assign]
112 ctrl.get_next_item = Mock(return_value=None) # type: ignore[method-assign]
113 ctrl.get_config_value = Mock(return_value=QueueOption.REPLACE.value) # type: ignore[method-assign]
114 ctrl._managed_pool = Mock()
115 ctrl._managed_pool.fill = AsyncMock(
116 side_effect=lambda *_args, **_kwargs: [_track(item_id) for item_id in ALBUM_TRACKS]
117 )
118 ctrl._smart_shuffle = Mock()
119 ctrl._smart_shuffle.is_enabled = Mock(return_value=True)
120 ctrl._smart_shuffle.arrange = AsyncMock(side_effect=lambda _queue, items: list(items)[::-1])
121 ctrl._media_resolver = Mock()
122 ctrl._media_resolver._resolve_media_items = AsyncMock(
123 side_effect=lambda *_args, **_kwargs: [_track(item_id) for item_id in ALBUM_TRACKS]
124 )
125 queue = PlayerQueue(
126 queue_id="q1", active=True, display_name="Q1", available=True, items=0, **queue_kwargs
127 )
128 ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
129 return ctrl
130
131
132def _queue(ctrl: Any) -> PlayerQueue:
133 """Return the controller's queue."""
134 return cast("PlayerQueue", ctrl._queue_data["q1"].queue)
135
136
137def _played_order(ctrl: Any) -> list[str]:
138 """Return the item ids of the tracks currently loaded in the queue, in play order."""
139 return [
140 item.media_item.item_id
141 for item in ctrl._queue_data["q1"].items
142 if item.media_item is not None
143 ]
144
145
146def _load_shuffled_queue(ctrl: Any) -> None:
147 """Fill the queue with items whose list order disagrees with their original sort order."""
148 items = []
149 for item_id, sort_index in SHUFFLED_QUEUE:
150 item = QueueItem.from_media_item("q1", _track(item_id))
151 item.sort_index = sort_index
152 items.append(item)
153 ctrl._queue_data["q1"].items = items
154 _queue(ctrl).items = len(items)
155
156
157@pytest.mark.parametrize("option", START_OPTIONS)
158async def test_shuffle_left_on_by_previous_session_is_reset(option: QueueOption) -> None:
159 """An album started on a queue that still had shuffle on plays in its own track order."""
160 ctrl = _controller(shuffle_enabled=True, smart_shuffle_active=True)
161
162 await ctrl.play_media("q1", _album(), option)
163
164 assert _queue(ctrl).shuffle_enabled is False
165 assert _queue(ctrl).smart_shuffle_active is False
166 # the album is played front to back, not in the (reversed) shuffle order
167 assert _played_order(ctrl) == ALBUM_TRACKS
168 # shuffle was settled before the items were resolved: a shuffled queue asks the resolver to
169 # keep the items preceding a chosen track, an in-order one does not
170 resolve_call = ctrl._media_resolver._resolve_media_items.call_args
171 assert resolve_call.kwargs["keep_preceding_items"] is False
172
173
174async def test_shuffle_left_on_is_reset_for_a_derived_enqueue_option() -> None:
175 """The reset also applies when the enqueue option comes from the media type's config default."""
176 ctrl = _controller(shuffle_enabled=True)
177
178 await ctrl.play_media("q1", _album())
179
180 # the option was derived from the album's configured default (which is 'replace')
181 assert ctrl.get_config_value.call_args.args[0] == "default_enqueue_option_album"
182 assert _queue(ctrl).shuffle_enabled is False
183 assert _played_order(ctrl) == ALBUM_TRACKS
184
185
186async def test_the_reset_restores_the_order_of_the_items_that_stay_in_the_queue() -> None:
187 """
188 Media played onto a shuffled queue puts the items it keeps back in their original order.
189
190 Unlike a replace, a play keeps the tail behind the current item, so the reset has to un-shuffle
191 that tail as well: items left in shuffled order behind a queue that now reads "shuffle off"
192 would contradict its own flag.
193 """
194 ctrl = _controller(shuffle_enabled=True, current_index=0)
195 _load_shuffled_queue(ctrl)
196
197 await ctrl.play_media("q1", _album(), QueueOption.PLAY)
198
199 assert _queue(ctrl).shuffle_enabled is False
200 # the item being played, then the album, then the kept tail back in its original order
201 assert _played_order(ctrl) == ["e1", *ALBUM_TRACKS, "e2", "e3", "e4", "e5"]
202
203
204async def test_shuffle_just_switched_on_by_the_user_is_honoured() -> None:
205 """Switching shuffle on and then pressing play shuffles the album the user picked."""
206 ctrl = _controller()
207 await ctrl.set_shuffle("q1", True)
208
209 await ctrl.play_media("q1", _album(), QueueOption.REPLACE)
210
211 assert _queue(ctrl).shuffle_enabled is True
212 assert _played_order(ctrl) == ALBUM_TRACKS[::-1]
213
214
215@pytest.mark.parametrize("option", START_OPTIONS)
216async def test_shuffle_intent_is_only_good_for_one_play(option: QueueOption) -> None:
217 """The toggle carries into the album the user starts next, but not into the one after it."""
218 ctrl = _controller()
219 await ctrl.set_shuffle("q1", True)
220 await ctrl.play_media("q1", _album(), option)
221
222 await ctrl.play_media("q1", _album(), option)
223
224 assert ctrl._queue_data["q1"].shuffle_set_at is None
225 assert _queue(ctrl).shuffle_enabled is False
226 assert _played_order(ctrl) == ALBUM_TRACKS
227
228
229async def test_switching_shuffle_off_drops_the_intent() -> None:
230 """Turning shuffle back off leaves no intent behind for the next play to pick up."""
231 ctrl = _controller()
232 await ctrl.set_shuffle("q1", True)
233 await ctrl.set_shuffle("q1", False)
234
235 await ctrl.play_media("q1", _album(), QueueOption.REPLACE)
236
237 assert ctrl._queue_data["q1"].shuffle_set_at is None
238 assert _queue(ctrl).shuffle_enabled is False
239 assert _played_order(ctrl) == ALBUM_TRACKS
240
241
242@pytest.mark.parametrize("option", START_OPTIONS)
243async def test_stale_shuffle_intent_is_not_honoured(option: QueueOption) -> None:
244 """A toggle from well before the play command is a leftover, not intent for this album."""
245 ctrl = _controller(shuffle_enabled=True)
246 ctrl._queue_data["q1"].shuffle_set_at = time.monotonic() - SHUFFLE_INTENT_WINDOW - 100
247
248 await ctrl.play_media("q1", _album(), option)
249
250 assert _queue(ctrl).shuffle_enabled is False
251 assert _played_order(ctrl) == ALBUM_TRACKS
252
253
254async def test_explicit_shuffle_request_shuffles_an_unshuffled_queue() -> None:
255 """A caller asking for a shuffled play gets one, without having to toggle shuffle first."""
256 ctrl = _controller()
257
258 await ctrl.play_media("q1", _album(), QueueOption.REPLACE, shuffle=True)
259
260 assert _queue(ctrl).shuffle_enabled is True
261 assert _played_order(ctrl) == ALBUM_TRACKS[::-1]
262
263
264@pytest.mark.parametrize("option", START_OPTIONS)
265async def test_explicit_no_shuffle_wins_over_a_fresh_intent(option: QueueOption) -> None:
266 """An explicit "play in order" beats the shuffle the user switched on moments earlier."""
267 ctrl = _controller()
268 await ctrl.set_shuffle("q1", True)
269
270 await ctrl.play_media("q1", _album(), option, shuffle=False)
271
272 assert _queue(ctrl).shuffle_enabled is False
273 assert _played_order(ctrl) == ALBUM_TRACKS
274
275
276async def test_dynamic_source_overrides_an_explicit_play_in_order() -> None:
277 """A dynamic source is an always-on smart mix, so it outranks an explicit "play in order"."""
278 ctrl = _controller()
279
280 await ctrl.play_media("q1", _dynamic_playlist(), QueueOption.REPLACE, shuffle=False)
281
282 assert _queue(ctrl).is_dynamic is True
283 assert _queue(ctrl).shuffle_enabled is True
284 assert _queue(ctrl).smart_shuffle_active is True
285
286
287async def test_replacing_a_dynamic_queue_drops_the_smart_mix_indicator() -> None:
288 """
289 An album started over a dynamic queue is a plain queue, so it must not report a smart mix.
290
291 The dynamic source is what made smart shuffle active here (the per-queue setting is off), so
292 dropping it has to take the indicator with it.
293 """
294 ctrl = _controller(shuffle_enabled=True, smart_shuffle_active=True, is_dynamic=True)
295 ctrl._smart_shuffle.is_enabled = Mock(return_value=False)
296
297 await ctrl.play_media("q1", _album(), QueueOption.REPLACE, shuffle=True)
298
299 assert _queue(ctrl).is_dynamic is False
300 # the caller asked for a shuffled play, but a plain random shuffle is not a smart mix
301 assert _queue(ctrl).shuffle_enabled is True
302 assert _queue(ctrl).smart_shuffle_active is False
303
304
305async def test_playing_over_a_dynamic_queue_honours_an_explicit_play_in_order() -> None:
306 """
307 An album played over a dynamic queue takes over from it, so it plays in the order asked for.
308
309 Play does not clear the queue up front, so the queue is still dynamic when the shuffle state is
310 settled - and a dynamic queue's toggle is locked, so the request cannot be routed through
311 set_shuffle. The requested state still has to reach the items, which are resolved against it.
312 """
313 ctrl = _controller(shuffle_enabled=True, smart_shuffle_active=True, is_dynamic=True)
314
315 await ctrl.play_media("q1", _album(), QueueOption.PLAY, shuffle=False)
316
317 # the album took over as the queue's only source, so the queue is a plain one again
318 assert _queue(ctrl).is_dynamic is False
319 assert _queue(ctrl).shuffle_enabled is False
320 assert _queue(ctrl).smart_shuffle_active is False
321 # the flag alone proves little here: the album itself has to come out in its own order
322 assert _played_order(ctrl) == ALBUM_TRACKS
323
324
325async def test_playing_media_on_an_ended_queue_resets_shuffle() -> None:
326 """A finished queue is started over by a play, so the shuffle goes with the old session."""
327 ctrl = _controller(shuffle_enabled=True, ended=True)
328
329 await ctrl.play_media("q1", _album(), QueueOption.PLAY)
330
331 assert _queue(ctrl).shuffle_enabled is False
332 assert _played_order(ctrl) == ALBUM_TRACKS
333
334
335@pytest.mark.parametrize("option", STAGE_OPTIONS)
336async def test_staging_media_onto_an_ended_queue_keeps_shuffle(option: QueueOption) -> None:
337 """
338 Staging media onto a finished queue keeps its shuffle, even though the queue restarts.
339
340 This is deliberate, not an oversight: the shuffle is only reset for the options the user reaches
341 for to start something now. Whether the queue happens to have played to its end does not change
342 what "queue this up" means, so it must not change the shuffle state either.
343 """
344 ctrl = _controller(shuffle_enabled=True, ended=True)
345 # a marker rather than a realistic timestamp: it only has to show the intent was left alone
346 ctrl._queue_data["q1"].shuffle_set_at = 12345.0
347
348 await ctrl.play_media("q1", _album(), option)
349
350 assert _queue(ctrl).shuffle_enabled is True
351 assert ctrl._queue_data["q1"].shuffle_set_at == 12345.0
352 # the batch landed in shuffle order instead of the album's own track order
353 assert sorted(_played_order(ctrl)) == sorted(ALBUM_TRACKS)
354 assert _played_order(ctrl) != ALBUM_TRACKS
355
356
357@pytest.mark.parametrize("shuffle_enabled", [True, False])
358@pytest.mark.parametrize("option", STAGE_OPTIONS)
359async def test_enqueueing_leaves_shuffle_untouched(
360 option: QueueOption, shuffle_enabled: bool
361) -> None:
362 """
363 Staging media onto the queue is not a new listening session, so it keeps its shuffle.
364
365 These all keep (part of) the existing queue, whose items are already in shuffled order, so
366 switching shuffle off here would leave those items contradicting the queue's own flag.
367 """
368 # a running queue, as opposed to the finished one the staging options start over from
369 ctrl = _controller(shuffle_enabled=shuffle_enabled, ended=False)
370 # a marker rather than a realistic timestamp: it only has to show the intent was left alone
371 ctrl._queue_data["q1"].shuffle_set_at = 12345.0
372
373 await ctrl.play_media("q1", _album(), option)
374
375 assert _queue(ctrl).shuffle_enabled is shuffle_enabled
376 # the intent is not consumed either: it still belongs to the next media the user starts
377 assert ctrl._queue_data["q1"].shuffle_set_at == 12345.0
378
379
380def test_clear_command_resets_shuffle() -> None:
381 """Clearing the queue is an explicit "start over", so the shuffle goes with the content."""
382 ctrl = _controller(shuffle_enabled=True, smart_shuffle_active=True)
383 ctrl._queue_data["q1"].items = [
384 QueueItem.from_media_item("q1", _track(item_id)) for item_id in ALBUM_TRACKS
385 ]
386
387 ctrl.clear("q1")
388
389 assert _queue(ctrl).shuffle_enabled is False
390 assert _queue(ctrl).smart_shuffle_active is False
391 assert ctrl._queue_data["q1"].shuffle_set_at is None
392
393
394def test_empty_queue_reaching_its_end_keeps_shuffle() -> None:
395 """An end reached with nothing to replay is still not the user clearing the queue."""
396 ctrl = _controller(shuffle_enabled=True)
397
398 ctrl.mark_ended("q1")
399
400 assert ctrl._queue_data["q1"].items == []
401 assert _queue(ctrl).shuffle_enabled is True
402