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