/
/
/
1"""Tests for PlayerQueuesController.transfer_queue protocol/ungroup and settings handover."""
2
3from __future__ import annotations
4
5from typing import cast
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.enums import AlbumType, PlaybackState, PlayerType
10from music_assistant_models.errors import PlayerCommandFailed
11from music_assistant_models.media_items import Album, Track
12from music_assistant_models.player_queue import PlayerQueue
13
14from music_assistant.controllers.player_queues import PlayerQueuesController
15from music_assistant.controllers.player_queues.state import PlayerQueueData
16
17
18class _DummyACM:
19 """Minimal async context manager stand-in for wait_for_player_update."""
20
21 async def __aenter__(self) -> None:
22 return None
23
24 async def __aexit__(self, *args: object) -> bool:
25 return False
26
27
28def _fake_controller(source_id: str, target_player: MagicMock) -> MagicMock:
29 """
30 Build a MagicMock standing in for the controller for transfer_queue tests.
31
32 The source queue is idle so the capture/stop/resume path is a no-op; the test
33 only exercises the pre-transfer ungroup branch.
34 """
35 source_queue = MagicMock()
36 source_queue.state = PlaybackState.IDLE
37 source_queue.resume_pos = 0
38 source_queue.elapsed_time = 0
39 source_queue.current_index = 0
40 source_queue.current_item = None
41 target_queue = MagicMock()
42
43 fake = MagicMock()
44 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == source_id else target_queue)
45 fake.stop = AsyncMock()
46 fake.load = AsyncMock()
47 fake.resume = AsyncMock()
48 fake.clear = MagicMock()
49 fake.update_items = MagicMock()
50 fake._notify_audio_source_transferred = AsyncMock()
51 fake.mass.players.get_player = MagicMock(return_value=target_player)
52 fake.mass.players.cmd_ungroup = AsyncMock()
53 fake.mass.players.wait_for_player_update = MagicMock(return_value=_DummyACM())
54 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
55 return fake
56
57
58async def test_transfer_queue_ad_hoc_member_ungroups_target_not_leader() -> None:
59 """
60 Transferring onto an ad-hoc sync member frees the target itself, not its leader.
61
62 Ungrouping the leader would transfer leadership to a remaining member and
63 recurse back into transfer_queue, so only the target may be ungrouped.
64 """
65 target_player = MagicMock()
66 target_player.state.type = PlayerType.PLAYER
67 target_player.state.synced_to = "leaderA"
68 target_player.state.active_group = None
69 fake = _fake_controller("src", target_player)
70
71 await PlayerQueuesController.transfer_queue(
72 cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
73 )
74
75 fake.mass.players.cmd_ungroup.assert_awaited_once_with("memberB")
76
77
78async def test_transfer_queue_refuses_non_audio_target() -> None:
79 """
80 A target that can never render audio is refused before anything is touched.
81
82 The source queue must survive intact: no stop, no clear, no item handover.
83 """
84 target_player = MagicMock()
85 target_player.state.type = PlayerType.VISUALIZER
86 fake = _fake_controller("src", target_player)
87
88 with pytest.raises(PlayerCommandFailed):
89 await PlayerQueuesController.transfer_queue(
90 cast("PlayerQueuesController", fake), "src", "viz", auto_play=False
91 )
92
93 fake.stop.assert_not_awaited()
94 fake._clear.assert_not_called()
95 fake.load.assert_not_awaited()
96 fake.mass.players.cmd_ungroup.assert_not_awaited()
97
98
99async def test_transfer_queue_group_member_ungroups_group() -> None:
100 """Transferring onto a virtual-group member releases the group player itself."""
101 target_player = MagicMock()
102 target_player.state.type = PlayerType.PLAYER
103 target_player.state.synced_to = None
104 target_player.state.active_group = "groupP"
105 fake = _fake_controller("src", target_player)
106
107 await PlayerQueuesController.transfer_queue(
108 cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
109 )
110
111 fake.mass.players.cmd_ungroup.assert_awaited_once_with("groupP")
112
113
114def _shuffle_controller(
115 source_shuffle_enabled: bool,
116 target_shuffle_enabled: bool = False,
117 source_is_dynamic: bool = False,
118 *,
119 source_crossfade_override: bool | None = None,
120 source_autoplay_override: bool | None = None,
121 target_crossfade_override: bool | None = None,
122 target_autoplay_override: bool | None = None,
123) -> MagicMock:
124 """
125 Build a controller stand-in whose two queues carry real state records.
126
127 The shuffle flag lives on the per-queue state record, so unlike the ungroup tests above these
128 need real queues rather than one shared mock.
129
130 :param source_shuffle_enabled: Whether shuffle is on for the queue being handed over.
131 :param target_shuffle_enabled: Whether shuffle is on for the queue being handed to.
132 :param source_is_dynamic: Whether the source queue is managed by a dynamic source.
133 :param source_crossfade_override: Pinned crossfade override on the source queue, if any.
134 :param source_autoplay_override: Pinned autoplay override on the source queue, if any.
135 :param target_crossfade_override: Pinned crossfade override already on the target, if any.
136 :param target_autoplay_override: Pinned autoplay override already on the target, if any.
137 """
138 source_queue = PlayerQueue(
139 queue_id="src",
140 active=True,
141 display_name="Src",
142 available=True,
143 items=0,
144 shuffle_enabled=source_shuffle_enabled,
145 smart_shuffle_active=source_is_dynamic,
146 is_dynamic=source_is_dynamic,
147 # the wire fields carry the already-resolved effective value for these tests
148 crossfade_enabled=bool(source_crossfade_override),
149 autoplay_enabled=bool(source_autoplay_override),
150 )
151 target_queue = PlayerQueue(
152 queue_id="tgt",
153 active=True,
154 display_name="Tgt",
155 available=True,
156 items=0,
157 shuffle_enabled=target_shuffle_enabled,
158 crossfade_enabled=bool(target_crossfade_override),
159 autoplay_enabled=bool(target_autoplay_override),
160 )
161
162 fake = MagicMock()
163 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == "src" else target_queue)
164 fake._queue_data = {
165 "src": PlayerQueueData(
166 queue=source_queue,
167 crossfade_override=source_crossfade_override,
168 autoplay_override=source_autoplay_override,
169 ),
170 "tgt": PlayerQueueData(
171 queue=target_queue,
172 crossfade_override=target_crossfade_override,
173 autoplay_override=target_autoplay_override,
174 ),
175 }
176 fake.stop = AsyncMock()
177 fake.load = AsyncMock()
178 fake.resume = AsyncMock()
179 fake._clear = MagicMock()
180 fake.update_items = MagicMock()
181 fake._notify_audio_source_transferred = AsyncMock()
182 fake.is_smart_shuffle_active = MagicMock(side_effect=lambda queue: queue.is_dynamic)
183 # bind the real toggle resolver (with stubbed globals) so the handover under test
184 # exercises actual logic
185 defaults = {"autoplay_enabled": True, "crossfade_enabled": False}
186 fake.mass.config.get_raw_core_config_value = MagicMock(
187 side_effect=lambda _core_module, key, _default: defaults[key]
188 )
189 fake._resolve_default_toggles = lambda queue_data: (
190 PlayerQueuesController._resolve_default_toggles(fake, queue_data)
191 )
192 target_player = MagicMock()
193 target_player.state.type = PlayerType.PLAYER
194 target_player.state.synced_to = None
195 target_player.state.active_group = None
196 fake.mass.players.get_player = MagicMock(return_value=target_player)
197 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
198 return fake
199
200
201async def test_transfer_queue_carries_the_album_credit_bookkeeping() -> None:
202 """A credited album stays credited on the player the queue is handed to."""
203 fake = _shuffle_controller(source_shuffle_enabled=False)
204 album = Album(
205 item_id="a1",
206 provider="library",
207 name="A",
208 provider_mappings=set(),
209 album_type=AlbumType.ALBUM,
210 )
211 track = Track(item_id="t1", provider="library", name="T1", provider_mappings=set(), album=album)
212 source_data = fake._queue_data["src"]
213 source_data.enqueued_media_items = [album]
214 source_data.credited_albums = {album}
215
216 await PlayerQueuesController.transfer_queue(
217 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
218 )
219
220 target_data = fake._queue_data["tgt"]
221 # the target holds its own copy, so the source's set no longer drives it
222 assert target_data.credited_albums == {album}
223 assert target_data.credited_albums is not source_data.credited_albums
224 # and the album is not credited a second time on the new player
225 assert (
226 PlayerQueuesController._claim_enqueued_album_credit(
227 cast("PlayerQueuesController", fake), target_data, track
228 )
229 is None
230 )
231
232
233async def test_transfer_queue_overwrites_the_targets_own_shuffle() -> None:
234 """The queue brings its own shuffle state, so the target's previous one does not survive."""
235 fake = _shuffle_controller(source_shuffle_enabled=False, target_shuffle_enabled=True)
236
237 await PlayerQueuesController.transfer_queue(
238 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
239 )
240
241 assert fake.get("tgt").shuffle_enabled is False
242
243
244async def test_transfer_queue_carries_the_source_shuffle() -> None:
245 """A shuffled queue handed to another player stays shuffled there."""
246 fake = _shuffle_controller(source_shuffle_enabled=True)
247
248 await PlayerQueuesController.transfer_queue(
249 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
250 )
251
252 assert fake.get("tgt").shuffle_enabled is True
253
254
255async def test_transfer_queue_drops_dynamic_shuffle_from_source() -> None:
256 """The shuffle imposed by a dynamic source follows it to the target queue."""
257 fake = _shuffle_controller(source_shuffle_enabled=True, source_is_dynamic=True)
258 fake._clear.side_effect = lambda queue_id, skip_stop=False: PlayerQueuesController._clear(
259 cast("PlayerQueuesController", fake), queue_id, skip_stop
260 )
261
262 await PlayerQueuesController.transfer_queue(
263 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
264 )
265
266 assert fake.get("tgt").is_dynamic is True
267 assert fake.get("tgt").shuffle_enabled is True
268 assert fake.get("src").is_dynamic is False
269 assert fake.get("src").shuffle_enabled is False
270 assert fake.get("src").smart_shuffle_active is False
271
272
273async def test_transfer_queue_carries_pinned_toggle_overrides() -> None:
274 """Pinned autoplay/crossfade overrides, and their already-resolved wire values, reach the target."""
275 fake = _shuffle_controller(
276 source_shuffle_enabled=False,
277 source_crossfade_override=True,
278 source_autoplay_override=False,
279 target_crossfade_override=False,
280 target_autoplay_override=True,
281 )
282
283 await PlayerQueuesController.transfer_queue(
284 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
285 )
286
287 target_data = fake._queue_data["tgt"]
288 assert target_data.crossfade_override is True
289 assert target_data.autoplay_override is False
290 assert target_data.queue.crossfade_enabled is True
291 assert target_data.queue.autoplay_enabled is False
292