/
/
/
1"""Tests for the Sonos cloud queue window served to the speakers."""
2
3import json
4import logging
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8from aiosonos.exceptions import FailedCommand
9from music_assistant_models.enums import MediaType
10from music_assistant_models.errors import InvalidDataError
11from music_assistant_models.player import PlayerMedia
12from music_assistant_models.queue_item import QueueItem
13
14from music_assistant.providers.sonos.player import SonosPlayer, SonosQueueWindow
15from music_assistant.providers.sonos.provider import (
16 SonosPlayerProvider,
17 _refresh_task_id,
18 _requested_max,
19)
20
21QUEUE_ID = "party_queue"
22
23
24def _make_queue_item(item_id: str) -> QueueItem:
25 """Build a minimal playable queue item."""
26 return QueueItem(
27 queue_id=QUEUE_ID,
28 queue_item_id=item_id,
29 name=item_id,
30 duration=180,
31 )
32
33
34class _FakeQueues:
35 """The slice of the player-queues controller the cloud queue reads."""
36
37 def __init__(self, items: list[QueueItem], current_index: int = 0) -> None:
38 self.items = items
39 self.queue = MagicMock()
40 self.queue.current_index = current_index
41 self.queue.index_in_buffer = current_index
42
43 def get(self, queue_id: str) -> MagicMock | None:
44 """Return the queue for the given id."""
45 return self.queue if queue_id == QUEUE_ID else None
46
47 def index_by_id(self, queue_id: str, item_id: str) -> int | None:
48 """Return the index of the item with the given id."""
49 return next((i for i, x in enumerate(self.items) if x.queue_item_id == item_id), None)
50
51 def get_item(self, queue_id: str, index: int) -> QueueItem | None:
52 """Return the item at the given index."""
53 return self.items[index] if 0 <= index < len(self.items) else None
54
55 def get_next_item(self, queue_id: str, index_or_id: int | str) -> QueueItem | None:
56 """Return the item that plays after the given index or item id."""
57 if isinstance(index_or_id, int):
58 index = index_or_id
59 elif (found := self.index_by_id(queue_id, index_or_id)) is None:
60 return None
61 else:
62 index = found
63 # the real controller walks past items it cannot play
64 for candidate in self.items[index + 1 :]:
65 if candidate.available:
66 return candidate
67 return None
68
69 async def player_media_from_queue_item(self, queue_item: QueueItem) -> PlayerMedia:
70 """Return the media for the given queue item."""
71 return PlayerMedia(
72 uri=queue_item.uri,
73 media_type=MediaType.TRACK,
74 title=queue_item.name,
75 queue_item_id=queue_item.queue_item_id,
76 source_id=QUEUE_ID,
77 )
78
79
80def _make_player(items: list[QueueItem], current_index: int = 0) -> tuple[SonosPlayer, _FakeQueues]:
81 """Create a SonosPlayer serving a cloud queue for the given items."""
82 queues = _FakeQueues(items, current_index)
83 mass = MagicMock()
84 mass.player_queues = queues
85 mass.streams.resolve_stream_url = AsyncMock(side_effect=lambda _player_id, media: media.uri)
86 player = SonosPlayer.__new__(SonosPlayer)
87 player.mass = mass
88 player.logger = logging.getLogger("test.sonos.cloud_queue")
89 player._player_id = "sonos_player"
90 player.connected = True
91 player.cloud_queue_id = QUEUE_ID
92 player.cloud_queue_version = 1.0
93 player._announcement_media = None
94 return player, queues
95
96
97async def test_window_is_the_requested_item_and_the_one_after_it() -> None:
98 """Test only the item asked about and its neighbours are served, however long the queue."""
99 items = [_make_queue_item(f"track{i}") for i in range(10)]
100 player, _ = _make_player(items)
101
102 window = await player.build_cloud_queue_window("track5")
103
104 # a deeper window would let the speaker play several tracks out of a cache we cannot
105 # update; this way it has to ask again for every one
106 assert [x.queue_item_id for x in window.items] == ["track4", "track5", "track6"]
107 assert window.includes_beginning is False
108 assert window.includes_end is False
109
110
111@pytest.mark.parametrize("item_id", [None, ""], ids=["omitted", "empty"])
112async def test_window_without_an_item_id_starts_at_the_queue_head(item_id: str | None) -> None:
113 """Test an omitted or empty item id asks for the start of the queue, as Sonos specifies."""
114 items = [_make_queue_item(f"track{i}") for i in range(5)]
115 player, _ = _make_player(items, current_index=3)
116
117 window = await player.build_cloud_queue_window(item_id)
118
119 assert [x.queue_item_id for x in window.items] == ["track0", "track1"]
120 assert window.includes_beginning is True
121
122
123async def test_window_for_an_unknown_item_falls_back_to_the_playing_one() -> None:
124 """Test an item id the queue no longer holds is answered around the playing item."""
125 items = [_make_queue_item(f"track{i}") for i in range(5)]
126 player, queues = _make_player(items, current_index=1)
127 # with crossfade the buffered index runs an item ahead of what is playing
128 queues.queue.index_in_buffer = 2
129
130 window = await player.build_cloud_queue_window("gone")
131
132 assert [x.queue_item_id for x in window.items] == ["track0", "track1", "track2"]
133
134
135async def test_window_flags_the_end_of_the_queue() -> None:
136 """Test the end-of-queue flag is set once the window reaches the last item."""
137 items = [_make_queue_item(f"track{i}") for i in range(2)]
138 player, _ = _make_player(items)
139
140 window = await player.build_cloud_queue_window("track0")
141
142 assert [x.queue_item_id for x in window.items] == ["track0", "track1"]
143 assert window.includes_beginning is True
144 assert window.includes_end is True
145
146
147async def test_window_serves_an_item_added_after_the_last_enqueue() -> None:
148 """Test a track added mid-playback is served as the next one, with no enqueue in between."""
149 items = [_make_queue_item(f"track{i}") for i in range(4)]
150 player, queues = _make_player(items)
151 # the speaker has already loaded the next track, which is what stops the queue
152 # controller from announcing any further change
153 queues.queue.index_in_buffer = 1
154
155 assert [x.queue_item_id for x in (await player.build_cloud_queue_window("track0")).items] == [
156 "track0",
157 "track1",
158 ]
159
160 # a party guest adds a track behind the one the speaker already buffered
161 queues.items.insert(2, _make_queue_item("guest"))
162
163 # the speaker comes back when that buffered track starts, and is handed the new one
164 window = await player.build_cloud_queue_window("track1")
165
166 assert [x.queue_item_id for x in window.items] == ["track0", "track1", "guest"]
167
168
169async def test_unavailable_items_are_left_out() -> None:
170 """Test an item that cannot be played is not offered to the speaker."""
171 items = [_make_queue_item(f"track{i}") for i in range(3)]
172 items[0].available = False
173 player, _ = _make_player(items, current_index=1)
174
175 window = await player.build_cloud_queue_window("track1")
176
177 assert [x.queue_item_id for x in window.items] == ["track1", "track2"]
178
179
180async def test_announcement_is_served_as_a_single_item_queue() -> None:
181 """Test an announcement is the only item in the window while it plays."""
182 player, _ = _make_player([_make_queue_item("track0")])
183 player._announcement_media = PlayerMedia(
184 uri="http://announcement", media_type=MediaType.ANNOUNCEMENT, queue_item_id="announcement"
185 )
186
187 window = await player.build_cloud_queue_window("track0")
188
189 assert [x.queue_item_id for x in window.items] == ["announcement"]
190 assert window.includes_beginning is True
191 assert window.includes_end is True
192
193
194async def test_window_is_empty_without_a_queue() -> None:
195 """Test a speaker with no MA queue loaded is told the queue is over, not just empty."""
196 player, _ = _make_player([_make_queue_item("track0")])
197 player.cloud_queue_id = None
198
199 window = await player.build_cloud_queue_window(None)
200
201 assert window.items == []
202 # both ends flagged, or the speaker keeps what it cached and polls on
203 assert window.includes_beginning is True
204 assert window.includes_end is True
205
206
207async def test_refresh_bumps_the_version_and_signals_the_speaker() -> None:
208 """Test a refresh tells the speaker to re-read and invalidates its cached version."""
209 player, _ = _make_player([_make_queue_item("track0")])
210 client = MagicMock()
211 client.player.group.active_session_id = "session1"
212 client.api.playback_session.refresh_cloud_queue = AsyncMock()
213 player.client = client
214 version_before = player.cloud_queue_version
215
216 await player.refresh_cloud_queue()
217
218 assert player.cloud_queue_version > version_before
219 client.api.playback_session.refresh_cloud_queue.assert_awaited_once_with("session1")
220
221
222async def test_refresh_without_a_session_only_bumps_the_version() -> None:
223 """Test a speaker with no cloud queue loaded is not sent a refresh."""
224 player, _ = _make_player([_make_queue_item("track0")])
225 client = MagicMock()
226 client.player.group.active_session_id = None
227 client.api.playback_session.refresh_cloud_queue = AsyncMock()
228 player.client = client
229 version_before = player.cloud_queue_version
230
231 await player.refresh_cloud_queue()
232
233 assert player.cloud_queue_version > version_before
234 client.api.playback_session.refresh_cloud_queue.assert_not_awaited()
235
236
237async def test_stop_forgets_the_cloud_queue() -> None:
238 """Test a stopped speaker is no longer signalled about that queue."""
239 player, _ = _make_player([_make_queue_item("track0")])
240 client = MagicMock()
241 client.player.is_passive = False
242 client.player.group.stop = AsyncMock()
243 player.client = client
244 player.mark_stop_called = MagicMock() # type: ignore[misc, method-assign]
245 player.update_state = MagicMock() # type: ignore[misc, method-assign]
246 player._announcement_media = PlayerMedia(
247 uri="http://announcement", media_type=MediaType.ANNOUNCEMENT
248 )
249
250 await player.stop()
251
252 assert player.cloud_queue_id is None
253 assert player._announcement_media is None
254
255
256async def test_refresh_survives_a_session_the_speaker_forgot() -> None:
257 """Test a rejected refresh is not an error: the next read carries the change anyway."""
258 player, _ = _make_player([_make_queue_item("track0")])
259 client = MagicMock()
260 client.player.group.active_session_id = "stale_session"
261 client.api.playback_session.refresh_cloud_queue = AsyncMock(
262 side_effect=FailedCommand("no cloud queue loaded")
263 )
264 player.client = client
265
266 await player.refresh_cloud_queue()
267
268 client.api.playback_session.refresh_cloud_queue.assert_awaited_once()
269
270
271def _make_provider() -> SonosPlayerProvider:
272 """Create a bare provider for the cloud-queue request handlers."""
273 provider = SonosPlayerProvider.__new__(SonosPlayerProvider)
274 provider.mass = MagicMock()
275 provider.logger = logging.getLogger("test.sonos.cloud_queue")
276 provider._pending_refresh_tasks = set()
277 return provider
278
279
280async def test_itemwindow_passes_the_speakers_request_through() -> None:
281 """Test the sizes and centre the speaker asks for reach the window builder."""
282 player = MagicMock(spec=SonosPlayer)
283 player.cloud_queue_version = 12.5
284 player.build_cloud_queue_window = AsyncMock(
285 return_value=SonosQueueWindow(includes_beginning=True, includes_end=False)
286 )
287 provider = _make_provider()
288 request = MagicMock()
289 request.query = {
290 "itemId": "track7",
291 "previousWindowSize": "9",
292 "upcomingWindowSize": "10",
293 "contextVersion": "3",
294 }
295
296 response = await provider._handle_sonos_queue_itemwindow(player, request)
297
298 player.build_cloud_queue_window.assert_awaited_once_with(
299 "track7", max_previous=9, max_upcoming=10
300 )
301 body = json.loads(response.text or "{}")
302 assert body["queueVersion"] == "12.5"
303 assert body["contextVersion"] == "3"
304 assert body["includesBeginningOfQueue"] is True
305
306
307async def test_itemwindow_reports_end_of_queue_when_it_cannot_be_described() -> None:
308 """Test a queue that went away answers with an empty window instead of an error."""
309 player = MagicMock(spec=SonosPlayer)
310 player.display_name = "Kantoor"
311 player.cloud_queue_version = 1.0
312 player.build_cloud_queue_window = AsyncMock(side_effect=InvalidDataError("no session"))
313 provider = _make_provider()
314 request = MagicMock()
315 request.query = {}
316
317 response = await provider._handle_sonos_queue_itemwindow(player, request)
318
319 body = json.loads(response.text or "{}")
320 assert body["items"] == []
321 assert body["includesEndOfQueue"] is True
322
323
324@pytest.mark.parametrize(
325 ("requested", "expected_upcoming"),
326 [("10", ["track1"]), ("1", ["track1"]), ("0", []), ("", ["track1"]), (None, ["track1"])],
327 ids=["ten", "one", "zero", "unreadable", "absent"],
328)
329async def test_upcoming_is_capped_by_what_the_speaker_allows(
330 requested: str | None, expected_upcoming: list[str]
331) -> None:
332 """Test we never serve more than the speaker's maximum, though we usually serve fewer."""
333 items = [_make_queue_item(f"track{i}") for i in range(4)]
334 player, _ = _make_player(items)
335
336 window = await player.build_cloud_queue_window("track0", max_upcoming=_requested_max(requested))
337
338 assert [x.queue_item_id for x in window.items] == ["track0", *expected_upcoming]
339
340
341async def test_play_media_keeps_describing_the_queue_until_the_new_one_is_loaded() -> None:
342 """Test the still-playing queue is not blanked while its replacement is being loaded."""
343 player, _ = _make_player([_make_queue_item("track0")])
344 client = MagicMock()
345 client.player.is_passive = False
346 loaded_with_queue_id: list[str | None] = []
347
348 async def _play_cloud_queue(*_args: object, **_kwargs: object) -> None:
349 # by the time the speaker is told to load, the id must already point at the new queue
350 loaded_with_queue_id.append(player.cloud_queue_id)
351
352 client.player.group.play_cloud_queue = _play_cloud_queue
353 player.client = client
354
355 with pytest.MonkeyPatch.context() as patch:
356 patch.setattr(SonosPlayer, "flow_mode", property(lambda _self: False))
357 await player.play_media(
358 PlayerMedia(
359 uri="library://track/1",
360 media_type=MediaType.TRACK,
361 source_id="new_queue",
362 queue_item_id="item1",
363 )
364 )
365
366 assert loaded_with_queue_id == ["new_queue"]
367 assert player.cloud_queue_id == "new_queue"
368
369
370async def test_a_failed_load_leaves_no_cloud_queue_described() -> None:
371 """Test a load the speaker never accepted is not described as a queue afterwards."""
372 player, _ = _make_player([_make_queue_item("track0")])
373 client = MagicMock()
374 client.player.is_passive = False
375 client.player.group.play_cloud_queue = AsyncMock(side_effect=FailedCommand("no can do"))
376 player.client = client
377
378 with pytest.MonkeyPatch.context() as patch:
379 patch.setattr(SonosPlayer, "flow_mode", property(lambda _self: False))
380 with pytest.raises(FailedCommand):
381 await player.play_media(
382 PlayerMedia(
383 uri="library://track/1",
384 media_type=MediaType.TRACK,
385 source_id="new_queue",
386 queue_item_id="item1",
387 )
388 )
389
390 # the session was reset before the load, so claiming either queue would be a lie
391 assert player.cloud_queue_id is None
392
393
394def test_queue_change_only_reaches_the_speakers_playing_it() -> None:
395 """Test a queue edit is signalled to the speakers serving that queue."""
396 provider = SonosPlayerProvider.__new__(SonosPlayerProvider)
397 provider.mass = MagicMock()
398 provider.logger = logging.getLogger("test.sonos.cloud_queue")
399 provider._pending_refresh_tasks = set()
400 playing, other = MagicMock(spec=SonosPlayer), MagicMock(spec=SonosPlayer)
401 playing.player_id = "playing"
402 playing.cloud_queue_id = QUEUE_ID
403 other.player_id = "other"
404 other.cloud_queue_id = "another_queue"
405 with pytest.MonkeyPatch.context() as patch:
406 patch.setattr(type(provider), "players", property(lambda _self: [playing, other]))
407 provider._handle_queue_items_updated(MagicMock(object_id=QUEUE_ID))
408
409 # the version must be invalidated before the (debounced) command goes out, or a window
410 # served in between carries a version the speaker reads as current
411 playing.bump_cloud_queue_version.assert_called_once_with()
412 other.bump_cloud_queue_version.assert_not_called()
413 assert provider.mass.call_later.call_count == 1
414 assert provider.mass.call_later.call_args.args[1] == playing.refresh_cloud_queue
415 # the id must be per speaker, or one speaker's refresh cancels another's
416 assert provider.mass.call_later.call_args.kwargs["task_id"] == _refresh_task_id("playing")
417 assert provider._pending_refresh_tasks == {_refresh_task_id("playing")}
418