/
/
/
1"""Tests for the Autoplay helper and the per-queue autoplay/crossfade config entries."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import TYPE_CHECKING, cast
7from unittest.mock import AsyncMock, MagicMock
8
9from music_assistant_models.media_items import Playlist, Track
10from music_assistant_models.media_items.provider_mapping import ProviderMapping
11
12from music_assistant.constants import CONF_VALUE_GLOBAL
13from music_assistant.controllers.player_queues import PlayerQueuesController
14from music_assistant.controllers.player_queues.autoplay import (
15 AUTOPLAY_BATCH_SIZE,
16 Autoplay,
17 AutoplayMode,
18)
19from music_assistant.controllers.player_queues.constants import (
20 CONF_AUTOPLAY_MODE,
21 CONF_AUTOPLAY_PLAYLIST,
22)
23
24if TYPE_CHECKING:
25 from music_assistant_models.player_queue import PlayerQueue
26
27
28def _track(item_id: str, *, provider: str = "library", available: bool = True) -> Track:
29 """Build a minimal Track for the given id/provider (availability via its provider mapping)."""
30 return Track(
31 item_id=item_id,
32 provider=provider,
33 name=f"Track {item_id}",
34 provider_mappings={
35 ProviderMapping(
36 item_id=item_id,
37 provider_domain=provider,
38 provider_instance=provider,
39 available=available,
40 )
41 },
42 )
43
44
45def _playlist(item_id: str = "pl1") -> Playlist:
46 """Build a minimal Playlist."""
47 return Playlist(
48 item_id=item_id,
49 provider="library",
50 name="My Playlist",
51 provider_mappings={
52 ProviderMapping(item_id=item_id, provider_domain="library", provider_instance="library")
53 },
54 )
55
56
57def _helper() -> tuple[Autoplay, MagicMock]:
58 """
59 Build an Autoplay and return it together with its mocked controller.
60
61 The returned MagicMock stands in for the owning controller; configure expectations and
62 make assertions through it (its attributes are untyped, so mock wiring stays simple).
63 """
64 queues = MagicMock()
65 return Autoplay(queues), queues
66
67
68def _queue(queues: MagicMock, *seeds: Track, queue_id: str = "q1") -> PlayerQueue:
69 """Build a queue stand-in with the given enqueued items, wired into the mock controller."""
70 queue = SimpleNamespace(
71 queue_id=queue_id, display_name="Queue", enqueued_media_items=list(seeds)
72 )
73 # enqueued items are server-side state; the helper reads them via queue_data()
74 queues.queue_data = MagicMock(return_value=queue)
75 return cast("PlayerQueue", queue)
76
77
78# --- resolve_mode ---
79
80
81def test_resolve_mode_reads_config() -> None:
82 """A stored mode value is parsed into the matching enum."""
83 helper, queues = _helper()
84 queues.mass.config.get_effective_player_queue_config_value = MagicMock(return_value="library")
85 assert helper.resolve_mode("q1") == AutoplayMode.LIBRARY
86
87
88def test_resolve_mode_defaults_to_auto() -> None:
89 """An unknown/blank value falls back to the AUTO mode."""
90 helper, queues = _helper()
91 queues.mass.config.get_effective_player_queue_config_value = MagicMock(return_value="bogus")
92 assert helper.resolve_mode("q1") == AutoplayMode.AUTO
93
94
95# --- get_library_tracks ---
96
97
98async def test_get_library_tracks_genre_then_topup() -> None:
99 """Genre-matched tracks come first, then a whole-library top-up, capped to the batch size."""
100 helper, queues = _helper()
101 queues.mass.config.get_raw_player_queue_config_value = MagicMock(return_value="library")
102 queues.mass.music.genres.get_genres_for_media_item = AsyncMock(
103 return_value=[SimpleNamespace(item_id="10")]
104 )
105 genre_tracks = [_track("g1"), _track("g2")]
106 topup_tracks = [_track(f"r{i}") for i in range(30)]
107 queues.mass.music.tracks.library_items = AsyncMock(side_effect=[genre_tracks, topup_tracks])
108
109 result = await helper.get_library_tracks(_queue(queues, _track("seed")), exclude=set())
110
111 # genre query used random_play_count ordering and the genre id we collected
112 first_call = queues.mass.music.tracks.library_items.await_args_list[0]
113 assert first_call.kwargs["order_by"] == "random_play_count"
114 assert first_call.kwargs["genre"] == [10]
115 # genre matches lead, batch size is respected
116 assert result[:2] == genre_tracks
117 assert len(result) == AUTOPLAY_BATCH_SIZE
118
119
120async def test_get_library_tracks_without_genre_uses_random_mix() -> None:
121 """With no genre signal, only the whole-library random mix is queried."""
122 helper, queues = _helper()
123 queues.mass.music.genres.get_genres_for_media_item = AsyncMock(return_value=[])
124 queues.mass.music.tracks.library_items = AsyncMock(return_value=[_track("r1"), _track("r2")])
125
126 result = await helper.get_library_tracks(_queue(queues, _track("seed")), exclude=set())
127
128 queues.mass.music.tracks.library_items.assert_awaited_once()
129 assert "genre" not in queues.mass.music.tracks.library_items.await_args.kwargs
130 assert {t.item_id for t in result} == {"r1", "r2"}
131
132
133async def test_get_library_tracks_excludes_and_dedupes() -> None:
134 """Already-queued, duplicate and unavailable tracks are filtered out."""
135 helper, queues = _helper()
136 queues.mass.music.genres.get_genres_for_media_item = AsyncMock(return_value=[])
137 dupe = _track("dupe")
138 excluded = _track("excluded")
139 gone = _track("gone", available=False)
140 queues.mass.music.tracks.library_items = AsyncMock(
141 return_value=[dupe, dupe, excluded, gone, _track("fresh")]
142 )
143
144 result = await helper.get_library_tracks(_queue(queues), exclude={excluded})
145
146 assert [t.item_id for t in result] == ["dupe", "fresh"]
147
148
149async def test_get_library_tracks_tops_up_past_unavailable_genre_matches() -> None:
150 """A genre batch that looks full but is all unavailable still triggers the library top-up."""
151 helper, queues = _helper()
152 queues.mass.music.genres.get_genres_for_media_item = AsyncMock(
153 return_value=[SimpleNamespace(item_id="1")]
154 )
155 genre_tracks = [_track(f"g{i}", available=False) for i in range(AUTOPLAY_BATCH_SIZE)]
156 topup_tracks = [_track(f"r{i}") for i in range(AUTOPLAY_BATCH_SIZE)]
157 queues.mass.music.tracks.library_items = AsyncMock(side_effect=[genre_tracks, topup_tracks])
158
159 result = await helper.get_library_tracks(_queue(queues, _track("seed")), exclude=set())
160
161 assert queues.mass.music.tracks.library_items.await_count == 2
162 assert len(result) == AUTOPLAY_BATCH_SIZE
163 assert all(track.available for track in result)
164
165
166# --- get_playlist_tracks ---
167
168
169async def test_get_playlist_tracks_returns_filtered_batch() -> None:
170 """The configured playlist's tracks are returned minus the excluded ones."""
171 helper, queues = _helper()
172 # a per-queue playlist override (mode set to 'playlist' on the queue)
173 queues.mass.config.get_raw_player_queue_config_value = MagicMock(
174 side_effect=lambda _qid, key, default=None: {
175 CONF_AUTOPLAY_MODE: AutoplayMode.PLAYLIST.value,
176 CONF_AUTOPLAY_PLAYLIST: "library://playlist/pl1",
177 }.get(key, default)
178 )
179 queues.mass.music.get_item_by_uri = AsyncMock(return_value=_playlist())
180 excluded = _track("t2")
181 queues.get_playlist_tracks = AsyncMock(return_value=[_track("t1"), excluded, _track("t3")])
182
183 result = await helper.get_playlist_tracks(_queue(queues), exclude={excluded})
184
185 assert {t.item_id for t in result} == {"t1", "t3"}
186
187
188async def test_get_playlist_tracks_without_config_returns_empty() -> None:
189 """Playlist mode with no configured playlist yields nothing (and does not crash)."""
190 helper, queues = _helper()
191 queues.mass.config.get_raw_player_queue_config_value = MagicMock(
192 side_effect=lambda _qid, key, default=None: {
193 CONF_AUTOPLAY_MODE: AutoplayMode.PLAYLIST.value
194 }.get(key, default)
195 )
196
197 assert await helper.get_playlist_tracks(_queue(queues), exclude=set()) == []
198
199
200async def test_get_playlist_tracks_global_mode_uses_global_playlist() -> None:
201 """A queue following the global mode uses the global playlist, not a stale per-queue one."""
202 helper, queues = _helper()
203 # queue follows the global mode, yet a stale per-queue playlist is still stored
204 queues.mass.config.get_raw_player_queue_config_value = MagicMock(
205 side_effect=lambda _qid, key, default=None: {
206 CONF_AUTOPLAY_MODE: CONF_VALUE_GLOBAL,
207 CONF_AUTOPLAY_PLAYLIST: "library://playlist/stale",
208 }.get(key, default)
209 )
210 queues.mass.config.get_raw_core_config_value = MagicMock(
211 return_value="library://playlist/global"
212 )
213 captured: dict[str, str] = {}
214
215 async def _get_item(uri: str) -> Playlist:
216 captured["uri"] = uri
217 return _playlist()
218
219 queues.mass.music.get_item_by_uri = AsyncMock(side_effect=_get_item)
220 queues.get_playlist_tracks = AsyncMock(return_value=[_track("t1")])
221
222 await helper.get_playlist_tracks(_queue(queues), exclude=set())
223
224 assert captured["uri"] == "library://playlist/global"
225
226
227# --- config entries ---
228
229
230def test_get_queue_config_entries_categories_and_dependencies() -> None:
231 """Per-queue entries land in their categories, default to 'global' and offer a global option."""
232 fake = MagicMock()
233 fake.mass.music.providers = []
234 fake.mass.streams.smart_fades_available = False
235
236 entries = PlayerQueuesController.get_queue_config_entries(cast("PlayerQueuesController", fake))
237 by_key = {entry.key: entry for entry in entries}
238
239 assert by_key["autoplay_mode"].category == "autoplay"
240 assert by_key["autoplay_mode"].default_value == CONF_VALUE_GLOBAL
241 assert by_key["autoplay_playlist"].depends_on == "autoplay_mode"
242 assert by_key["autoplay_playlist"].depends_on_value == AutoplayMode.PLAYLIST.value
243 assert by_key["crossfade_mode"].category == "crossfade"
244 assert by_key["crossfade_mode"].default_value == CONF_VALUE_GLOBAL
245 assert by_key["volume_normalization"].category == "audio"
246 assert by_key["volume_normalization"].default_value == CONF_VALUE_GLOBAL
247 # crossfade duration and the recency windows are global-only, so they are not per-queue entries
248 assert "crossfade_duration" not in by_key
249 assert "smart_shuffle_song_recency" not in by_key
250 # every headline setting offers a 'global' option that follows the queue-controller default
251 for key in ("autoplay_mode", "crossfade_mode", "volume_normalization", "smart_shuffle_enabled"):
252 assert CONF_VALUE_GLOBAL in {opt.value for opt in by_key[key].options}
253 # the 'similar' option is disabled when no provider can supply similar tracks
254 similar_option = next(
255 opt for opt in by_key["autoplay_mode"].options if opt.value == AutoplayMode.SIMILAR.value
256 )
257 assert similar_option.disabled is True
258