music-assistant-server

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