/
/
/
1"""Tests for aligning Music Assistant queue indices with Plex play queue indices."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import Any
7from unittest.mock import AsyncMock, Mock
8
9from music_assistant_models.enums import PlaybackState
10
11from music_assistant.providers.plex_connect.queue_commands import QueueCommandsMixin
12from music_assistant.providers.plex_connect.queue_sync import QueueSyncMixin
13
14
15def _fake_track(track_key: str) -> SimpleNamespace:
16 """Return a stand-in for the MA track a Plex item resolves to."""
17 return SimpleNamespace(key=track_key, name=track_key)
18
19
20class _QueueHandler(QueueSyncMixin, QueueCommandsMixin):
21 """The queue mixins with the host-class attributes mocked."""
22
23 def __init__(self) -> None:
24 self.play_media = AsyncMock()
25 provider = Mock()
26 provider.get_track = AsyncMock(side_effect=_fake_track)
27 provider.mass.player_queues.play_media = self.play_media
28 self.queue = SimpleNamespace(
29 state=PlaybackState.PLAYING, current_index=0, index_in_buffer=0
30 )
31 provider.mass.player_queues.get = Mock(return_value=self.queue)
32 self.provider = provider
33 self._ma_player_id = "player1"
34 self._updating_from_plex = False
35 self.play_queue_id = "1093"
36 self.play_queue_item_ids: dict[int, int] = {}
37
38 def queued_keys(self) -> list[str]:
39 """Return the Plex keys handed to play_media, in order."""
40 await_args = self.play_media.await_args
41 assert await_args is not None
42 return [track.key for track in await_args.kwargs["media"]]
43
44
45def _make_playqueue(count: int, selected_index: int) -> Any:
46 """
47 Build a fake Plex PlayQueue of ``count`` items in album order.
48
49 Item N gets key ``/library/metadata/N`` and playQueueItemID ``1000+N``.
50 """
51 items = [
52 SimpleNamespace(key=f"/library/metadata/{n}", playQueueItemID=1000 + n)
53 for n in range(count)
54 ]
55 return SimpleNamespace(
56 items=items,
57 playQueueSelectedItemID=1000 + selected_index,
58 playQueueSelectedItemOffset=selected_index,
59 )
60
61
62async def test_refresh_does_not_requeue_the_playing_track() -> None:
63 """
64 A refresh must not put the currently playing track back into the queue.
65
66 MA's queue is the Plex queue rotated to start at the selected item, so MA index 0
67 is Plex index 1 here. Slicing Plex at the MA index re-queued Plex item 1.
68 """
69 handler = _QueueHandler()
70 playqueue = _make_playqueue(12, selected_index=1)
71 # Playback started at Plex index 1, so that item sits at MA index 0.
72 handler.play_queue_item_ids = {0: 1001}
73
74 await handler._replace_remaining_queue("player1", playqueue, 0)
75
76 assert "/library/metadata/1" not in handler.queued_keys()
77
78
79async def test_refresh_preserves_the_wrapped_queue_order() -> None:
80 """Items before the playing track wrap to the tail, as on the initial load."""
81 handler = _QueueHandler()
82 playqueue = _make_playqueue(12, selected_index=1)
83 handler.play_queue_item_ids = {0: 1001}
84
85 await handler._replace_remaining_queue("player1", playqueue, 0)
86
87 expected = [f"/library/metadata/{n}" for n in [*range(2, 12), 0]]
88 assert handler.queued_keys() == expected
89
90
91async def test_refresh_keeps_item_ids_in_ma_index_space() -> None:
92 """The playQueueItemID map stays keyed by MA index, not by Plex index."""
93 handler = _QueueHandler()
94 playqueue = _make_playqueue(12, selected_index=1)
95 handler.play_queue_item_ids = {0: 1001}
96
97 await handler._replace_remaining_queue("player1", playqueue, 0)
98
99 # MA index 1 now holds Plex item 2, and the wrapped Plex item 0 lands last.
100 assert handler.play_queue_item_ids[0] == 1001
101 assert handler.play_queue_item_ids[1] == 1002
102 assert handler.play_queue_item_ids[11] == 1000
103
104
105async def test_refresh_falls_back_to_the_plex_selected_item() -> None:
106 """Without a mapping for the MA index, trust the item Plex reports as selected."""
107 handler = _QueueHandler()
108 playqueue = _make_playqueue(12, selected_index=3)
109 handler.play_queue_item_ids = {}
110
111 await handler._replace_remaining_queue("player1", playqueue, 0)
112
113 assert "/library/metadata/3" not in handler.queued_keys()
114
115
116async def test_refresh_does_not_requeue_the_buffered_track() -> None:
117 """
118 REPLACE_NEXT keeps the buffered item as well, so it must not be re-queued.
119
120 The player has already been handed MA index 1, so REPLACE_NEXT inserts after it.
121 Slicing from the playing index hands that same track over a second time.
122 """
123 handler = _QueueHandler()
124 playqueue = _make_playqueue(12, selected_index=1)
125 handler.play_queue_item_ids = {0: 1001, 1: 1002}
126 handler.queue.index_in_buffer = 1
127
128 await handler._replace_remaining_queue("player1", playqueue, 0)
129
130 assert "/library/metadata/2" not in handler.queued_keys()
131
132
133async def test_refresh_replaces_only_what_follows_the_buffered_item() -> None:
134 """The buffered item is kept, so the replacement starts one track later."""
135 handler = _QueueHandler()
136 playqueue = _make_playqueue(12, selected_index=1)
137 handler.play_queue_item_ids = {0: 1001, 1: 1002}
138 handler.queue.index_in_buffer = 1
139
140 await handler._replace_remaining_queue("player1", playqueue, 0)
141
142 expected = [f"/library/metadata/{n}" for n in [*range(3, 12), 0]]
143 assert handler.queued_keys() == expected
144
145
146async def test_refresh_on_a_windowed_queue_resumes_after_the_current_item() -> None:
147 """
148 A window that omits MA index 0 must still resume right after the current item.
149
150 The server returns a window centred on the playing track, so on a long queue the
151 item MA started from is not in the response and no rotation can be derived from it.
152 """
153 handler = _QueueHandler()
154 items = [
155 SimpleNamespace(key=f"/library/metadata/{n}", playQueueItemID=1000 + n)
156 for n in range(40, 100)
157 ]
158 playqueue = SimpleNamespace(
159 items=items,
160 playQueueTotalCount=200,
161 playQueueSelectedItemID=1060,
162 playQueueSelectedItemOffset=60,
163 )
164 # MA started from track 0, which this window does not contain.
165 handler.play_queue_item_ids = {0: 1000, 60: 1060}
166 handler.queue.current_index = 60
167 handler.queue.index_in_buffer = 60
168
169 await handler._replace_remaining_queue("player1", playqueue, 60)
170
171 assert handler.queued_keys() == [f"/library/metadata/{n}" for n in range(61, 100)]
172
173
174async def test_refresh_after_playback_wrapped_past_the_origin() -> None:
175 """Once playback reaches the wrapped tail, only the tracks before the origin remain."""
176 handler = _QueueHandler()
177 playqueue = _make_playqueue(12, selected_index=4)
178 # Playback began at Plex index 4 and has reached MA index 9, which is Plex index 1.
179 handler.play_queue_item_ids = {0: 1004, 9: 1001}
180 handler.queue.current_index = 9
181 handler.queue.index_in_buffer = 9
182
183 await handler._replace_remaining_queue("player1", playqueue, 9)
184
185 assert handler.queued_keys() == ["/library/metadata/2", "/library/metadata/3"]
186
187
188async def test_refresh_when_the_origin_track_was_removed_from_plex() -> None:
189 """
190 A removed origin still leaves a wrap: the boundary moves to the next MA item.
191
192 An absent origin in a complete response means the track was deleted from the queue,
193 which is not the same as it falling outside a truncated one.
194 """
195 handler = _QueueHandler()
196 # MA holds [1, 2, 3, 0] and is playing item 1; Plex no longer lists item 1.
197 items = [
198 SimpleNamespace(key=f"/library/metadata/{n}", playQueueItemID=1000 + n) for n in (0, 2, 3)
199 ]
200 playqueue = SimpleNamespace(
201 items=items,
202 playQueueTotalCount=3,
203 playQueueSelectedItemID=1002,
204 playQueueSelectedItemOffset=1,
205 )
206 handler.play_queue_item_ids = {0: 1001, 1: 1002, 2: 1003, 3: 1000}
207 handler.queue.current_index = 1
208 handler.queue.index_in_buffer = 1
209
210 await handler._replace_remaining_queue("player1", playqueue, 1)
211
212 assert handler.queued_keys() == ["/library/metadata/3", "/library/metadata/0"]
213