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_from_cache_keeps_settings_when_item_unreadable() -> None:
175    """One unreadable queue item is skipped without losing the other items or the settings."""
176    queue = _queue(shuffle_enabled=True, crossfade_enabled=True)
177    data = PlayerQueueData(queue=queue, items=[QueueItem.from_media_item("q1", _track("t1"))])
178    items_data = [*data.items_to_cache(), {"sort_index": 0}]  # 2nd entry misses required fields
179
180    restored = PlayerQueueData.from_cache(data.to_cache(), items_data)
181
182    # the good item is kept, the unreadable one is dropped
183    assert [item.queue_item_id for item in restored.items] == [
184        item.queue_item_id for item in data.items
185    ]
186    # the wire-facing count matches the items actually restored (not the cached count)
187    assert restored.queue.items == len(restored.items) == 1
188    # and the settings are untouched by the item failure
189    assert restored.queue.shuffle_enabled is True
190    assert restored.queue.crossfade_enabled is True
191
192
193def test_from_cache_keeps_settings_when_source_unreadable() -> None:
194    """An unreadable source mapping is skipped instead of aborting the whole restore."""
195    queue = _queue(shuffle_enabled=True, sources=[ItemMapping.from_item(_dynamic_playlist())])
196    data = PlayerQueueData(queue=queue)
197    state = data.to_cache()
198    # corrupt the persisted source mappings so their deserialization raises
199    state["queue"]["sources"] = [{"not": "a mapping"}]
200
201    restored = PlayerQueueData.from_cache(state, data.items_to_cache())
202
203    assert restored.queue.sources == []
204    assert restored.queue.shuffle_enabled is True
205
206
207def test_from_cache_keeps_settings_when_enqueued_media_unreadable() -> None:
208    """Unreadable enqueued media clears only the media payload, never the queue settings."""
209    queue = _queue(shuffle_enabled=True, crossfade_enabled=True)
210    data = PlayerQueueData(queue=queue, enqueued_media_items=[_track("seed")])
211    state = data.to_cache()
212    # corrupt the persisted enqueued media so its deserialization raises
213    state["enqueued_media_items"] = [{"not": "a media item"}]
214
215    restored = PlayerQueueData.from_cache(state, data.items_to_cache())
216
217    assert restored.enqueued_media_items == []
218    assert restored.queue.shuffle_enabled is True
219    assert restored.queue.crossfade_enabled is True
220
221
222def test_from_cache_incompatible_version_raises() -> None:
223    """A cache written with an incompatible format version is rejected rather than misread."""
224    data = _data_with_dynamic_source()
225    state = data.to_cache()
226    state["cache_format_version"] = CACHE_FORMAT_VERSION + 1
227
228    with pytest.raises(ValueError, match="incompatible queue cache format"):
229        PlayerQueueData.from_cache(state, data.items_to_cache())
230
231
232def test_from_cache_reads_legacy_flat_layout() -> None:
233    """A pre-refactor cache (wire fields at the top level, no envelope) is still restored."""
234    data = _data_with_dynamic_source()
235    nested = data.to_cache()
236    # rebuild the old flat layout: wire fields at the top level, server-state keys alongside, and
237    # no cache_format_version (caches written before the refactor had no version stamp)
238    flat = {
239        **nested["queue"],
240        "enqueued_media_items": nested["enqueued_media_items"],
241        "source_items": nested["source_items"],
242        "userid": nested["userid"],
243    }
244
245    restored = PlayerQueueData.from_cache(flat, data.items_to_cache())
246
247    assert restored.queue.is_dynamic is True
248    assert [item.uri for item in restored.source_items] == [item.uri for item in data.source_items]
249    assert [item.uri for item in restored.enqueued_media_items] == [
250        item.uri for item in data.enqueued_media_items
251    ]
252    assert restored.userid == "user-1"
253
254
255def test_cache_significant_ignores_playback_progress() -> None:
256    """Change detection ignores volatile playback-progress fields but catches real changes."""
257    data = _data_with_dynamic_source()
258    state = data.to_cache()
259    # only elapsed time / playback speed advanced -> not a persist-worthy change
260    progressed = {
261        **state,
262        "queue": {**state["queue"], "elapsed_time": 999.0, "playback_speed": 2.0},
263    }
264    assert PlayerQueueData.cache_significant(progressed) == PlayerQueueData.cache_significant(state)
265    # a real settings change -> detected
266    shuffled = {
267        **state,
268        "queue": {**state["queue"], "shuffle_enabled": not state["queue"]["shuffle_enabled"]},
269    }
270    assert PlayerQueueData.cache_significant(shuffled) != PlayerQueueData.cache_significant(state)
271