/
/
/
1"""Tests for the player queues controller helpers."""
2
3from __future__ import annotations
4
5import json
6from typing import TYPE_CHECKING, Any, Self, cast
7
8import pytest
9from music_assistant_models.enums import ImageType, MediaType
10from music_assistant_models.media_items import (
11 Album,
12 Artist,
13 ItemMapping,
14 MediaItemImage,
15 MediaItemMetadata,
16 Playlist,
17 Radio,
18 Track,
19)
20from music_assistant_models.media_items.provider_mapping import ProviderMapping
21from music_assistant_models.player_queue import PlayerQueue
22from music_assistant_models.queue_item import QueueItem
23from music_assistant_models.unique_list import UniqueList
24
25from music_assistant.constants import ATTR_PLAY_ACTION_IN_PROGRESS
26from music_assistant.controllers.player_queues.helpers import (
27 build_queue_item,
28 find_dynamic_source,
29 get_current_playback_speed,
30 handle_play_action,
31 has_dynamic_source,
32 is_dynamic_source,
33 space_by_artist,
34)
35from music_assistant.controllers.player_queues.state import PlayerQueueData
36
37if TYPE_CHECKING:
38 from music_assistant_models.media_items import MediaItemType
39
40 from music_assistant.controllers.player_queues.controller import PlayerQueuesController
41
42_PROVIDER_MAPPINGS = {
43 ProviderMapping(item_id="x", provider_domain="test", provider_instance="test")
44}
45
46
47def _playlist(*, is_dynamic: bool, name: str = "PL") -> Playlist:
48 return Playlist(
49 item_id=name.lower(),
50 provider="test",
51 name=name,
52 provider_mappings=_PROVIDER_MAPPINGS,
53 is_dynamic=is_dynamic,
54 )
55
56
57def _radio(*, is_dynamic: bool, name: str = "R") -> Radio:
58 return Radio(
59 item_id=name.lower(),
60 provider="test",
61 name=name,
62 provider_mappings=_PROVIDER_MAPPINGS,
63 is_dynamic=is_dynamic,
64 )
65
66
67def _track(name: str) -> Track:
68 return Track(
69 item_id=name.lower(),
70 provider="test",
71 name=name,
72 duration=100,
73 artists=UniqueList(),
74 provider_mappings=_PROVIDER_MAPPINGS,
75 )
76
77
78def _queue_item(name: str, *, item_id: str | None = None, **extra_attributes: Any) -> QueueItem:
79 return QueueItem(
80 queue_id="q1",
81 queue_item_id=item_id or name.lower(),
82 name=name,
83 duration=100,
84 extra_attributes=extra_attributes,
85 )
86
87
88def _queue() -> PlayerQueue:
89 return PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
90
91
92def _queue_data(
93 *,
94 source_items: list[MediaItemType] | None = None,
95 enqueued_media_items: list[MediaItemType] | None = None,
96) -> PlayerQueueData:
97 queue_data = PlayerQueueData(queue=_queue())
98 queue_data.source_items = source_items or []
99 queue_data.enqueued_media_items = enqueued_media_items or []
100 return queue_data
101
102
103class TestHasDynamicSource:
104 """Tests for has_dynamic_source."""
105
106 def test_single_dynamic_playlist(self) -> None:
107 """A single dynamic playlist puts the queue in dynamic mode."""
108 assert has_dynamic_source([_playlist(is_dynamic=True)]) is True
109
110 def test_single_non_dynamic_playlist(self) -> None:
111 """A single non-dynamic playlist is not a dynamic source."""
112 assert has_dynamic_source([_playlist(is_dynamic=False)]) is False
113
114 def test_empty_source(self) -> None:
115 """An empty source list is not dynamic."""
116 assert has_dynamic_source([]) is False
117
118 def test_multiple_dynamic_playlists(self) -> None:
119 """Any dynamic playlist among the sources puts the queue in dynamic mode."""
120 source: list[MediaItemType] = [
121 _playlist(is_dynamic=True, name="A"),
122 _playlist(is_dynamic=True, name="B"),
123 ]
124 assert has_dynamic_source(source) is True
125
126 def test_dynamic_playlist_mixed_with_finite_source(self) -> None:
127 """A dynamic playlist mixed with a finite source still counts as dynamic."""
128 source: list[MediaItemType] = [_track("Song"), _playlist(is_dynamic=True)]
129 assert has_dynamic_source(source) is True
130
131 def test_non_playlist_item(self) -> None:
132 """A non-playlist media item is not a dynamic source."""
133 assert has_dynamic_source([_track("Song")]) is False
134
135
136class TestIsDynamicSource:
137 """Tests for is_dynamic_source."""
138
139 def test_dynamic_playlist(self) -> None:
140 """A dynamic playlist supplies its own on-demand feed."""
141 assert is_dynamic_source(_playlist(is_dynamic=True)) is True
142
143 def test_non_dynamic_playlist(self) -> None:
144 """A non-dynamic playlist is not a dynamic source."""
145 assert is_dynamic_source(_playlist(is_dynamic=False)) is False
146
147 def test_dynamic_radio(self) -> None:
148 """A dynamic radio station supplies its own on-demand feed."""
149 assert is_dynamic_source(_radio(is_dynamic=True)) is True
150
151 def test_non_dynamic_radio(self) -> None:
152 """A non-dynamic (live-stream) radio is not a dynamic source."""
153 assert is_dynamic_source(_radio(is_dynamic=False)) is False
154
155 def test_track(self) -> None:
156 """A track is never a dynamic source."""
157 assert is_dynamic_source(_track("Song")) is False
158
159
160class TestFindDynamicSource:
161 """Tests for find_dynamic_source."""
162
163 def test_dynamic_radio_source(self) -> None:
164 """A dynamic radio station is found, so an idle queue can refill from it."""
165 station = _radio(is_dynamic=True)
166 queue_data = _queue_data(source_items=[station])
167 assert find_dynamic_source(queue_data) is station
168
169 def test_dynamic_playlist_source(self) -> None:
170 """A dynamic playlist is found the same way."""
171 playlist = _playlist(is_dynamic=True)
172 queue_data = _queue_data(source_items=[playlist])
173 assert find_dynamic_source(queue_data) is playlist
174
175 def test_prefers_the_last_added_source(self) -> None:
176 """The most recently added dynamic source wins."""
177 first = _playlist(is_dynamic=True, name="A")
178 last = _radio(is_dynamic=True, name="B")
179 queue_data = _queue_data(source_items=[first, _track("Song"), last])
180 assert find_dynamic_source(queue_data) is last
181
182 def test_falls_back_to_enqueued_items(self) -> None:
183 """A queue without dynamic sources falls back to what was enqueued on it."""
184 station = _radio(is_dynamic=True)
185 queue_data = _queue_data(source_items=[_track("Song")], enqueued_media_items=[station])
186 assert find_dynamic_source(queue_data) is station
187
188 def test_no_dynamic_source(self) -> None:
189 """A queue of finite items has nothing to refill from."""
190 queue_data = _queue_data(
191 source_items=[_playlist(is_dynamic=False)], enqueued_media_items=[_track("Song")]
192 )
193 assert find_dynamic_source(queue_data) is None
194
195
196class TestGetCurrentPlaybackSpeed:
197 """Tests for get_current_playback_speed."""
198
199 def test_no_current_item(self) -> None:
200 """Defaults to 1.0 when the queue has no current item."""
201 assert get_current_playback_speed(_queue()) == 1.0
202
203 def test_speed_from_extra_attributes(self) -> None:
204 """Reads the playback_speed from the current item's extra attributes."""
205 queue = _queue()
206 queue.current_item = _queue_item("Song", playback_speed=1.5)
207 assert get_current_playback_speed(queue) == 1.5
208
209 def test_unset_speed_defaults_to_one(self) -> None:
210 """Defaults to 1.0 when the current item has no playback_speed set."""
211 queue = _queue()
212 queue.current_item = _queue_item("Song")
213 assert get_current_playback_speed(queue) == 1.0
214
215 def test_zero_speed_falls_back_to_one(self) -> None:
216 """A falsy playback_speed falls back to 1.0."""
217 queue = _queue()
218 queue.current_item = _queue_item("Song", playback_speed=0)
219 assert get_current_playback_speed(queue) == 1.0
220
221
222class _FakeLock:
223 """No-op re-entrant async context manager standing in for the player lock."""
224
225 async def __aenter__(self) -> Self:
226 return self
227
228 async def __aexit__(self, *exc_info: object) -> bool:
229 return False
230
231
232class _FakePlayer:
233 def __init__(self, player_id: str) -> None:
234 self.player_id = player_id
235
236
237class _FakePlayers:
238 def __init__(self, player_ids: set[str]) -> None:
239 self._players = {player_id: _FakePlayer(player_id) for player_id in player_ids}
240
241 def get_player_lock(self, queue_id: str, purpose: object) -> _FakeLock:
242 return _FakeLock()
243
244 def get_player(self, player_id: str) -> _FakePlayer | None:
245 return self._players.get(player_id)
246
247
248class _FakeMass:
249 def __init__(self, player_ids: set[str]) -> None:
250 self.players = _FakePlayers(player_ids)
251
252
253class _FakeQueue:
254 def __init__(self) -> None:
255 self.extra_attributes: dict[str, Any] = {}
256
257
258class _FakeController:
259 """Minimal stand-in exposing only what handle_play_action touches."""
260
261 def __init__(self, queues: dict[str, _FakeQueue], with_players: bool = True) -> None:
262 self.mass = _FakeMass(set(queues) if with_players else set())
263 self._queue_data = {
264 queue_id: PlayerQueueData(queue=cast("PlayerQueue", queue))
265 for queue_id, queue in queues.items()
266 }
267 self.calls: list[str] = []
268
269 def signal_update(self, queue_id: str, items_changed: bool = False) -> None:
270 self.calls.append(f"signal:{queue_id}")
271
272 def on_player_update(
273 self, player: _FakePlayer, changed_values: dict[str, tuple[Any, Any]]
274 ) -> None:
275 self.calls.append(f"refresh:{player.player_id}")
276
277
278def _flag_value(ctrl: PlayerQueuesController, queue_id: str) -> bool:
279 """Return the current in-progress flag for a queue (False if unknown)."""
280 data = ctrl._queue_data.get(queue_id)
281 if data is None:
282 return False
283 return bool(data.queue.extra_attributes.get(ATTR_PLAY_ACTION_IN_PROGRESS, False))
284
285
286@handle_play_action
287async def _flag_during(self: PlayerQueuesController, queue_id: str, sink: list[bool]) -> str:
288 """Record the in-progress flag while running."""
289 sink.append(_flag_value(self, queue_id))
290 return "done"
291
292
293@handle_play_action
294async def _nested(self: PlayerQueuesController, queue_id: str, sink: list[bool]) -> str:
295 """Invoke a nested decorated action on the same queue."""
296 await _flag_during(self, queue_id, sink)
297 sink.append(_flag_value(self, queue_id))
298 return "outer"
299
300
301@handle_play_action
302async def _boom(self: PlayerQueuesController, queue_id: str) -> str: # noqa: ARG001
303 """Raise to exercise the cleanup path."""
304 raise RuntimeError("boom")
305
306
307class TestHandlePlayAction:
308 """Tests for the handle_play_action decorator."""
309
310 async def test_runs_without_known_queue(self) -> None:
311 """When the queue is unknown the action still runs and nothing is signalled."""
312 ctrl = _FakeController({})
313 sink: list[bool] = []
314 assert await _flag_during(cast("PlayerQueuesController", ctrl), "missing", sink) == "done"
315 assert sink == [False]
316 assert ctrl.calls == []
317
318 async def test_sets_and_clears_flag(self) -> None:
319 """The in-progress flag is set during the action and cleared afterwards."""
320 queue = _FakeQueue()
321 ctrl = _FakeController({"q1": queue})
322 sink: list[bool] = []
323 assert await _flag_during(cast("PlayerQueuesController", ctrl), "q1", sink) == "done"
324 assert sink == [True]
325 assert queue.extra_attributes[ATTR_PLAY_ACTION_IN_PROGRESS] is False
326 # signalled once on entry and once on exit, with the queue recalculated
327 # from the player in between so the exit signal carries the result
328 assert ctrl.calls == ["signal:q1", "refresh:q1", "signal:q1"]
329 assert ctrl._queue_data["q1"].play_action_refcount == 0
330
331 async def test_clears_flag_without_player(self) -> None:
332 """A queue without a registered player still clears and signals."""
333 queue = _FakeQueue()
334 ctrl = _FakeController({"q1": queue}, with_players=False)
335 sink: list[bool] = []
336 assert await _flag_during(cast("PlayerQueuesController", ctrl), "q1", sink) == "done"
337 assert queue.extra_attributes[ATTR_PLAY_ACTION_IN_PROGRESS] is False
338 assert ctrl.calls == ["signal:q1", "signal:q1"]
339
340 async def test_nested_actions_refcount(self) -> None:
341 """Nested actions keep the flag set until the outermost one finishes."""
342 queue = _FakeQueue()
343 ctrl = _FakeController({"q1": queue})
344 sink: list[bool] = []
345 assert await _nested(cast("PlayerQueuesController", ctrl), "q1", sink) == "outer"
346 # flag stayed True through the inner action and after it returned
347 assert sink == [True, True]
348 assert queue.extra_attributes[ATTR_PLAY_ACTION_IN_PROGRESS] is False
349 # only the outermost entry/exit signal an update (inner sees it already in progress)
350 assert ctrl.calls == ["signal:q1", "refresh:q1", "signal:q1"]
351 assert ctrl._queue_data["q1"].play_action_refcount == 0
352
353 async def test_flag_cleared_on_exception(self) -> None:
354 """The flag is cleared even when the action raises."""
355 queue = _FakeQueue()
356 ctrl = _FakeController({"q1": queue})
357 with pytest.raises(RuntimeError, match="boom"):
358 await _boom(cast("PlayerQueuesController", ctrl), "q1")
359 assert queue.extra_attributes[ATTR_PLAY_ACTION_IN_PROGRESS] is False
360 assert ctrl._queue_data["q1"].play_action_refcount == 0
361
362
363class TestSpaceByArtist:
364 """Tests for space_by_artist."""
365
366 def test_separates_adjacent_shared_artist(self) -> None:
367 """Adjacent entries sharing an artist are pulled apart (by intersection), dropping nothing."""
368 sets = [{"a"}, {"a", "b"}, {"b"}, {"c"}]
369 spaced = [sets[index] for index in space_by_artist(sets)]
370 assert sorted(spaced, key=sorted) == sorted(sets, key=sorted)
371 assert all(not (spaced[i] & spaced[i + 1]) for i in range(len(spaced) - 1))
372
373 def test_honours_preceding_seam(self) -> None:
374 """The first entry shares no artist with the preceding (seam) set."""
375 sets = [{"a", "b"}, {"c"}, {"d"}]
376 order = space_by_artist(sets, preceding={"a"})
377 assert not (sets[order[0]] & {"a"})
378
379 def test_identity_when_already_clear(self) -> None:
380 """With no clashes and no seam, the order is left untouched."""
381 assert space_by_artist([{"a"}, {"b"}, {"c"}]) == [0, 1, 2]
382
383 def test_empty_input(self) -> None:
384 """An empty input yields an empty order."""
385 assert space_by_artist([]) == []
386
387
388def _thumb(path: str = "http://img/t.jpg") -> MediaItemImage:
389 return MediaItemImage(
390 type=ImageType.THUMB, path=path, provider="test", remotely_accessible=True
391 )
392
393
394def _heavy_metadata() -> MediaItemMetadata:
395 """Build metadata resembling a fully enriched item (the bulk of a queue item's size)."""
396 return MediaItemMetadata(
397 description="A long track description. " * 20,
398 review="An even longer critical review. " * 20,
399 lyrics="\n".join(f"Lyrics line {index}" for index in range(60)),
400 lrc_lyrics="\n".join(f"[00:{index:02d}.00] line {index}" for index in range(60)),
401 images=UniqueList([_thumb()]),
402 genres={"rock", "indie"},
403 )
404
405
406def _heavy_track() -> Track:
407 return Track(
408 item_id="t1",
409 provider="test",
410 name="Song",
411 duration=210,
412 artists=UniqueList(
413 [
414 Artist(
415 item_id="a1",
416 provider="test",
417 name="Artist",
418 metadata=_heavy_metadata(),
419 provider_mappings=_PROVIDER_MAPPINGS,
420 )
421 ]
422 ),
423 album=Album(
424 item_id="al1",
425 provider="test",
426 name="Album",
427 metadata=_heavy_metadata(),
428 provider_mappings=_PROVIDER_MAPPINGS,
429 ),
430 metadata=_heavy_metadata(),
431 provider_mappings=_PROVIDER_MAPPINGS,
432 )
433
434
435def _heavy_radio() -> Radio:
436 return Radio(
437 item_id="r1",
438 provider="test",
439 name="Radio One",
440 metadata=_heavy_metadata(),
441 provider_mappings=_PROVIDER_MAPPINGS,
442 )
443
444
445class TestBuildQueueItem:
446 """Tests for build_queue_item."""
447
448 def test_slims_track_metadata(self) -> None:
449 """A track's heavy metadata is dropped while playback-relevant fields are kept."""
450 item = build_queue_item("q1", _heavy_track())
451 assert isinstance(item.media_item, Track)
452 # heavy metadata is dropped
453 assert item.media_item.metadata == MediaItemMetadata()
454 # provider mappings are kept (needed for stream resolution / failover)
455 assert item.media_item.provider_mappings == _PROVIDER_MAPPINGS
456 # the identifiers used to re-hydrate on promotion are kept
457 assert item.media_item.item_id == "t1"
458 assert item.media_item.provider == "test"
459 assert item.media_item.media_type is MediaType.TRACK
460 # artwork survives on the top-level image and artists/album stay slimmed to mappings
461 assert item.image is not None
462 assert item.image.type is ImageType.THUMB
463 assert all(isinstance(artist, ItemMapping) for artist in item.media_item.artists)
464 assert isinstance(item.media_item.album, ItemMapping)
465 # name/duration for compact list rows are kept
466 assert item.name == "Artist - Song"
467 assert item.duration == 210
468
469 def test_reduces_serialized_size(self) -> None:
470 """Slimming a track queue item substantially reduces its serialized size."""
471 fat = len(json.dumps(QueueItem.from_media_item("q1", _heavy_track()).to_dict()))
472 slim = len(json.dumps(build_queue_item("q1", _heavy_track()).to_dict()))
473 assert slim < fat * 0.6
474
475 def test_leaves_non_track_untouched(self) -> None:
476 """Non-track items (e.g. radio) keep their metadata as they are not re-hydrated."""
477 item = build_queue_item("q1", _heavy_radio())
478 assert isinstance(item.media_item, Radio)
479 assert item.media_item.metadata.description
480 assert item.media_item.metadata.images
481
482 def test_cache_roundtrip_preserves_slim_shape(self) -> None:
483 """A slim track queue item round-trips through the persisted cache unchanged."""
484 restored = QueueItem.from_cache(build_queue_item("q1", _heavy_track()).to_cache())
485 assert isinstance(restored.media_item, Track)
486 assert restored.media_item.metadata == MediaItemMetadata()
487 assert restored.media_item.provider_mappings == _PROVIDER_MAPPINGS
488 assert restored.media_item.item_id == "t1"
489 assert restored.image is not None
490 assert restored.image.type is ImageType.THUMB
491