/
/
/
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) -> MagicMock:
119 """
120 Build a controller stand-in whose two queues carry real state records.
121
122 The shuffle flag lives on the per-queue state record, so unlike the ungroup tests above these
123 need real queues rather than one shared mock.
124
125 :param source_shuffle_enabled: Whether shuffle is on for the queue being handed over.
126 :param target_shuffle_enabled: Whether shuffle is on for the queue being handed to.
127 :param source_is_dynamic: Whether the source queue is managed by a dynamic source.
128 """
129 source_queue = PlayerQueue(
130 queue_id="src",
131 active=True,
132 display_name="Src",
133 available=True,
134 items=0,
135 shuffle_enabled=source_shuffle_enabled,
136 smart_shuffle_active=source_is_dynamic,
137 is_dynamic=source_is_dynamic,
138 )
139 target_queue = PlayerQueue(
140 queue_id="tgt",
141 active=True,
142 display_name="Tgt",
143 available=True,
144 items=0,
145 shuffle_enabled=target_shuffle_enabled,
146 )
147
148 fake = MagicMock()
149 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == "src" else target_queue)
150 fake._queue_data = {
151 "src": PlayerQueueData(queue=source_queue),
152 "tgt": PlayerQueueData(queue=target_queue),
153 }
154 fake.stop = AsyncMock()
155 fake.load = AsyncMock()
156 fake.resume = AsyncMock()
157 fake._clear = MagicMock()
158 fake.update_items = MagicMock()
159 fake._notify_audio_source_transferred = AsyncMock()
160 fake.is_smart_shuffle_active = MagicMock(side_effect=lambda queue: queue.is_dynamic)
161 target_player = MagicMock()
162 target_player.state.type = PlayerType.PLAYER
163 target_player.state.synced_to = None
164 target_player.state.active_group = None
165 fake.mass.players.get_player = MagicMock(return_value=target_player)
166 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
167 return fake
168
169
170async def test_transfer_queue_carries_the_album_credit_bookkeeping() -> None:
171 """A credited album stays credited on the player the queue is handed to."""
172 fake = _shuffle_controller(source_shuffle_enabled=False)
173 album = Album(
174 item_id="a1",
175 provider="library",
176 name="A",
177 provider_mappings=set(),
178 album_type=AlbumType.ALBUM,
179 )
180 track = Track(item_id="t1", provider="library", name="T1", provider_mappings=set(), album=album)
181 source_data = fake._queue_data["src"]
182 source_data.enqueued_media_items = [album]
183 source_data.credited_albums = {album}
184
185 await PlayerQueuesController.transfer_queue(
186 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
187 )
188
189 target_data = fake._queue_data["tgt"]
190 # the target holds its own copy, so the source's set no longer drives it
191 assert target_data.credited_albums == {album}
192 assert target_data.credited_albums is not source_data.credited_albums
193 # and the album is not credited a second time on the new player
194 assert (
195 PlayerQueuesController._claim_enqueued_album_credit(
196 cast("PlayerQueuesController", fake), target_data, track
197 )
198 is None
199 )
200
201
202async def test_transfer_queue_overwrites_the_targets_own_shuffle() -> None:
203 """The queue brings its own shuffle state, so the target's previous one does not survive."""
204 fake = _shuffle_controller(source_shuffle_enabled=False, target_shuffle_enabled=True)
205
206 await PlayerQueuesController.transfer_queue(
207 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
208 )
209
210 assert fake.get("tgt").shuffle_enabled is False
211
212
213async def test_transfer_queue_carries_the_source_shuffle() -> None:
214 """A shuffled queue handed to another player stays shuffled there."""
215 fake = _shuffle_controller(source_shuffle_enabled=True)
216
217 await PlayerQueuesController.transfer_queue(
218 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
219 )
220
221 assert fake.get("tgt").shuffle_enabled is True
222
223
224async def test_transfer_queue_drops_dynamic_shuffle_from_source() -> None:
225 """The shuffle imposed by a dynamic source follows it to the target queue."""
226 fake = _shuffle_controller(source_shuffle_enabled=True, source_is_dynamic=True)
227 fake._clear.side_effect = lambda queue_id, skip_stop=False: PlayerQueuesController._clear(
228 cast("PlayerQueuesController", fake), queue_id, skip_stop
229 )
230
231 await PlayerQueuesController.transfer_queue(
232 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
233 )
234
235 assert fake.get("tgt").is_dynamic is True
236 assert fake.get("tgt").shuffle_enabled is True
237 assert fake.get("src").is_dynamic is False
238 assert fake.get("src").shuffle_enabled is False
239 assert fake.get("src").smart_shuffle_active is False
240