/
/
/
1"""
2Tests for ``QueueOption.NEXT`` ("Play next") of a track on a dynamic (managed-pool) queue.
3
4These exercise ``QueueLoaderMixin._handle_play_media`` end-to-end against a bare controller
5instance wired to a real ``ManagedPool``, mirroring ``test_user_initiated_plays`` and
6``test_enqueue_options``. On a dynamic queue, a NEXT track must be carved out of the pool: it
7is inserted literally after the buffered index instead of being folded into the pool as a
8source (which would place it at a random position and subject it to the pool's recency gate).
9A control test proves the linear (non-dynamic) path already does this correctly, scoping the
10carve-out to the dynamic path. Further tests prove ADD and NEXT-of-a-container keep feeding
11the pool, and that the enqueue transitioning a queue to dynamic plays the track exactly once.
12"""
13
14from __future__ import annotations
15
16import random
17from unittest.mock import AsyncMock, MagicMock, Mock
18
19from music_assistant_models.enums import MediaType, PlaybackState, QueueOption
20from music_assistant_models.media_items import Album, ItemMapping, ProviderMapping, Radio, Track
21from music_assistant_models.player_queue import PlayerQueue
22from music_assistant_models.queue_item import QueueItem
23from music_assistant_models.unique_list import UniqueList
24
25from music_assistant.controllers.music.recency import RecencySnapshot, RecencyWindows
26from music_assistant.controllers.player_queues import PlayerQueuesController
27from music_assistant.controllers.player_queues.managed_pool import ManagedPool
28from music_assistant.controllers.player_queues.state import PlayerQueueData
29
30NOW = 1_000_000
31DAY = 86_400
32WINDOWS = RecencyWindows(song_seconds=DAY, artist_seconds=None, duplicate_gap_seconds=3600)
33
34
35def _track(item_id: str, artist: str = "A") -> Track:
36 """Build a playable Track on the 'test' provider."""
37 return Track(
38 item_id=item_id,
39 provider="test",
40 name=f"Track {item_id}",
41 duration=60,
42 artists=UniqueList(
43 [ItemMapping(item_id=artist, provider="test", name=artist, media_type=MediaType.ARTIST)]
44 ),
45 provider_mappings={
46 ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
47 },
48 )
49
50
51def _radio(item_id: str) -> Radio:
52 """Build a dynamic Radio source on the 'test' provider."""
53 return Radio(
54 item_id=item_id,
55 provider="test",
56 name=f"Radio {item_id}",
57 is_dynamic=True,
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 that feeds the pool as a source)."""
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 _queue_item(queue_id: str, track: Track) -> QueueItem:
77 """Build a queue item wrapping the given track."""
78 return QueueItem(
79 queue_id=queue_id,
80 queue_item_id=track.item_id,
81 name=track.name,
82 duration=60,
83 media_item=track,
84 )
85
86
87def _controller(snapshot: RecencySnapshot) -> PlayerQueuesController:
88 """Build a bare controller wired to drive ``_handle_play_media`` with a real ``ManagedPool``."""
89 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
90 ctrl.logger = MagicMock()
91 ctrl.mass = MagicMock()
92 ctrl.mass.music.recency.snapshot = AsyncMock(return_value=snapshot)
93 ctrl.mass.players.get_player = Mock(return_value=Mock(extra_data={}))
94 lock_cm = MagicMock()
95 lock_cm.__aenter__ = AsyncMock(return_value=None)
96 lock_cm.__aexit__ = AsyncMock(return_value=None)
97 ctrl.mass.players.get_player_lock = Mock(return_value=lock_cm)
98 ctrl.signal_update = Mock() # type: ignore[method-assign]
99 ctrl.on_player_update = Mock() # type: ignore[method-assign]
100 ctrl._set_transitioning = Mock() # type: ignore[method-assign]
101 ctrl.get_next_item = Mock(return_value=None) # type: ignore[method-assign]
102 ctrl.recency_windows = Mock(return_value=WINDOWS) # type: ignore[method-assign]
103 ctrl._smart_shuffle = Mock()
104 ctrl._smart_shuffle.is_enabled = Mock(return_value=True)
105 ctrl._smart_shuffle.windows = Mock(return_value=WINDOWS)
106 ctrl._smart_shuffle.arrange = AsyncMock(side_effect=lambda _queue, items: list(items))
107 ctrl._managed_pool = ManagedPool(ctrl)
108 ctrl.play_index = AsyncMock() # type: ignore[method-assign]
109 # a carved-out NEXT track is expanded through the media resolver like on a linear queue;
110 # a bare track resolves to just itself
111 ctrl._media_resolver = Mock()
112 ctrl._media_resolver._resolve_media_items = AsyncMock(
113 side_effect=lambda item, *_args, **_kwargs: [item]
114 )
115 return ctrl
116
117
118def _dynamic_playing_queue(
119 ctrl: PlayerQueuesController, dynamic_candidates: list[Track]
120) -> PlayerQueue:
121 """Set up a playing dynamic queue: current item at index 0, fed by one dynamic radio source."""
122 current = _track("current", artist="Cur")
123 queue = PlayerQueue(
124 queue_id="q1",
125 active=True,
126 display_name="Q1",
127 available=True,
128 items=1,
129 state=PlaybackState.PLAYING,
130 current_index=0,
131 index_in_buffer=0,
132 shuffle_enabled=True,
133 is_dynamic=True,
134 )
135 ctrl._queue_data = {
136 "q1": PlayerQueueData(
137 queue=queue,
138 items=[_queue_item("q1", current)],
139 source_items=[_radio("dyn")],
140 )
141 }
142 ctrl.get = Mock(return_value=queue) # type: ignore[method-assign]
143 ctrl.get_dynamic_source_tracks = AsyncMock(return_value=dynamic_candidates) # type: ignore[method-assign]
144 # a bare track source materializes to just itself
145 ctrl.get_tracks_for_playback = AsyncMock( # type: ignore[method-assign]
146 side_effect=lambda item: [item] if isinstance(item, Track) else []
147 )
148 return queue
149
150
151async def test_play_next_on_dynamic_queue_places_track_next() -> None:
152 """NEXT of a track on a dynamic queue inserts it directly after the buffered index."""
153 random.seed(4)
154 snapshot = RecencySnapshot(now=NOW) # nothing played recently
155 ctrl = _controller(snapshot)
156 dynamic_candidates = [_track(f"d{i}", artist=f"Artist{i}") for i in range(40)]
157 _dynamic_playing_queue(ctrl, dynamic_candidates)
158 wish = _track("wish", artist="Wish")
159
160 await ctrl._handle_play_media("q1", wish, QueueOption.NEXT)
161
162 items = ctrl._queue_data["q1"].items
163 ids = [item.media_item.item_id for item in items if item.media_item is not None]
164 # inserted right after the buffered index; no source changed so the tail stays untouched
165 assert ids == ["current", "wish"], f"expected only ['current', 'wish'], got: {ids}"
166 assert wish not in ctrl._queue_data["q1"].source_items
167
168
169async def test_play_next_on_dynamic_queue_allows_recently_played_track() -> None:
170 """NEXT of a track heard within the recency window still gets enqueued on a dynamic queue."""
171 random.seed(4)
172 replay = _track("replay", artist="Replay")
173 # played 2 hours ago, well within the 1-day song window
174 snapshot = RecencySnapshot(now=NOW, song_ts={("test", "replay"): NOW - 2 * 3600})
175 ctrl = _controller(snapshot)
176 dynamic_candidates = [_track(f"d{i}", artist=f"Artist{i}") for i in range(40)]
177 _dynamic_playing_queue(ctrl, dynamic_candidates)
178
179 await ctrl._handle_play_media("q1", replay, QueueOption.NEXT)
180
181 items = ctrl._queue_data["q1"].items
182 ids = [item.media_item.item_id for item in items if item.media_item is not None]
183 assert ids[1] == "replay", f"expected 'replay' at index 1, got: {ids}"
184
185
186async def test_play_next_on_linear_queue_inserts_after_current() -> None:
187 """Control: NEXT of a track on a non-dynamic queue already inserts it right after current."""
188 random.seed(4)
189 snapshot = RecencySnapshot(now=NOW, song_ts={("test", "wish"): NOW - 2 * 3600})
190 ctrl = _controller(snapshot)
191 tail = [_track(f"t{i}", artist=f"Artist{i}") for i in range(5)]
192 current = _track("current", artist="Cur")
193 queue = PlayerQueue(
194 queue_id="q1",
195 active=True,
196 display_name="Q1",
197 available=True,
198 items=6,
199 state=PlaybackState.PLAYING,
200 current_index=0,
201 index_in_buffer=0,
202 shuffle_enabled=True,
203 is_dynamic=False,
204 )
205 ctrl._queue_data = {
206 "q1": PlayerQueueData(
207 queue=queue,
208 items=[_queue_item("q1", current)] + [_queue_item("q1", t) for t in tail],
209 source_items=[],
210 )
211 }
212 ctrl.get = Mock(return_value=queue) # type: ignore[method-assign]
213 wish = _track("wish", artist="Wish")
214 ctrl._media_resolver = Mock()
215 ctrl._media_resolver._resolve_media_items = AsyncMock(return_value=[wish])
216
217 await ctrl._handle_play_media("q1", wish, QueueOption.NEXT)
218
219 items = ctrl._queue_data["q1"].items
220 ids = [item.media_item.item_id for item in items if item.media_item is not None]
221 assert ids[0] == "current"
222 assert ids[1] == "wish", f"expected 'wish' at index 1: {ids}"
223
224
225async def test_add_track_on_dynamic_queue_still_feeds_pool() -> None:
226 """ADD of a track on a dynamic queue still feeds the pool as a source (unlike NEXT)."""
227 random.seed(4)
228 snapshot = RecencySnapshot(now=NOW)
229 ctrl = _controller(snapshot)
230 dynamic_candidates = [_track(f"d{i}", artist=f"Artist{i}") for i in range(40)]
231 _dynamic_playing_queue(ctrl, dynamic_candidates)
232 seed = _track("seed", artist="Seed")
233
234 await ctrl._handle_play_media("q1", seed, QueueOption.ADD)
235
236 items = ctrl._queue_data["q1"].items
237 ids = [item.media_item.item_id for item in items if item.media_item is not None]
238 # fed to the pool and mixed into the rebuilt tail, not inserted at index 1 like NEXT
239 # (a one-shot track source is retired from source_items right after dispatch)
240 assert "seed" in ids, f"expected 'seed' mixed into the pool tail: {ids}"
241 assert ids[1] != "seed", f"expected 'seed' mixed into the pool tail, not at index 1: {ids}"
242 # the pool actually ran (proving _enter_dynamic_mode fired), not left untouched
243 assert any(item_id.startswith("d") for item_id in ids), f"pool tail was not rebuilt: {ids}"
244
245
246async def test_play_next_container_on_dynamic_queue_feeds_pool() -> None:
247 """NEXT of a container on a dynamic queue still feeds the pool as a source (unlike a track)."""
248 random.seed(4)
249 snapshot = RecencySnapshot(now=NOW)
250 ctrl = _controller(snapshot)
251 dynamic_candidates = [_track(f"d{i}", artist=f"Artist{i}") for i in range(40)]
252 _dynamic_playing_queue(ctrl, dynamic_candidates)
253 album = _album("alb")
254 album_tracks = [_track(f"a{i}", artist=f"AlbArtist{i}") for i in range(30)]
255 ctrl.get_tracks_for_playback = AsyncMock( # type: ignore[method-assign]
256 side_effect=lambda item: album_tracks if item is album else []
257 )
258
259 await ctrl._handle_play_media("q1", album, QueueOption.NEXT)
260
261 ids = [
262 item.media_item.item_id
263 for item in ctrl._queue_data["q1"].items
264 if item.media_item is not None
265 ]
266 # the album's tracks are mixed into the rebuilt tail and the album stays a pool source
267 assert any(item_id.startswith("a") for item_id in ids), f"album did not feed the pool: {ids}"
268 source_ids = {item.item_id for item in ctrl._queue_data["q1"].source_items}
269 assert "alb" in source_ids, f"album missing from the pool sources: {source_ids}"
270 assert ctrl._queue_data["q1"].queue.is_dynamic
271
272
273async def test_play_next_mixed_batch_transition_plays_track_once() -> None:
274 """NEXT of [track, dynamic radio] on a linear queue inserts the track next, exactly once."""
275 random.seed(4)
276 snapshot = RecencySnapshot(now=NOW)
277 ctrl = _controller(snapshot)
278 current = _track("current", artist="Cur")
279 queue = PlayerQueue(
280 queue_id="q1",
281 active=True,
282 display_name="Q1",
283 available=True,
284 items=1,
285 state=PlaybackState.PLAYING,
286 current_index=0,
287 index_in_buffer=0,
288 shuffle_enabled=False,
289 is_dynamic=False,
290 )
291 ctrl._queue_data = {
292 "q1": PlayerQueueData(queue=queue, items=[_queue_item("q1", current)], source_items=[])
293 }
294 ctrl.get = Mock(return_value=queue) # type: ignore[method-assign]
295 dynamic_candidates = [_track(f"d{i}", artist=f"Artist{i}") for i in range(10)]
296 ctrl.get_dynamic_source_tracks = AsyncMock(return_value=dynamic_candidates) # type: ignore[method-assign]
297 ctrl.get_tracks_for_playback = AsyncMock( # type: ignore[method-assign]
298 side_effect=lambda item: [item] if isinstance(item, Track) else []
299 )
300 wish = _track("wish", artist="Wish")
301
302 await ctrl._handle_play_media("q1", [wish, _radio("dyn")], QueueOption.NEXT)
303
304 ids = [
305 item.media_item.item_id
306 for item in ctrl._queue_data["q1"].items
307 if item.media_item is not None
308 ]
309 # the radio feeds the new pool; the play-next track is inserted next, exactly once
310 assert ids.count("wish") == 1, f"'wish' must appear exactly once: {ids}"
311 assert ids[1] == "wish", f"expected 'wish' at index 1 (play next), got: {ids}"
312 assert not any(item.item_id == "wish" for item in ctrl._queue_data["q1"].source_items), (
313 "play-next track must not be recorded as a pool source"
314 )
315 assert ctrl._queue_data["q1"].queue.is_dynamic
316
317
318async def test_play_next_mixed_container_transition_no_duplicates() -> None:
319 """NEXT of [album, dynamic radio] on a linear queue pools the album without duplicating it."""
320 random.seed(4)
321 snapshot = RecencySnapshot(now=NOW)
322 ctrl = _controller(snapshot)
323 current = _track("current", artist="Cur")
324 queue = PlayerQueue(
325 queue_id="q1",
326 active=True,
327 display_name="Q1",
328 available=True,
329 items=1,
330 state=PlaybackState.PLAYING,
331 current_index=0,
332 index_in_buffer=0,
333 shuffle_enabled=False,
334 is_dynamic=False,
335 )
336 ctrl._queue_data = {
337 "q1": PlayerQueueData(queue=queue, items=[_queue_item("q1", current)], source_items=[])
338 }
339 ctrl.get = Mock(return_value=queue) # type: ignore[method-assign]
340 album = _album("alb")
341 album_tracks = [_track(f"a{i}", artist=f"AlbArtist{i}") for i in range(5)]
342 ctrl._media_resolver._resolve_media_items = AsyncMock( # type: ignore[method-assign]
343 return_value=list(album_tracks)
344 )
345 ctrl.get_dynamic_source_tracks = AsyncMock( # type: ignore[method-assign]
346 return_value=[_track(f"d{i}", artist=f"Artist{i}") for i in range(10)]
347 )
348 ctrl.get_tracks_for_playback = AsyncMock( # type: ignore[method-assign]
349 side_effect=lambda item: list(album_tracks) if item is album else []
350 )
351
352 await ctrl._handle_play_media("q1", [album, _radio("dyn")], QueueOption.NEXT)
353
354 ids = [
355 item.media_item.item_id
356 for item in ctrl._queue_data["q1"].items
357 if item.media_item is not None
358 ]
359 # the album feeds the new pool as a source; its expansion must not also be inserted
360 assert any(item_id.startswith("a") for item_id in ids), f"album did not feed the pool: {ids}"
361 assert len(ids) == len(set(ids)), f"duplicate items in queue: {ids}"
362