music-assistant-server

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