/
/
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
8from music_assistant_models.enums import AlbumType, PlaybackState
9from music_assistant_models.media_items import Album, Track
10from music_assistant_models.player_queue import PlayerQueue
11
12from music_assistant.controllers.player_queues import PlayerQueuesController
13from music_assistant.controllers.player_queues.state import PlayerQueueData
14
15
16class _DummyACM:
17 """Minimal async context manager stand-in for wait_for_player_update."""
18
19 async def __aenter__(self) -> None:
20 return None
21
22 async def __aexit__(self, *args: object) -> bool:
23 return False
24
25
26def _fake_controller(source_id: str, target_player: MagicMock) -> MagicMock:
27 """
28 Build a MagicMock standing in for the controller for transfer_queue tests.
29
30 The source queue is idle so the capture/stop/resume path is a no-op; the test
31 only exercises the pre-transfer ungroup branch.
32 """
33 source_queue = MagicMock()
34 source_queue.state = PlaybackState.IDLE
35 source_queue.resume_pos = 0
36 source_queue.elapsed_time = 0
37 source_queue.current_index = 0
38 source_queue.current_item = None
39 target_queue = MagicMock()
40
41 fake = MagicMock()
42 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == source_id else target_queue)
43 fake.stop = AsyncMock()
44 fake.load = AsyncMock()
45 fake.resume = AsyncMock()
46 fake.clear = MagicMock()
47 fake.update_items = MagicMock()
48 fake._notify_audio_source_transferred = AsyncMock()
49 fake.mass.players.get_player = MagicMock(return_value=target_player)
50 fake.mass.players.cmd_ungroup = AsyncMock()
51 fake.mass.players.wait_for_player_update = MagicMock(return_value=_DummyACM())
52 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
53 return fake
54
55
56async def test_transfer_queue_ad_hoc_member_ungroups_target_not_leader() -> None:
57 """
58 Transferring onto an ad-hoc sync member frees the target itself, not its leader.
59
60 Ungrouping the leader would transfer leadership to a remaining member and
61 recurse back into transfer_queue, so only the target may be ungrouped.
62 """
63 target_player = MagicMock()
64 target_player.state.synced_to = "leaderA"
65 target_player.state.active_group = None
66 fake = _fake_controller("src", target_player)
67
68 await PlayerQueuesController.transfer_queue(
69 cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
70 )
71
72 fake.mass.players.cmd_ungroup.assert_awaited_once_with("memberB")
73
74
75async def test_transfer_queue_group_member_ungroups_group() -> None:
76 """Transferring onto a virtual-group member releases the group player itself."""
77 target_player = MagicMock()
78 target_player.state.synced_to = None
79 target_player.state.active_group = "groupP"
80 fake = _fake_controller("src", target_player)
81
82 await PlayerQueuesController.transfer_queue(
83 cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
84 )
85
86 fake.mass.players.cmd_ungroup.assert_awaited_once_with("groupP")
87
88
89def _shuffle_controller(
90 source_shuffle_enabled: bool,
91 target_shuffle_enabled: bool = False,
92 source_is_dynamic: bool = False,
93) -> MagicMock:
94 """
95 Build a controller stand-in whose two queues carry real state records.
96
97 The shuffle flag lives on the per-queue state record, so unlike the ungroup tests above these
98 need real queues rather than one shared mock.
99
100 :param source_shuffle_enabled: Whether shuffle is on for the queue being handed over.
101 :param target_shuffle_enabled: Whether shuffle is on for the queue being handed to.
102 :param source_is_dynamic: Whether the source queue is managed by a dynamic source.
103 """
104 source_queue = PlayerQueue(
105 queue_id="src",
106 active=True,
107 display_name="Src",
108 available=True,
109 items=0,
110 shuffle_enabled=source_shuffle_enabled,
111 smart_shuffle_active=source_is_dynamic,
112 is_dynamic=source_is_dynamic,
113 )
114 target_queue = PlayerQueue(
115 queue_id="tgt",
116 active=True,
117 display_name="Tgt",
118 available=True,
119 items=0,
120 shuffle_enabled=target_shuffle_enabled,
121 )
122
123 fake = MagicMock()
124 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == "src" else target_queue)
125 fake._queue_data = {
126 "src": PlayerQueueData(queue=source_queue),
127 "tgt": PlayerQueueData(queue=target_queue),
128 }
129 fake.stop = AsyncMock()
130 fake.load = AsyncMock()
131 fake.resume = AsyncMock()
132 fake._clear = MagicMock()
133 fake.update_items = MagicMock()
134 fake._notify_audio_source_transferred = AsyncMock()
135 fake.is_smart_shuffle_active = MagicMock(side_effect=lambda queue: queue.is_dynamic)
136 target_player = MagicMock()
137 target_player.state.synced_to = None
138 target_player.state.active_group = None
139 fake.mass.players.get_player = MagicMock(return_value=target_player)
140 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
141 return fake
142
143
144async def test_transfer_queue_carries_the_album_credit_bookkeeping() -> None:
145 """A credited album stays credited on the player the queue is handed to."""
146 fake = _shuffle_controller(source_shuffle_enabled=False)
147 album = Album(
148 item_id="a1",
149 provider="library",
150 name="A",
151 provider_mappings=set(),
152 album_type=AlbumType.ALBUM,
153 )
154 track = Track(item_id="t1", provider="library", name="T1", provider_mappings=set(), album=album)
155 source_data = fake._queue_data["src"]
156 source_data.enqueued_media_items = [album]
157 source_data.credited_albums = {album}
158
159 await PlayerQueuesController.transfer_queue(
160 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
161 )
162
163 target_data = fake._queue_data["tgt"]
164 # the target holds its own copy, so the source's set no longer drives it
165 assert target_data.credited_albums == {album}
166 assert target_data.credited_albums is not source_data.credited_albums
167 # and the album is not credited a second time on the new player
168 assert (
169 PlayerQueuesController._claim_enqueued_album_credit(
170 cast("PlayerQueuesController", fake), target_data, track
171 )
172 is None
173 )
174
175
176async def test_transfer_queue_overwrites_the_targets_own_shuffle() -> None:
177 """The queue brings its own shuffle state, so the target's previous one does not survive."""
178 fake = _shuffle_controller(source_shuffle_enabled=False, target_shuffle_enabled=True)
179
180 await PlayerQueuesController.transfer_queue(
181 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
182 )
183
184 assert fake.get("tgt").shuffle_enabled is False
185
186
187async def test_transfer_queue_carries_the_source_shuffle() -> None:
188 """A shuffled queue handed to another player stays shuffled there."""
189 fake = _shuffle_controller(source_shuffle_enabled=True)
190
191 await PlayerQueuesController.transfer_queue(
192 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
193 )
194
195 assert fake.get("tgt").shuffle_enabled is True
196
197
198async def test_transfer_queue_drops_dynamic_shuffle_from_source() -> None:
199 """The shuffle imposed by a dynamic source follows it to the target queue."""
200 fake = _shuffle_controller(source_shuffle_enabled=True, source_is_dynamic=True)
201 fake._clear.side_effect = lambda queue_id, skip_stop=False: PlayerQueuesController._clear(
202 cast("PlayerQueuesController", fake), queue_id, skip_stop
203 )
204
205 await PlayerQueuesController.transfer_queue(
206 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
207 )
208
209 assert fake.get("tgt").is_dynamic is True
210 assert fake.get("tgt").shuffle_enabled is True
211 assert fake.get("src").is_dynamic is False
212 assert fake.get("src").shuffle_enabled is False
213 assert fake.get("src").smart_shuffle_active is False
214