/
/
/
1"""
2Tests for what a queue looks like once it played to its end.
3
4A queue that ran out of items keeps them and parks its playback position on the last one, flagged as
5ended, so clients can say it finished and pressing play starts it over. Everything with a natural end
6settles that same way; only a live source, which has no end, is left exactly as it was.
7"""
8
9from __future__ import annotations
10
11import time
12from types import SimpleNamespace
13from typing import TYPE_CHECKING, Any, cast
14from unittest.mock import AsyncMock, MagicMock, Mock
15
16import pytest
17from music_assistant_models.enums import MediaType, PlaybackState, QueueOption
18from music_assistant_models.errors import PlayerUnavailableError
19from music_assistant_models.player_queue import PlayerQueue
20from music_assistant_models.queue_item import QueueItem
21
22from music_assistant.controllers.player_queues import PlayerQueuesController
23from music_assistant.controllers.player_queues.playback_tracker import PlaybackTrackerMixin
24from music_assistant.controllers.player_queues.state import PlayerQueueData
25
26if TYPE_CHECKING:
27 from music_assistant.controllers.player_queues.helpers import CompareState
28
29QUEUE_ID = "q1"
30
31
32def _controller(*, player_available: bool = True) -> tuple[PlayerQueuesController, PlayerQueue]:
33 """
34 Build a bare controller holding a queue that just finished its last item.
35
36 :param player_available: Whether the queue's player can be resolved, so a caller can exercise
37 a playback start that fails before anything is loaded.
38 """
39 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
40 queue = PlayerQueue(queue_id=QUEUE_ID, active=True, display_name="Q1", available=True, items=0)
41 items = [
42 QueueItem(queue_id=QUEUE_ID, queue_item_id="first", name="first", duration=100),
43 QueueItem(queue_id=QUEUE_ID, queue_item_id="last", name="last", duration=100),
44 ]
45 queue.items = len(items)
46 queue.state = PlaybackState.IDLE
47 queue.current_index = 1
48 queue.current_item = items[1]
49 queue.elapsed_time = 99.0
50 queue.elapsed_time_last_updated = time.time() - 900
51 queue.resume_pos = 99
52 queue.index_in_buffer = 1
53 queue_data = PlayerQueueData(queue=queue)
54 queue_data.items = items
55 ctrl._queue_data = {QUEUE_ID: queue_data}
56 ctrl.signal_update = Mock() # type: ignore[method-assign]
57 ctrl.on_player_update = Mock() # type: ignore[method-assign]
58 ctrl._check_player_permission = Mock() # type: ignore[method-assign]
59 ctrl.play_index = AsyncMock() # type: ignore[method-assign]
60 ctrl._managed_pool = MagicMock()
61 ctrl.mass = MagicMock()
62 ctrl.mass.create_task = Mock(side_effect=lambda coro, **_kw: coro.close())
63 ctrl.mass.players.play_media = AsyncMock()
64 ctrl.mass.music.get_playback_speed = AsyncMock(return_value=1.0)
65 ctrl.mass.players.get_player = Mock(
66 return_value=MagicMock(state=MagicMock(playback_state=PlaybackState.IDLE))
67 if player_available
68 else None
69 )
70 ctrl.logger = MagicMock()
71 return ctrl, queue
72
73
74def test_mark_ended_keeps_the_items_and_parks_on_the_last_one() -> None:
75 """Ending a queue leaves it replayable, with the position on the item that finished."""
76 ctrl, queue = _controller()
77
78 ctrl.mark_ended(QUEUE_ID)
79
80 assert queue.ended is True
81 # the items survive so the queue can be replayed, and there is still evidence it played
82 assert queue.items == 2
83 assert [x.queue_item_id for x in ctrl._queue_data[QUEUE_ID].items] == ["first", "last"]
84 # parked on the last item: a null index reads as "never started", one past the end is
85 # silently misread by everything that does arithmetic on it
86 assert queue.current_index == 1
87 assert queue.current_item is not None
88 assert queue.current_item.queue_item_id == "last"
89 assert queue.next_item is None
90 assert queue.elapsed_time == 0
91 assert queue.resume_pos == 0
92 assert queue.index_in_buffer is None
93
94
95def test_mark_ended_on_an_itemless_queue_clears_it() -> None:
96 """With nothing to replay there is nothing to advertise as finished either."""
97 ctrl, queue = _controller()
98 ctrl._queue_data[QUEUE_ID].items = []
99 queue.items = 0
100
101 ctrl.mark_ended(QUEUE_ID)
102
103 assert queue.ended is False
104 assert queue.items == 0
105 assert queue.current_index is None
106
107
108def test_clear_still_empties_the_queue() -> None:
109 """An explicit clear keeps emptying the queue, and an emptied queue is not "ended"."""
110 ctrl, queue = _controller()
111
112 ctrl.clear(QUEUE_ID)
113
114 assert queue.items == 0
115 assert ctrl._queue_data[QUEUE_ID].items == []
116 assert queue.current_index is None
117 assert queue.current_item is None
118 assert queue.ended is False
119
120
121async def test_play_on_an_ended_queue_restarts_from_the_beginning() -> None:
122 """Pressing play on a queue that reached its end replays it from the first item."""
123 ctrl, _queue = _controller()
124 ctrl.mark_ended(QUEUE_ID)
125
126 await ctrl.resume(QUEUE_ID)
127
128 ctrl.play_index.assert_awaited_once() # type: ignore[attr-defined]
129 queue_id, item_id, seek_pos = ctrl.play_index.await_args.args[:3] # type: ignore[attr-defined]
130 assert queue_id == QUEUE_ID
131 assert item_id == "first"
132 assert seek_pos == 0
133
134
135def _stub_play_index_deps(ctrl: PlayerQueuesController) -> None:
136 """Stub out what the real play_index reaches for, so it can run on the bare controller."""
137 ctrl._set_transitioning = Mock() # type: ignore[method-assign]
138 ctrl._load_item = AsyncMock() # type: ignore[method-assign]
139 ctrl._get_next_index = Mock(return_value=None) # type: ignore[method-assign]
140 ctrl.player_media_from_queue_item = AsyncMock() # type: ignore[method-assign]
141
142
143def test_reordering_a_finished_queue_leaves_it_finished() -> None:
144 """A reorder (or a shuffle toggle, which is one) must not silently un-finish the queue."""
145 ctrl, queue = _controller()
146 ctrl.mark_ended(QUEUE_ID)
147
148 ctrl.update_items(QUEUE_ID, [*reversed(ctrl._queue_data[QUEUE_ID].items)])
149
150 # still finished, so a play press still restarts it instead of replaying its last item
151 assert queue.ended is True
152
153
154async def test_starting_playback_ends_the_ended_state() -> None:
155 """Playing anything means the queue is running again, not sitting at its end."""
156 ctrl, queue = _controller()
157 ctrl.mark_ended(QUEUE_ID)
158 _stub_play_index_deps(ctrl)
159
160 # the helper mocks play_index away for the resume test; call the real one here
161 await PlayerQueuesController.play_index(ctrl, QUEUE_ID, 0)
162
163 assert queue.ended is False
164
165
166async def test_a_failed_start_leaves_the_queue_finished() -> None:
167 """If playback never gets going, the queue stays finished instead of losing its position."""
168 ctrl, queue = _controller(player_available=False)
169 ctrl.mark_ended(QUEUE_ID)
170 _stub_play_index_deps(ctrl)
171
172 with pytest.raises(PlayerUnavailableError):
173 await PlayerQueuesController.play_index(ctrl, QUEUE_ID, 0)
174
175 # still finished, so the queue can still be started over rather than replaying its last item
176 assert queue.ended is True
177
178
179async def test_restarting_a_finished_queue_ignores_a_left_over_resume_point() -> None:
180 """A restart must start over, not seek an audiobook back to where it was left off."""
181 ctrl, _queue = _controller()
182 first_item = ctrl._queue_data[QUEUE_ID].items[0]
183 first_item.media_item = SimpleNamespace( # type: ignore[assignment]
184 resume_position_ms=3_600_000, media_type=MediaType.AUDIOBOOK
185 )
186 ctrl.mark_ended(QUEUE_ID)
187 _stub_play_index_deps(ctrl)
188
189 await PlayerQueuesController.play_index(ctrl, QUEUE_ID, 0)
190
191 assert ctrl._load_item.await_args is not None # type: ignore[attr-defined]
192 assert ctrl._load_item.await_args.kwargs["seek_position"] == 0 # type: ignore[attr-defined]
193
194
195async def test_playing_an_unfinished_item_still_honours_its_resume_point() -> None:
196 """The resume point still applies on a queue that did not play to its end."""
197 ctrl, _queue = _controller()
198 first_item = ctrl._queue_data[QUEUE_ID].items[0]
199 first_item.media_item = SimpleNamespace( # type: ignore[assignment]
200 resume_position_ms=3_600_000, media_type=MediaType.AUDIOBOOK
201 )
202 _stub_play_index_deps(ctrl)
203
204 await PlayerQueuesController.play_index(ctrl, QUEUE_ID, 0)
205
206 assert ctrl._load_item.await_args is not None # type: ignore[attr-defined]
207 assert ctrl._load_item.await_args.kwargs["seek_position"] == 3599 # type: ignore[attr-defined]
208
209
210# --- enqueueing onto a finished queue ---
211
212
213def _enqueue_controller() -> tuple[PlayerQueuesController, PlayerQueue, dict[str, Any]]:
214 """Build a finished queue whose `load` splices items the way the real one does."""
215 ctrl, queue = _controller()
216 ctrl.mark_ended(QUEUE_ID)
217 calls: dict[str, Any] = {}
218
219 async def _fake_load(
220 queue_id: str,
221 queue_items: list[QueueItem],
222 insert_at_index: int = 0,
223 keep_remaining: bool = True,
224 keep_played: bool = True,
225 shuffle: bool = False, # noqa: ARG001
226 ) -> None:
227 current = ctrl._queue_data[queue_id].items
228 prev_items = current[:insert_at_index] if keep_played else []
229 next_items = current[insert_at_index:] if keep_remaining else []
230 ctrl.update_items(queue_id, [*prev_items, *queue_items, *next_items])
231
232 ctrl.load = AsyncMock(side_effect=_fake_load) # type: ignore[method-assign]
233 ctrl.play_index = AsyncMock( # type: ignore[method-assign]
234 side_effect=lambda _qid, index, *_a, **_kw: calls.__setitem__("played", index)
235 )
236 ctrl._load_pinned_first = AsyncMock() # type: ignore[method-assign]
237 return ctrl, queue, calls
238
239
240def _new_item() -> list[QueueItem]:
241 """Build the batch being enqueued."""
242 return [QueueItem(queue_id=QUEUE_ID, queue_item_id="new", name="new", duration=100)]
243
244
245async def test_playing_something_new_replaces_a_finished_queue() -> None:
246 """A finished queue is done, so a play/replace request starts a fresh one."""
247 for option in (QueueOption.PLAY, QueueOption.REPLACE, QueueOption.REPLACE_NEXT):
248 ctrl, queue, _calls = _enqueue_controller()
249
250 await ctrl._enqueue_with_option(QUEUE_ID, _new_item(), option)
251
252 assert [x.queue_item_id for x in ctrl._queue_data[QUEUE_ID].items] == ["new"], option
253 assert queue.ended is False, option
254
255
256async def test_queueing_something_next_replaces_a_finished_queue() -> None:
257 """NEXT onto a finished queue starts fresh too, and stages without playing."""
258 ctrl, queue, calls = _enqueue_controller()
259
260 await ctrl._enqueue_with_option(QUEUE_ID, _new_item(), QueueOption.NEXT)
261
262 assert [x.queue_item_id for x in ctrl._queue_data[QUEUE_ID].items] == ["new"]
263 assert queue.current_index == 0
264 assert queue.ended is False
265 assert "played" not in calls
266
267
268async def test_adding_to_a_finished_queue_keeps_it_and_continues_from_the_end() -> None:
269 """An explicit ADD keeps what played and points the queue at what was just added."""
270 ctrl, queue, calls = _enqueue_controller()
271
272 await ctrl._enqueue_with_option(QUEUE_ID, _new_item(), QueueOption.ADD)
273
274 # the tracks that played are kept, so the queue still shows what it played
275 assert [x.queue_item_id for x in ctrl._queue_data[QUEUE_ID].items] == ["first", "last", "new"]
276 # ...and the position moves onto the added item, so a play press starts there
277 assert queue.current_index == 2
278 assert queue.current_item is not None
279 assert queue.current_item.queue_item_id == "new"
280 assert queue.ended is False
281 # adding never starts playback by itself
282 assert "played" not in calls
283
284
285async def test_adding_a_batch_to_a_finished_queue_reports_what_follows() -> None:
286 """Ending the queue cleared its next item; a batch of added items must refresh it."""
287 ctrl, queue, _calls = _enqueue_controller()
288 batch = [
289 QueueItem(queue_id=QUEUE_ID, queue_item_id=x, name=x, duration=100) for x in ("new", "new2")
290 ]
291
292 await ctrl._enqueue_with_option(QUEUE_ID, batch, QueueOption.ADD)
293
294 assert queue.current_item is not None
295 assert queue.current_item.queue_item_id == "new"
296 assert queue.next_item is not None
297 assert queue.next_item.queue_item_id == "new2"
298
299
300# --- what "finished" settles into, per media type ---
301
302
303def _tracker(*items: Any) -> Any:
304 """Build a playback tracker stand-in owning a queue holding the given items."""
305 tracker = MagicMock()
306 tracker._queue_data = {QUEUE_ID: SimpleNamespace(items=list(items), flow_mode_stream_log=[])}
307 return tracker
308
309
310def _queue_stub() -> PlayerQueue:
311 """Build a queue stand-in that has no next item left to play."""
312 return cast(
313 "PlayerQueue",
314 SimpleNamespace(queue_id=QUEUE_ID, display_name="Q1", next_item=None, flow_mode=False),
315 )
316
317
318def _item(media_type: MediaType) -> Any:
319 """Build a queue item stand-in of the given media type."""
320 return SimpleNamespace(
321 media_type=media_type, streamdetails=None, duration=3600, name=media_type.value
322 )
323
324
325def test_finished_music_is_marked_ended() -> None:
326 """A queue that ran out of music keeps its items and says it finished."""
327 item = _item(MediaType.TRACK)
328 tracker = _tracker(item)
329
330 PlaybackTrackerMixin._finish_queue(tracker, _queue_stub(), item)
331
332 tracker.mark_ended.assert_called_once_with(QUEUE_ID)
333 tracker.clear.assert_not_called()
334
335
336def test_a_finished_audiobook_or_episode_is_marked_ended_too() -> None:
337 """Anything with a natural end settles the same way, so the outcome stays predictable."""
338 for media_type in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE):
339 item = _item(media_type)
340 tracker = _tracker(item)
341
342 PlaybackTrackerMixin._finish_queue(tracker, _queue_stub(), item)
343
344 tracker.mark_ended.assert_called_once_with(QUEUE_ID)
345 tracker.clear.assert_not_called()
346
347
348def test_a_live_source_is_left_alone_even_without_a_previous_item() -> None:
349 """A radio that stopped did not run out; leave the queue for a later resume."""
350 for media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
351 tracker = _tracker(_item(media_type))
352
353 # prev_item is None here: the media type is recovered from the queue's last item
354 PlaybackTrackerMixin._finish_queue(tracker, _queue_stub(), None)
355
356 tracker.clear.assert_not_called()
357 tracker.mark_ended.assert_not_called()
358
359
360def test_an_unknown_ending_item_keeps_the_items() -> None:
361 """With no idea what ended, keep the queue rather than destroy it."""
362 tracker = _tracker()
363
364 PlaybackTrackerMixin._finish_queue(tracker, _queue_stub(), None)
365
366 tracker.mark_ended.assert_called_once_with(QUEUE_ID)
367 tracker.clear.assert_not_called()
368
369
370# --- which stops reach the end-of-queue handling at all ---
371
372
373def _stop_states(prev_item: Any) -> tuple[CompareState, CompareState]:
374 """Build the state pair for a queue that went from playing the given item to idle."""
375 prev_state = {
376 "state": PlaybackState.PLAYING,
377 "current_item_id": "i1",
378 "current_item": prev_item,
379 "last_playing_elapsed_time": 3600,
380 }
381 return cast("CompareState", prev_state), cast("CompareState", {"state": PlaybackState.IDLE})
382
383
384def test_a_live_source_never_reaches_the_end_of_queue_handling() -> None:
385 """Radio, audio sources and sound effects short-circuit before the settle is scheduled."""
386 for media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
387 tracker = _tracker()
388 prev_state, new_state = _stop_states(_item(media_type))
389
390 PlaybackTrackerMixin._handle_end_of_queue(tracker, _queue_stub(), prev_state, new_state)
391
392 tracker.mass.create_task.assert_not_called()
393
394
395def test_a_queue_with_something_left_to_play_never_settles() -> None:
396 """Repeat wraps to a next item, so a repeating queue can never reach its end."""
397 tracker = _tracker()
398 queue = _queue_stub()
399 queue.next_item = _item(MediaType.TRACK)
400 prev_state, new_state = _stop_states(_item(MediaType.TRACK))
401
402 PlaybackTrackerMixin._handle_end_of_queue(tracker, queue, prev_state, new_state)
403
404 tracker.mass.create_task.assert_not_called()
405
406
407def test_stopping_part_way_through_a_track_does_not_settle_the_queue() -> None:
408 """The near-completion heuristic is what separates "ran out" from "user stopped it"."""
409 tracker = _tracker()
410 prev_state, new_state = _stop_states(_item(MediaType.TRACK))
411 prev_state["last_playing_elapsed_time"] = 30 # 30s into a 3600s track
412
413 PlaybackTrackerMixin._handle_end_of_queue(tracker, _queue_stub(), prev_state, new_state)
414
415 tracker.mass.create_task.assert_not_called()
416
417
418def test_playing_a_track_to_its_end_settles_the_queue() -> None:
419 """Running out of music schedules the settle (debounced, so late arrivals still win)."""
420 tracker = _tracker()
421 prev_state, new_state = _stop_states(_item(MediaType.TRACK))
422
423 PlaybackTrackerMixin._handle_end_of_queue(tracker, _queue_stub(), prev_state, new_state)
424
425 tracker.mass.create_task.assert_called_once()
426