music-assistant-server

1"""Tests for the server-side PlayerQueueData container and its cache round-trip."""
2
3from __future__ import annotations
4
5import pytest
6from music_assistant_models.enums import AlbumType, RepeatMode
7from music_assistant_models.media_items import (
8    Album,
9    ItemMapping,
10    Playlist,
11    ProviderMapping,
12    Track,
13)
14from music_assistant_models.player_queue import PlayerQueue
15from music_assistant_models.queue_item import QueueItem
16
17from music_assistant.controllers.player_queues.constants import CACHE_FORMAT_VERSION
18from music_assistant.controllers.player_queues.state import PlayerQueueData
19
20
21def _track(item_id: str) -> Track:
22    """Build a minimal library Track."""
23    return Track(
24        item_id=item_id,
25        provider="library",
26        name=f"Track {item_id}",
27        provider_mappings={
28            ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
29        },
30    )
31
32
33def _album(item_id: str) -> Album:
34    """Build a minimal library Album."""
35    return Album(
36        item_id=item_id,
37        provider="library",
38        name=f"Album {item_id}",
39        provider_mappings={
40            ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
41        },
42        album_type=AlbumType.ALBUM,
43    )
44
45
46def _dynamic_playlist(item_id: str = "radio1") -> Playlist:
47    """Build a dynamic (radio) Playlist."""
48    playlist = Playlist(
49        item_id=item_id,
50        provider="radio_playlist",
51        name="Radio",
52        provider_mappings={
53            ProviderMapping(
54                item_id=item_id,
55                provider_domain="radio_playlist",
56                provider_instance="radio_playlist",
57            )
58        },
59    )
60    playlist.is_dynamic = True
61    return playlist
62
63
64def _queue(**kwargs: object) -> PlayerQueue:
65    """Build a PlayerQueue with the given overrides."""
66    return PlayerQueue(
67        queue_id="q1",
68        active=True,
69        display_name="Queue",
70        available=True,
71        items=0,
72        **kwargs,  # type: ignore[arg-type]
73    )
74
75
76def _data_with_dynamic_source() -> PlayerQueueData:
77    """Build a PlayerQueueData for a queue playing a dynamic radio playlist plus one queued track."""
78    playlist = _dynamic_playlist()
79    album = _album("a1")
80    queue = _queue(is_dynamic=True, sources=[ItemMapping.from_item(playlist)])
81    return PlayerQueueData(
82        queue=queue,
83        items=[QueueItem.from_media_item("q1", _track("t1"))],
84        source_items=[playlist],
85        enqueued_media_items=[playlist, album],
86        credited_albums={album},
87        userid="user-1",
88        # runtime-only fields set to non-defaults to prove they do not survive the round-trip
89        transitioning=True,
90        play_action_refcount=3,
91        last_counted_play="t1",
92        flow_buffer_completed="sess-1",
93    )
94
95
96def test_cache_round_trip_restores_queue_items_and_sources() -> None:
97    """to_cache/from_cache restores the queue, items and source items; runtime fields reset."""
98    data = _data_with_dynamic_source()
99
100    restored = PlayerQueueData.from_cache(data.to_cache(), data.items_to_cache())
101
102    # the persisted bits survive
103    assert restored.queue.queue_id == "q1"
104    assert [item.queue_item_id for item in restored.items] == [
105        item.queue_item_id for item in data.items
106    ]
107    assert [item.uri for item in restored.source_items] == [item.uri for item in data.source_items]
108    # the server-only persisted fields survive too
109    assert [item.uri for item in restored.enqueued_media_items] == [
110        item.uri for item in data.enqueued_media_items
111    ]
112    assert restored.userid == "user-1"
113    # a credited album survives, so a restart does not re-credit the same enqueue
114    assert {item.uri for item in restored.credited_albums} == {
115        item.uri for item in data.credited_albums
116    }
117    # is_dynamic is recomputed from the restored source items (a dynamic playlist is present)
118    assert restored.queue.is_dynamic is True
119    # runtime-only fields are reset to their defaults, never persisted
120    assert restored.transitioning is False
121    assert restored.play_action_refcount == 0
122    assert restored.last_counted_play is None
123    assert restored.flow_buffer_completed is None
124
125
126def test_cache_round_trip_non_dynamic_has_no_sources() -> None:
127    """A queue with no dynamic source restores with is_dynamic False and empty sources."""
128    queue = _queue()
129    data = PlayerQueueData(
130        queue=queue,
131        items=[QueueItem.from_media_item("q1", _track("t1"))],
132        source_items=[],
133        enqueued_media_items=[_track("seed")],
134    )
135
136    restored = PlayerQueueData.from_cache(data.to_cache(), data.items_to_cache())
137
138    assert restored.source_items == []
139    assert restored.queue.is_dynamic is False
140
141
142def test_from_cache_legacy_rebuilds_source_items_from_sources() -> None:
143    """A legacy cache without persisted source_items rebuilds them from sources + enqueued items."""
144    data = _data_with_dynamic_source()
145    state = data.to_cache()
146    # simulate a pre-source_items cache entry: only the wire `sources` ItemMappings persisted
147    state.pop("source_items")
148
149    restored = PlayerQueueData.from_cache(state, data.items_to_cache())
150
151    # the full source item is recovered by matching `sources` against the enqueued media items
152    assert [item.uri for item in restored.source_items] == [item.uri for item in data.source_items]
153    assert restored.queue.is_dynamic is True
154
155
156def test_cache_round_trip_restores_settings() -> None:
157    """The queue settings (shuffle/repeat/crossfade/autoplay) survive a cache round-trip."""
158    queue = _queue(
159        shuffle_enabled=True,
160        repeat_mode=RepeatMode.ALL,
161        crossfade_enabled=True,
162        autoplay_enabled=True,
163    )
164    data = PlayerQueueData(queue=queue)
165
166    restored = PlayerQueueData.from_cache(data.to_cache(), data.items_to_cache())
167
168    assert restored.queue.shuffle_enabled is True
169    assert restored.queue.repeat_mode == RepeatMode.ALL
170    assert restored.queue.crossfade_enabled is True
171    assert restored.queue.autoplay_enabled is True
172
173
174def test_cache_round_trip_restores_toggle_overrides() -> None:
175    """A pinned autoplay/crossfade override survives a cache round-trip."""
176    queue = _queue()
177    data = PlayerQueueData(queue=queue, autoplay_override=False, crossfade_override=True)
178
179    restored = PlayerQueueData.from_cache(data.to_cache(), data.items_to_cache())
180
181    assert restored.autoplay_override is False
182    assert restored.crossfade_override is True
183
184
185def test_from_cache_legacy_without_toggle_overrides_yields_none() -> None:
186    """A legacy cache with no override keys restores as None, i.e. follow the global default."""
187    queue = _queue()
188    data = PlayerQueueData(queue=queue)
189    state = data.to_cache()
190    del state["autoplay_override"]
191    del state["crossfade_override"]
192
193    restored = PlayerQueueData.from_cache(state, data.items_to_cache())
194
195    assert restored.autoplay_override is None
196    assert restored.crossfade_override is None
197
198
199def test_from_cache_keeps_settings_when_item_unreadable() -> None:
200    """One unreadable queue item is skipped without losing the other items or the settings."""
201    queue = _queue(shuffle_enabled=True, crossfade_enabled=True)
202    data = PlayerQueueData(queue=queue, items=[QueueItem.from_media_item("q1", _track("t1"))])
203    items_data = [*data.items_to_cache(), {"sort_index": 0}]  # 2nd entry misses required fields
204
205    restored = PlayerQueueData.from_cache(data.to_cache(), items_data)
206
207    # the good item is kept, the unreadable one is dropped
208    assert [item.queue_item_id for item in restored.items] == [
209        item.queue_item_id for item in data.items
210    ]
211    # the wire-facing count matches the items actually restored (not the cached count)
212    assert restored.queue.items == len(restored.items) == 1
213    # and the settings are untouched by the item failure
214    assert restored.queue.shuffle_enabled is True
215    assert restored.queue.crossfade_enabled is True
216
217
218def test_from_cache_keeps_settings_when_source_unreadable() -> None:
219    """An unreadable source mapping is skipped instead of aborting the whole restore."""
220    queue = _queue(shuffle_enabled=True, sources=[ItemMapping.from_item(_dynamic_playlist())])
221    data = PlayerQueueData(queue=queue)
222    state = data.to_cache()
223    # corrupt the persisted source mappings so their deserialization raises
224    state["queue"]["sources"] = [{"not": "a mapping"}]
225
226    restored = PlayerQueueData.from_cache(state, data.items_to_cache())
227
228    assert restored.queue.sources == []
229    assert restored.queue.shuffle_enabled is True
230
231
232def test_from_cache_keeps_settings_when_enqueued_media_unreadable() -> None:
233    """Unreadable enqueued media clears only the media payload, never the queue settings."""
234    queue = _queue(shuffle_enabled=True, crossfade_enabled=True)
235    data = PlayerQueueData(queue=queue, enqueued_media_items=[_track("seed")])
236    state = data.to_cache()
237    # corrupt the persisted enqueued media so its deserialization raises
238    state["enqueued_media_items"] = [{"not": "a media item"}]
239
240    restored = PlayerQueueData.from_cache(state, data.items_to_cache())
241
242    assert restored.enqueued_media_items == []
243    assert restored.queue.shuffle_enabled is True
244    assert restored.queue.crossfade_enabled is True
245
246
247def test_from_cache_incompatible_version_raises() -> None:
248    """A cache written with an incompatible format version is rejected rather than misread."""
249    data = _data_with_dynamic_source()
250    state = data.to_cache()
251    state["cache_format_version"] = CACHE_FORMAT_VERSION + 1
252
253    with pytest.raises(ValueError, match="incompatible queue cache format"):
254        PlayerQueueData.from_cache(state, data.items_to_cache())
255
256
257def test_from_cache_reads_legacy_flat_layout() -> None:
258    """A pre-refactor cache (wire fields at the top level, no envelope) is still restored."""
259    data = _data_with_dynamic_source()
260    nested = data.to_cache()
261    # rebuild the old flat layout: wire fields at the top level, server-state keys alongside, and
262    # no cache_format_version (caches written before the refactor had no version stamp)
263    flat = {
264        **nested["queue"],
265        "enqueued_media_items": nested["enqueued_media_items"],
266        "source_items": nested["source_items"],
267        "userid": nested["userid"],
268    }
269
270    restored = PlayerQueueData.from_cache(flat, data.items_to_cache())
271
272    assert restored.queue.is_dynamic is True
273    assert [item.uri for item in restored.source_items] == [item.uri for item in data.source_items]
274    assert [item.uri for item in restored.enqueued_media_items] == [
275        item.uri for item in data.enqueued_media_items
276    ]
277    assert restored.userid == "user-1"
278
279
280def test_cache_significant_ignores_playback_progress() -> None:
281    """Change detection ignores volatile playback-progress fields but catches real changes."""
282    data = _data_with_dynamic_source()
283    state = data.to_cache()
284    # only elapsed time / playback speed advanced -> not a persist-worthy change
285    progressed = {
286        **state,
287        "queue": {**state["queue"], "elapsed_time": 999.0, "playback_speed": 2.0},
288    }
289    assert PlayerQueueData.cache_significant(progressed) == PlayerQueueData.cache_significant(state)
290    # a real settings change -> detected
291    shuffled = {
292        **state,
293        "queue": {**state["queue"], "shuffle_enabled": not state["queue"]["shuffle_enabled"]},
294    }
295    assert PlayerQueueData.cache_significant(shuffled) != PlayerQueueData.cache_significant(state)
296