/
/
/
1"""
2Tests for what Autoplay appends per media type.
3
4Autoplay is a single "keep going" switch: music continues with more music, a podcast episode
5or audiobook with its own successor, and a live source (radio/audio source) is left alone -
6including when the user stops it manually.
7"""
8
9from __future__ import annotations
10
11from types import SimpleNamespace
12from typing import TYPE_CHECKING, Any, cast
13from unittest.mock import AsyncMock, MagicMock, patch
14
15import pytest
16from music_assistant_models.enums import MediaType, PlaybackState
17from music_assistant_models.errors import ProviderUnavailableError
18from music_assistant_models.media_items import (
19 Audiobook,
20 MediaCollection,
21 Podcast,
22 PodcastEpisode,
23 Radio,
24 SoundEffect,
25 Track,
26 UniqueList,
27)
28from music_assistant_models.media_items.metadata import MediaItemCollection, MediaItemMetadata
29from music_assistant_models.media_items.provider_mapping import ProviderMapping
30
31from music_assistant.controllers.player_queues.autoplay import AutoplayMode
32from music_assistant.controllers.player_queues.media_resolver import MediaResolver
33from music_assistant.controllers.player_queues.playback_tracker import PlaybackTrackerMixin
34from music_assistant.controllers.player_queues.queue_loader import QueueLoaderMixin
35from music_assistant.helpers.collections import get_collection_item_id
36
37if TYPE_CHECKING:
38 from collections.abc import AsyncGenerator
39
40 from music_assistant_models.media_items import PlayableMediaItemType
41 from music_assistant_models.player_queue import PlayerQueue
42
43 from music_assistant.controllers.player_queues.helpers import CompareState
44
45
46def _mappings(item_id: str, provider: str, available: bool = True) -> set[ProviderMapping]:
47 """Build a single provider mapping for the given item."""
48 return {
49 ProviderMapping(
50 item_id=item_id,
51 provider_domain=provider,
52 provider_instance=provider,
53 available=available,
54 )
55 }
56
57
58def _podcast(item_id: str = "show1", provider: str = "test_prov") -> Podcast:
59 """Build a minimal Podcast."""
60 return Podcast(
61 item_id=item_id,
62 provider=provider,
63 name="Show",
64 provider_mappings=_mappings(item_id, provider),
65 )
66
67
68def _episode(
69 item_id: str,
70 position: int,
71 podcast: Podcast,
72 *,
73 fully_played: bool | None = None,
74 provider: str = "test_prov",
75) -> PodcastEpisode:
76 """Build a minimal PodcastEpisode of the given podcast."""
77 return PodcastEpisode(
78 item_id=item_id,
79 provider=provider,
80 name=f"Episode {position}",
81 provider_mappings=_mappings(item_id, provider),
82 position=position,
83 podcast=podcast,
84 fully_played=fully_played,
85 )
86
87
88def _audiobook(
89 item_id: str, *, provider: str = "library", collections: list[str] | None = None
90) -> Audiobook:
91 """Build a minimal Audiobook, optionally part of the given collection(s)."""
92 return Audiobook(
93 item_id=item_id,
94 provider=provider,
95 name=f"Book {item_id}",
96 provider_mappings=_mappings(item_id, provider),
97 metadata=MediaItemMetadata(
98 collections=UniqueList([MediaItemCollection(title=x) for x in collections or []])
99 if collections
100 else None
101 ),
102 )
103
104
105def _collection(name: str, *books: Audiobook) -> MediaCollection[Audiobook]:
106 """Build a library audiobook collection holding the given books, in order."""
107 return MediaCollection(
108 item_id=get_collection_item_id(name, MediaType.AUDIOBOOK),
109 name=name,
110 provider="library",
111 provider_mappings=set(),
112 items=UniqueList(list(books)),
113 )
114
115
116async def _aiter(episodes: list[PodcastEpisode]) -> AsyncGenerator[PodcastEpisode]:
117 """Yield the given episodes as the podcasts controller does."""
118 for episode in episodes:
119 yield episode
120
121
122def _resolver() -> tuple[MediaResolver, MagicMock]:
123 """Build a MediaResolver and return it with its mocked owning controller."""
124 queues = MagicMock()
125 return MediaResolver(queues), queues
126
127
128# --- next podcast episode ---
129
130
131async def test_next_podcast_episode_follows_the_played_one() -> None:
132 """The episode after the given one is returned, with its resume position applied."""
133 resolver, queues = _resolver()
134 podcast = _podcast()
135 played = _episode("ep1", 1, podcast)
136 queues.mass.music.podcasts.episodes = MagicMock(
137 return_value=_aiter(
138 [_episode("ep1", 1, podcast), _episode("ep2", 2, podcast), _episode("ep3", 3, podcast)]
139 )
140 )
141 queues.mass.music.get_resume_position = AsyncMock(return_value=(False, 42_000))
142
143 result = await resolver.get_next_podcast_episode(played)
144
145 assert result is not None
146 assert result.item_id == "ep2"
147 assert result.resume_position_ms == 42_000
148
149
150async def test_next_podcast_episode_skips_fully_played() -> None:
151 """An episode the user already finished is passed over."""
152 resolver, queues = _resolver()
153 podcast = _podcast()
154 queues.mass.music.podcasts.episodes = MagicMock(
155 return_value=_aiter(
156 [
157 _episode("ep1", 1, podcast),
158 _episode("ep2", 2, podcast, fully_played=True),
159 _episode("ep3", 3, podcast),
160 ]
161 )
162 )
163 queues.mass.music.get_resume_position = AsyncMock(return_value=(False, 0))
164
165 result = await resolver.get_next_podcast_episode(_episode("ep1", 1, podcast))
166
167 assert result is not None
168 assert result.item_id == "ep3"
169
170
171async def test_next_podcast_episode_at_end_of_feed() -> None:
172 """The last episode of a podcast has no successor, so playback stops there."""
173 resolver, queues = _resolver()
174 podcast = _podcast()
175 queues.mass.music.podcasts.episodes = MagicMock(
176 return_value=_aiter([_episode("ep1", 1, podcast), _episode("ep2", 2, podcast)])
177 )
178
179 assert await resolver.get_next_podcast_episode(_episode("ep2", 2, podcast)) is None
180
181
182# --- next audiobook ---
183
184
185async def test_next_audiobook_follows_the_collection_order() -> None:
186 """The next book of the series is returned, with its resume position applied."""
187 resolver, queues = _resolver()
188 book1 = _audiobook("1", collections=["Series"])
189 book2 = _audiobook("2", collections=["Series"])
190 queues.mass.music.audiobooks.get_collection = AsyncMock(
191 return_value=_collection("Series", book1, book2)
192 )
193 queues.mass.music.get_resume_position = AsyncMock(return_value=(False, 120_000))
194
195 result = await resolver.get_next_audiobook(book1)
196
197 assert result is not None
198 assert result.item_id == "2"
199 assert result.resume_position_ms == 120_000
200 assert queues.mass.music.audiobooks.get_collection.await_args.args == (
201 get_collection_item_id("Series", MediaType.AUDIOBOOK),
202 )
203
204
205async def test_next_audiobook_skips_fully_played_books() -> None:
206 """A book of the series that was already finished is passed over."""
207 resolver, queues = _resolver()
208 books = [_audiobook(str(i), collections=["Series"]) for i in (1, 2, 3)]
209 queues.mass.music.audiobooks.get_collection = AsyncMock(
210 return_value=_collection("Series", *books)
211 )
212 queues.mass.music.get_resume_position = AsyncMock(side_effect=[(True, 0), (False, 0)])
213
214 result = await resolver.get_next_audiobook(books[0])
215
216 assert result is not None
217 assert result.item_id == "3"
218
219
220async def test_next_audiobook_at_end_of_collection() -> None:
221 """The last book of a series has no successor, so playback stops there."""
222 resolver, queues = _resolver()
223 book1 = _audiobook("1", collections=["Series"])
224 book2 = _audiobook("2", collections=["Series"])
225 queues.mass.music.audiobooks.get_collection = AsyncMock(
226 return_value=_collection("Series", book1, book2)
227 )
228
229 assert await resolver.get_next_audiobook(book2) is None
230
231
232async def test_standalone_audiobook_has_no_successor() -> None:
233 """A book that is not part of a collection ends the queue."""
234 resolver, queues = _resolver()
235
236 assert await resolver.get_next_audiobook(_audiobook("1")) is None
237 queues.mass.music.audiobooks.get_collection.assert_not_called()
238
239
240async def test_audiobook_outside_the_library_has_no_successor() -> None:
241 """Collections are built from library metadata, so a provider-only book stops the queue."""
242 resolver, queues = _resolver()
243 queues.mass.music.audiobooks.get_library_item_by_prov_id = AsyncMock(return_value=None)
244
245 assert await resolver.get_next_audiobook(_audiobook("1", provider="test_prov")) is None
246
247
248# --- dispatch on the queue's last item ---
249
250
251def _queue_item(media_item: PlayableMediaItemType) -> Any:
252 """Build a queue item stand-in for the given media item."""
253 return SimpleNamespace(
254 media_item=media_item, media_type=media_item.media_type, name=media_item.name
255 )
256
257
258def _loader(*items: Any, seeds: list[Any] | None = None) -> Any:
259 """Build a queue loader stand-in holding a queue with the given items (autoplay on)."""
260 loader = MagicMock()
261 loader.get = MagicMock(
262 return_value=SimpleNamespace(queue_id="q1", display_name="Queue", autoplay_enabled=True)
263 )
264 loader._queue_data = {
265 "q1": SimpleNamespace(items=list(items), enqueued_media_items=seeds or [], userid=None)
266 }
267 loader._fill_autoplay_music_tracks = AsyncMock()
268 loader._fill_autoplay_next_in_series = AsyncMock()
269 loader.load = AsyncMock()
270 return loader
271
272
273def _track(item_id: str = "t1") -> Track:
274 """Build a minimal Track."""
275 return Track(
276 item_id=item_id,
277 provider="library",
278 name=f"Track {item_id}",
279 provider_mappings=_mappings(item_id, "library"),
280 )
281
282
283async def test_queue_ending_on_music_appends_music() -> None:
284 """A queue that ends on a track keeps using the configured (music) Autoplay mode."""
285 loader = _loader(_queue_item(_track()), seeds=[_track()])
286
287 await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
288
289 loader._fill_autoplay_music_tracks.assert_awaited_once_with("q1")
290 loader._fill_autoplay_next_in_series.assert_not_awaited()
291
292
293async def test_queue_ending_on_audiobook_never_appends_music() -> None:
294 """Music enqueued earlier must not leak into a queue that now ends on an audiobook."""
295 book = _queue_item(_audiobook("1"))
296 loader = _loader(_queue_item(_track()), book, seeds=[_track()])
297
298 await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
299
300 loader._fill_autoplay_music_tracks.assert_not_awaited()
301 loader._fill_autoplay_next_in_series.assert_awaited_once_with("q1", book)
302
303
304async def test_queue_ending_on_podcast_episode_continues_the_podcast() -> None:
305 """A queue that ends on a podcast episode continues with the next episode."""
306 episode = _queue_item(_episode("ep1", 1, _podcast()))
307 loader = _loader(episode, seeds=[_track()])
308
309 await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
310
311 loader._fill_autoplay_music_tracks.assert_not_awaited()
312 loader._fill_autoplay_next_in_series.assert_awaited_once_with("q1", episode)
313
314
315async def test_queue_ending_on_live_source_appends_nothing() -> None:
316 """Radio and audio sources have no natural end, so Autoplay does not apply."""
317 for media_item in (
318 Radio(
319 item_id="r1",
320 provider="test_prov",
321 name="Radio",
322 provider_mappings=_mappings("r1", "test_prov"),
323 ),
324 SimpleNamespace(media_type=MediaType.AUDIO_SOURCE, name="Line in"),
325 ):
326 loader = _loader(_queue_item(cast("PlayableMediaItemType", media_item)), seeds=[_track()])
327
328 await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
329
330 loader._fill_autoplay_music_tracks.assert_not_awaited()
331 loader._fill_autoplay_next_in_series.assert_not_awaited()
332
333
334async def test_queue_ending_on_a_sound_effect_appends_nothing() -> None:
335 """A one-off clip (a notification, a TTS message) is not music to continue from."""
336 effect = SoundEffect(
337 item_id="http://example.com/notification.mp3",
338 provider="builtin",
339 name="notification",
340 provider_mappings=_mappings("http://example.com/notification.mp3", "builtin"),
341 )
342 loader = _loader(_queue_item(effect), seeds=[_track()])
343
344 await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
345
346 loader._fill_autoplay_music_tracks.assert_not_awaited()
347 loader._fill_autoplay_next_in_series.assert_not_awaited()
348
349
350async def test_autoplay_disabled_appends_nothing() -> None:
351 """With Autoplay off nothing is appended, whatever the queue ends on."""
352 loader = _loader(_queue_item(_track()), seeds=[_track()])
353 loader.get.return_value.autoplay_enabled = False
354
355 await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
356
357 loader._fill_autoplay_music_tracks.assert_not_awaited()
358 loader._fill_autoplay_next_in_series.assert_not_awaited()
359
360
361@pytest.mark.parametrize("reregistered", [False, True])
362async def test_autoplay_dispatch_bails_when_the_queue_is_removed_mid_lookup(
363 reregistered: bool,
364) -> None:
365 """A queue removed (or re-registered fresh) mid-lookup gets no refill."""
366 loader = _loader(_queue_item(_track()), seeds=[_track()])
367 loader._queue_data["q1"].userid = "u1"
368
369 async def _lookup_and_remove_queue(*_args: Any, **_kwargs: Any) -> Any:
370 loader._queue_data.pop("q1")
371 if reregistered:
372 loader._queue_data["q1"] = SimpleNamespace(
373 items=[], enqueued_media_items=[], userid="u1"
374 )
375 return MagicMock()
376
377 loader.mass.webserver.auth.get_user = AsyncMock(side_effect=_lookup_and_remove_queue)
378
379 await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
380
381 loader._fill_autoplay_music_tracks.assert_not_awaited()
382 loader._fill_autoplay_next_in_series.assert_not_awaited()
383
384
385@pytest.mark.parametrize("reregistered", [False, True])
386async def test_dynamic_refill_bails_when_the_queue_is_removed(reregistered: bool) -> None:
387 """A dynamic refill firing for a removed (or re-registered) queue loads nothing."""
388 loader = MagicMock()
389 loader.load = AsyncMock()
390 if reregistered:
391 loader._queue_data = {
392 "q1": SimpleNamespace(userid=None, queue=SimpleNamespace(current_index=None))
393 }
394
395 async def _fill_and_replace_queue(*_args: Any, **_kwargs: Any) -> list[Track]:
396 loader._queue_data["q1"] = SimpleNamespace(
397 userid=None, queue=SimpleNamespace(current_index=None), items=[]
398 )
399 return [_track()]
400
401 loader._managed_pool.fill = AsyncMock(side_effect=_fill_and_replace_queue)
402 else:
403 # the delayed refill timer fires after the queue was already removed
404 loader._queue_data = {}
405
406 await QueueLoaderMixin._fill_dynamic_tracks(loader, "q1")
407
408 loader.load.assert_not_called()
409 if not reregistered:
410 loader._managed_pool.fill.assert_not_called()
411
412
413async def test_music_refill_without_seeds_appends_nothing() -> None:
414 """The music refill needs an enqueued item as its seed."""
415 loader = _loader(_queue_item(_track()))
416
417 await QueueLoaderMixin._fill_autoplay_music_tracks(loader, "q1")
418
419 loader.load.assert_not_awaited()
420
421
422async def test_auto_music_refill_falls_back_to_library_after_provider_failure() -> None:
423 """AUTO mode falls back to library tracks when similar-track providers fail."""
424 loader = _loader(_queue_item(_track()), seeds=[_track()])
425 loader._autoplay.resolve_mode.return_value = AutoplayMode.AUTO
426 loader._autoplay.get_library_tracks = AsyncMock(return_value=[])
427 loader._get_similar_tracks = AsyncMock(side_effect=ProviderUnavailableError("offline"))
428 loader.mass.music.recency.snapshot = AsyncMock(return_value=MagicMock())
429
430 with patch(
431 "music_assistant.controllers.player_queues.queue_loader.gate_tracks", return_value=[]
432 ):
433 await QueueLoaderMixin._fill_autoplay_music_tracks(loader, "q1")
434
435 loader._autoplay.get_library_tracks.assert_awaited_once()
436
437
438@pytest.mark.parametrize("reregistered", [False, True])
439async def test_music_refill_bails_when_the_queue_is_removed_mid_fetch(
440 reregistered: bool,
441) -> None:
442 """A queue removed (or re-registered fresh) mid-fetch must not receive the stale batch."""
443 track = _track()
444 loader = _loader(_queue_item(track), seeds=[track])
445 loader._autoplay.resolve_mode.return_value = AutoplayMode.LIBRARY
446
447 async def _fetch_and_remove_queue(*_args: Any, **_kwargs: Any) -> list[Track]:
448 loader._queue_data.pop("q1")
449 if reregistered:
450 loader._queue_data["q1"] = SimpleNamespace(
451 items=[], enqueued_media_items=[], userid=None
452 )
453 return [track]
454
455 loader._autoplay.get_library_tracks = AsyncMock(side_effect=_fetch_and_remove_queue)
456 loader.mass.music.recency.snapshot = AsyncMock(return_value=MagicMock())
457
458 with patch(
459 "music_assistant.controllers.player_queues.queue_loader.gate_tracks",
460 return_value=[track],
461 ):
462 await QueueLoaderMixin._fill_autoplay_music_tracks(loader, "q1")
463
464 loader.load.assert_not_awaited()
465
466
467# --- appending the resolved successor ---
468
469
470async def test_next_in_series_appends_the_successor() -> None:
471 """The resolved next episode is appended behind the queue's last item."""
472 podcast = _podcast()
473 last_item = _queue_item(_episode("ep1", 1, podcast))
474 loader = _loader(last_item)
475 loader._media_resolver.get_next_podcast_episode = AsyncMock(
476 return_value=_episode("ep2", 2, podcast)
477 )
478
479 await QueueLoaderMixin._fill_autoplay_next_in_series(loader, "q1", last_item)
480
481 loader.load.assert_awaited_once()
482 queue_items = loader.load.await_args.args[1]
483 assert [x.media_item.item_id for x in queue_items] == ["ep2"]
484
485
486async def test_next_in_series_without_successor_appends_nothing() -> None:
487 """Nothing is appended when the podcast/series has nothing left to play."""
488 last_item = _queue_item(_episode("ep1", 1, _podcast()))
489 loader = _loader(last_item)
490 loader._media_resolver.get_next_podcast_episode = AsyncMock(return_value=None)
491
492 await QueueLoaderMixin._fill_autoplay_next_in_series(loader, "q1", last_item)
493
494 loader.load.assert_not_awaited()
495
496
497async def test_next_in_series_skips_an_already_queued_item() -> None:
498 """A successor the user already queued themselves is not added twice."""
499 podcast = _podcast()
500 next_episode = _episode("ep2", 2, podcast)
501 last_item = _queue_item(_episode("ep1", 1, podcast))
502 loader = _loader(_queue_item(next_episode), last_item)
503 loader._media_resolver.get_next_podcast_episode = AsyncMock(return_value=next_episode)
504
505 await QueueLoaderMixin._fill_autoplay_next_in_series(loader, "q1", last_item)
506
507 loader.load.assert_not_awaited()
508
509
510# --- live sources survive a manual stop ---
511
512
513def _stop_states(prev_item: Any) -> tuple[CompareState, CompareState]:
514 """Build the state pair for a queue that went from playing the given item to idle."""
515 prev_state = {
516 "state": PlaybackState.PLAYING,
517 "current_item_id": "i1",
518 "current_item": prev_item,
519 "last_playing_elapsed_time": 3600,
520 }
521 return cast("CompareState", prev_state), cast("CompareState", {"state": PlaybackState.IDLE})
522
523
524def _tracker() -> Any:
525 """Build a playback tracker stand-in for a single non-flow queue."""
526 tracker = MagicMock()
527 tracker._queue_data = {"q1": SimpleNamespace(flow_mode_stream_log=[])}
528 return tracker
529
530
531def _stopped_queue() -> PlayerQueue:
532 """Build a queue stand-in that has no next item left to play."""
533 return cast("PlayerQueue", SimpleNamespace(queue_id="q1", next_item=None, flow_mode=False))
534
535
536def test_live_source_queue_survives_a_manual_stop() -> None:
537 """Stopping a radio means the source stopped, not that the queue ran out: keep the items."""
538 for media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
539 tracker = _tracker()
540 prev_state, new_state = _stop_states(
541 SimpleNamespace(media_type=media_type, streamdetails=None, duration=None)
542 )
543
544 PlaybackTrackerMixin._handle_end_of_queue(tracker, _stopped_queue(), prev_state, new_state)
545
546 tracker.mass.create_task.assert_not_called()
547
548
549def test_finished_track_queue_is_settled_on_stop() -> None:
550 """A queue that ran out of music is still settled (resumed, ended or cleared) as before."""
551 tracker = _tracker()
552 prev_state, new_state = _stop_states(
553 SimpleNamespace(media_type=MediaType.TRACK, streamdetails=None, duration=3600)
554 )
555
556 PlaybackTrackerMixin._handle_end_of_queue(tracker, _stopped_queue(), prev_state, new_state)
557
558 tracker.mass.create_task.assert_called_once()
559
560
561@pytest.mark.parametrize("reregistered", [False, True])
562async def test_settle_task_bails_when_the_queue_is_removed_while_waiting(
563 reregistered: bool,
564) -> None:
565 """A queue removed (or re-registered) during the settle delay is left alone entirely."""
566 tracker = _tracker()
567 prev_state, new_state = _stop_states(
568 SimpleNamespace(media_type=MediaType.TRACK, streamdetails=None, duration=3600)
569 )
570 PlaybackTrackerMixin._handle_end_of_queue(tracker, _stopped_queue(), prev_state, new_state)
571 settle_coro = tracker.mass.create_task.call_args.args[0]
572
573 tracker._queue_data.pop("q1")
574 if reregistered:
575 tracker._queue_data["q1"] = SimpleNamespace(flow_mode_stream_log=[])
576 with patch(
577 "music_assistant.controllers.player_queues.playback_tracker.asyncio.sleep",
578 AsyncMock(),
579 ):
580 await settle_coro
581
582 tracker.play_index.assert_not_called()
583 tracker.load.assert_not_called()
584 tracker._finish_queue.assert_not_called()
585
586
587async def test_settle_task_skips_finish_when_the_queue_vanishes_during_fetch() -> None:
588 """A failed dynamic fetch on a removed queue must not mark anything as ended."""
589 tracker = _tracker()
590 tracker._queue_data["q1"].userid = None
591 prev_state, new_state = _stop_states(
592 SimpleNamespace(media_type=MediaType.TRACK, streamdetails=None, duration=3600)
593 )
594 queue = cast(
595 "PlayerQueue",
596 SimpleNamespace(
597 queue_id="q1",
598 next_item=None,
599 flow_mode=False,
600 state=PlaybackState.IDLE,
601 current_index=None,
602 display_name="Q1",
603 ),
604 )
605 PlaybackTrackerMixin._handle_end_of_queue(tracker, queue, prev_state, new_state)
606 settle_coro = tracker.mass.create_task.call_args.args[0]
607
608 async def _fetch_and_remove_queue(*_args: Any, **_kwargs: Any) -> list[Track]:
609 tracker._queue_data.pop("q1")
610 raise ProviderUnavailableError("provider unloaded")
611
612 tracker._media_resolver.get_dynamic_source_tracks = AsyncMock(
613 side_effect=_fetch_and_remove_queue
614 )
615 with (
616 patch(
617 "music_assistant.controllers.player_queues.playback_tracker.asyncio.sleep",
618 AsyncMock(),
619 ),
620 patch(
621 "music_assistant.controllers.player_queues.playback_tracker.find_dynamic_source",
622 return_value=MagicMock(),
623 ),
624 ):
625 await settle_coro
626
627 tracker._finish_queue.assert_not_called()
628