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