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
14
15from music_assistant_models.enums import MediaType, PlaybackState
16from music_assistant_models.media_items import (
17    Audiobook,
18    MediaCollection,
19    Podcast,
20    PodcastEpisode,
21    Radio,
22    Track,
23    UniqueList,
24)
25from music_assistant_models.media_items.metadata import MediaItemCollection, MediaItemMetadata
26from music_assistant_models.media_items.provider_mapping import ProviderMapping
27
28from music_assistant.controllers.player_queues.media_resolver import MediaResolver
29from music_assistant.controllers.player_queues.playback_tracker import PlaybackTrackerMixin
30from music_assistant.controllers.player_queues.queue_loader import QueueLoaderMixin
31from music_assistant.helpers.collections import get_collection_item_id
32
33if TYPE_CHECKING:
34    from collections.abc import AsyncGenerator
35
36    from music_assistant_models.media_items import PlayableMediaItemType
37    from music_assistant_models.player_queue import PlayerQueue
38
39    from music_assistant.controllers.player_queues.helpers import CompareState
40
41
42def _mappings(item_id: str, provider: str, available: bool = True) -> set[ProviderMapping]:
43    """Build a single provider mapping for the given item."""
44    return {
45        ProviderMapping(
46            item_id=item_id,
47            provider_domain=provider,
48            provider_instance=provider,
49            available=available,
50        )
51    }
52
53
54def _podcast(item_id: str = "show1", provider: str = "test_prov") -> Podcast:
55    """Build a minimal Podcast."""
56    return Podcast(
57        item_id=item_id,
58        provider=provider,
59        name="Show",
60        provider_mappings=_mappings(item_id, provider),
61    )
62
63
64def _episode(
65    item_id: str,
66    position: int,
67    podcast: Podcast,
68    *,
69    fully_played: bool | None = None,
70    provider: str = "test_prov",
71) -> PodcastEpisode:
72    """Build a minimal PodcastEpisode of the given podcast."""
73    return PodcastEpisode(
74        item_id=item_id,
75        provider=provider,
76        name=f"Episode {position}",
77        provider_mappings=_mappings(item_id, provider),
78        position=position,
79        podcast=podcast,
80        fully_played=fully_played,
81    )
82
83
84def _audiobook(
85    item_id: str, *, provider: str = "library", collections: list[str] | None = None
86) -> Audiobook:
87    """Build a minimal Audiobook, optionally part of the given collection(s)."""
88    return Audiobook(
89        item_id=item_id,
90        provider=provider,
91        name=f"Book {item_id}",
92        provider_mappings=_mappings(item_id, provider),
93        metadata=MediaItemMetadata(
94            collections=UniqueList([MediaItemCollection(title=x) for x in collections or []])
95            if collections
96            else None
97        ),
98    )
99
100
101def _collection(name: str, *books: Audiobook) -> MediaCollection[Audiobook]:
102    """Build a library audiobook collection holding the given books, in order."""
103    return MediaCollection(
104        item_id=get_collection_item_id(name, MediaType.AUDIOBOOK),
105        name=name,
106        provider="library",
107        provider_mappings=set(),
108        items=UniqueList(list(books)),
109    )
110
111
112async def _aiter(episodes: list[PodcastEpisode]) -> AsyncGenerator[PodcastEpisode]:
113    """Yield the given episodes as the podcasts controller does."""
114    for episode in episodes:
115        yield episode
116
117
118def _resolver() -> tuple[MediaResolver, MagicMock]:
119    """Build a MediaResolver and return it with its mocked owning controller."""
120    queues = MagicMock()
121    return MediaResolver(queues), queues
122
123
124# --- next podcast episode ---
125
126
127async def test_next_podcast_episode_follows_the_played_one() -> None:
128    """The episode after the given one is returned, with its resume position applied."""
129    resolver, queues = _resolver()
130    podcast = _podcast()
131    played = _episode("ep1", 1, podcast)
132    queues.mass.music.podcasts.episodes = MagicMock(
133        return_value=_aiter(
134            [_episode("ep1", 1, podcast), _episode("ep2", 2, podcast), _episode("ep3", 3, podcast)]
135        )
136    )
137    queues.mass.music.get_resume_position = AsyncMock(return_value=(False, 42_000))
138
139    result = await resolver.get_next_podcast_episode(played)
140
141    assert result is not None
142    assert result.item_id == "ep2"
143    assert result.resume_position_ms == 42_000
144
145
146async def test_next_podcast_episode_skips_fully_played() -> None:
147    """An episode the user already finished is passed over."""
148    resolver, queues = _resolver()
149    podcast = _podcast()
150    queues.mass.music.podcasts.episodes = MagicMock(
151        return_value=_aiter(
152            [
153                _episode("ep1", 1, podcast),
154                _episode("ep2", 2, podcast, fully_played=True),
155                _episode("ep3", 3, podcast),
156            ]
157        )
158    )
159    queues.mass.music.get_resume_position = AsyncMock(return_value=(False, 0))
160
161    result = await resolver.get_next_podcast_episode(_episode("ep1", 1, podcast))
162
163    assert result is not None
164    assert result.item_id == "ep3"
165
166
167async def test_next_podcast_episode_at_end_of_feed() -> None:
168    """The last episode of a podcast has no successor, so playback stops there."""
169    resolver, queues = _resolver()
170    podcast = _podcast()
171    queues.mass.music.podcasts.episodes = MagicMock(
172        return_value=_aiter([_episode("ep1", 1, podcast), _episode("ep2", 2, podcast)])
173    )
174
175    assert await resolver.get_next_podcast_episode(_episode("ep2", 2, podcast)) is None
176
177
178# --- next audiobook ---
179
180
181async def test_next_audiobook_follows_the_collection_order() -> None:
182    """The next book of the series is returned, with its resume position applied."""
183    resolver, queues = _resolver()
184    book1 = _audiobook("1", collections=["Series"])
185    book2 = _audiobook("2", collections=["Series"])
186    queues.mass.music.audiobooks.get_collection = AsyncMock(
187        return_value=_collection("Series", book1, book2)
188    )
189    queues.mass.music.get_resume_position = AsyncMock(return_value=(False, 120_000))
190
191    result = await resolver.get_next_audiobook(book1)
192
193    assert result is not None
194    assert result.item_id == "2"
195    assert result.resume_position_ms == 120_000
196    assert queues.mass.music.audiobooks.get_collection.await_args.args == (
197        get_collection_item_id("Series", MediaType.AUDIOBOOK),
198    )
199
200
201async def test_next_audiobook_skips_fully_played_books() -> None:
202    """A book of the series that was already finished is passed over."""
203    resolver, queues = _resolver()
204    books = [_audiobook(str(i), collections=["Series"]) for i in (1, 2, 3)]
205    queues.mass.music.audiobooks.get_collection = AsyncMock(
206        return_value=_collection("Series", *books)
207    )
208    queues.mass.music.get_resume_position = AsyncMock(side_effect=[(True, 0), (False, 0)])
209
210    result = await resolver.get_next_audiobook(books[0])
211
212    assert result is not None
213    assert result.item_id == "3"
214
215
216async def test_next_audiobook_at_end_of_collection() -> None:
217    """The last book of a series has no successor, so playback stops there."""
218    resolver, queues = _resolver()
219    book1 = _audiobook("1", collections=["Series"])
220    book2 = _audiobook("2", collections=["Series"])
221    queues.mass.music.audiobooks.get_collection = AsyncMock(
222        return_value=_collection("Series", book1, book2)
223    )
224
225    assert await resolver.get_next_audiobook(book2) is None
226
227
228async def test_standalone_audiobook_has_no_successor() -> None:
229    """A book that is not part of a collection ends the queue."""
230    resolver, queues = _resolver()
231
232    assert await resolver.get_next_audiobook(_audiobook("1")) is None
233    queues.mass.music.audiobooks.get_collection.assert_not_called()
234
235
236async def test_audiobook_outside_the_library_has_no_successor() -> None:
237    """Collections are built from library metadata, so a provider-only book stops the queue."""
238    resolver, queues = _resolver()
239    queues.mass.music.audiobooks.get_library_item_by_prov_id = AsyncMock(return_value=None)
240
241    assert await resolver.get_next_audiobook(_audiobook("1", provider="test_prov")) is None
242
243
244# --- dispatch on the queue's last item ---
245
246
247def _queue_item(media_item: PlayableMediaItemType) -> Any:
248    """Build a queue item stand-in for the given media item."""
249    return SimpleNamespace(
250        media_item=media_item, media_type=media_item.media_type, name=media_item.name
251    )
252
253
254def _loader(*items: Any, seeds: list[Any] | None = None) -> Any:
255    """Build a queue loader stand-in holding a queue with the given items (autoplay on)."""
256    loader = MagicMock()
257    loader.get = MagicMock(
258        return_value=SimpleNamespace(queue_id="q1", display_name="Queue", autoplay_enabled=True)
259    )
260    loader._queue_data = {
261        "q1": SimpleNamespace(items=list(items), enqueued_media_items=seeds or [], userid=None)
262    }
263    loader._fill_autoplay_music_tracks = AsyncMock()
264    loader._fill_autoplay_next_in_series = AsyncMock()
265    loader.load = AsyncMock()
266    return loader
267
268
269def _track(item_id: str = "t1") -> Track:
270    """Build a minimal Track."""
271    return Track(
272        item_id=item_id,
273        provider="library",
274        name=f"Track {item_id}",
275        provider_mappings=_mappings(item_id, "library"),
276    )
277
278
279async def test_queue_ending_on_music_appends_music() -> None:
280    """A queue that ends on a track keeps using the configured (music) Autoplay mode."""
281    loader = _loader(_queue_item(_track()), seeds=[_track()])
282
283    await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
284
285    loader._fill_autoplay_music_tracks.assert_awaited_once_with("q1")
286    loader._fill_autoplay_next_in_series.assert_not_awaited()
287
288
289async def test_queue_ending_on_audiobook_never_appends_music() -> None:
290    """Music enqueued earlier must not leak into a queue that now ends on an audiobook."""
291    book = _queue_item(_audiobook("1"))
292    loader = _loader(_queue_item(_track()), book, seeds=[_track()])
293
294    await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
295
296    loader._fill_autoplay_music_tracks.assert_not_awaited()
297    loader._fill_autoplay_next_in_series.assert_awaited_once_with("q1", book)
298
299
300async def test_queue_ending_on_podcast_episode_continues_the_podcast() -> None:
301    """A queue that ends on a podcast episode continues with the next episode."""
302    episode = _queue_item(_episode("ep1", 1, _podcast()))
303    loader = _loader(episode, seeds=[_track()])
304
305    await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
306
307    loader._fill_autoplay_music_tracks.assert_not_awaited()
308    loader._fill_autoplay_next_in_series.assert_awaited_once_with("q1", episode)
309
310
311async def test_queue_ending_on_live_source_appends_nothing() -> None:
312    """Radio and audio sources have no natural end, so Autoplay does not apply."""
313    for media_item in (
314        Radio(
315            item_id="r1",
316            provider="test_prov",
317            name="Radio",
318            provider_mappings=_mappings("r1", "test_prov"),
319        ),
320        SimpleNamespace(media_type=MediaType.AUDIO_SOURCE, name="Line in"),
321    ):
322        loader = _loader(_queue_item(cast("PlayableMediaItemType", media_item)), seeds=[_track()])
323
324        await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
325
326        loader._fill_autoplay_music_tracks.assert_not_awaited()
327        loader._fill_autoplay_next_in_series.assert_not_awaited()
328
329
330async def test_autoplay_disabled_appends_nothing() -> None:
331    """With Autoplay off nothing is appended, whatever the queue ends on."""
332    loader = _loader(_queue_item(_track()), seeds=[_track()])
333    loader.get.return_value.autoplay_enabled = False
334
335    await QueueLoaderMixin._fill_autoplay_tracks(loader, "q1")
336
337    loader._fill_autoplay_music_tracks.assert_not_awaited()
338    loader._fill_autoplay_next_in_series.assert_not_awaited()
339
340
341async def test_music_refill_without_seeds_appends_nothing() -> None:
342    """The music refill needs an enqueued item as its seed."""
343    loader = _loader(_queue_item(_track()))
344
345    await QueueLoaderMixin._fill_autoplay_music_tracks(loader, "q1")
346
347    loader.load.assert_not_awaited()
348
349
350# --- appending the resolved successor ---
351
352
353async def test_next_in_series_appends_the_successor() -> None:
354    """The resolved next episode is appended behind the queue's last item."""
355    podcast = _podcast()
356    last_item = _queue_item(_episode("ep1", 1, podcast))
357    loader = _loader(last_item)
358    loader._media_resolver.get_next_podcast_episode = AsyncMock(
359        return_value=_episode("ep2", 2, podcast)
360    )
361
362    await QueueLoaderMixin._fill_autoplay_next_in_series(loader, "q1", last_item)
363
364    loader.load.assert_awaited_once()
365    queue_items = loader.load.await_args.args[1]
366    assert [x.media_item.item_id for x in queue_items] == ["ep2"]
367
368
369async def test_next_in_series_without_successor_appends_nothing() -> None:
370    """Nothing is appended when the podcast/series has nothing left to play."""
371    last_item = _queue_item(_episode("ep1", 1, _podcast()))
372    loader = _loader(last_item)
373    loader._media_resolver.get_next_podcast_episode = AsyncMock(return_value=None)
374
375    await QueueLoaderMixin._fill_autoplay_next_in_series(loader, "q1", last_item)
376
377    loader.load.assert_not_awaited()
378
379
380async def test_next_in_series_skips_an_already_queued_item() -> None:
381    """A successor the user already queued themselves is not added twice."""
382    podcast = _podcast()
383    next_episode = _episode("ep2", 2, podcast)
384    last_item = _queue_item(_episode("ep1", 1, podcast))
385    loader = _loader(_queue_item(next_episode), last_item)
386    loader._media_resolver.get_next_podcast_episode = AsyncMock(return_value=next_episode)
387
388    await QueueLoaderMixin._fill_autoplay_next_in_series(loader, "q1", last_item)
389
390    loader.load.assert_not_awaited()
391
392
393# --- live sources survive a manual stop ---
394
395
396def _stop_states(prev_item: Any) -> tuple[CompareState, CompareState]:
397    """Build the state pair for a queue that went from playing the given item to idle."""
398    prev_state = {
399        "state": PlaybackState.PLAYING,
400        "current_item_id": "i1",
401        "current_item": prev_item,
402        "last_playing_elapsed_time": 3600,
403    }
404    return cast("CompareState", prev_state), cast("CompareState", {"state": PlaybackState.IDLE})
405
406
407def _tracker() -> Any:
408    """Build a playback tracker stand-in for a single non-flow queue."""
409    tracker = MagicMock()
410    tracker._queue_data = {"q1": SimpleNamespace(flow_mode_stream_log=[])}
411    return tracker
412
413
414def _stopped_queue() -> PlayerQueue:
415    """Build a queue stand-in that has no next item left to play."""
416    return cast("PlayerQueue", SimpleNamespace(queue_id="q1", next_item=None, flow_mode=False))
417
418
419def test_live_source_queue_survives_a_manual_stop() -> None:
420    """Stopping a radio means the source stopped, not that the queue ran out: keep the items."""
421    for media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
422        tracker = _tracker()
423        prev_state, new_state = _stop_states(
424            SimpleNamespace(media_type=media_type, streamdetails=None, duration=None)
425        )
426
427        PlaybackTrackerMixin._handle_end_of_queue(tracker, _stopped_queue(), prev_state, new_state)
428
429        tracker.mass.create_task.assert_not_called()
430
431
432def test_finished_track_queue_is_settled_on_stop() -> None:
433    """A queue that ran out of music is still settled (resumed, ended or cleared) as before."""
434    tracker = _tracker()
435    prev_state, new_state = _stop_states(
436        SimpleNamespace(media_type=MediaType.TRACK, streamdetails=None, duration=3600)
437    )
438
439    PlaybackTrackerMixin._handle_end_of_queue(tracker, _stopped_queue(), prev_state, new_state)
440
441    tracker.mass.create_task.assert_called_once()
442