music-assistant-server

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