/
/
1"""Tests for PlayerQueuesController.transfer_queue protocol/ungroup and settings handover."""
2
3from __future__ import annotations
4
5import time
6from typing import cast
7from unittest.mock import AsyncMock, MagicMock
8
9from music_assistant_models.enums import PlaybackState
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.mass.players.get_player = MagicMock(return_value=target_player)
49 fake.mass.players.cmd_ungroup = AsyncMock()
50 fake.mass.players.wait_for_player_update = MagicMock(return_value=_DummyACM())
51 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
52 return fake
53
54
55async def test_transfer_queue_ad_hoc_member_ungroups_target_not_leader() -> None:
56 """
57 Transferring onto an ad-hoc sync member frees the target itself, not its leader.
58
59 Ungrouping the leader would transfer leadership to a remaining member and
60 recurse back into transfer_queue, so only the target may be ungrouped.
61 """
62 target_player = MagicMock()
63 target_player.state.synced_to = "leaderA"
64 target_player.state.active_group = None
65 fake = _fake_controller("src", target_player)
66
67 await PlayerQueuesController.transfer_queue(
68 cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
69 )
70
71 fake.mass.players.cmd_ungroup.assert_awaited_once_with("memberB")
72
73
74async def test_transfer_queue_group_member_ungroups_group() -> None:
75 """Transferring onto a virtual-group member releases the group player itself."""
76 target_player = MagicMock()
77 target_player.state.synced_to = None
78 target_player.state.active_group = "groupP"
79 fake = _fake_controller("src", target_player)
80
81 await PlayerQueuesController.transfer_queue(
82 cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
83 )
84
85 fake.mass.players.cmd_ungroup.assert_awaited_once_with("groupP")
86
87
88def _shuffle_controller(
89 source_shuffle_enabled: bool,
90 source_shuffle_set_at: float | None,
91 target_shuffle_set_at: float | None,
92 source_is_dynamic: bool = False,
93) -> MagicMock:
94 """
95 Build a controller stand-in whose two queues carry real state records.
96
97 Both the shuffle flag and the shuffle intent behind it live on separate per-queue objects, so
98 unlike the ungroup tests above these 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 source_shuffle_set_at: When the user last switched shuffle on for the source queue.
102 :param target_shuffle_set_at: When the user last switched shuffle on for the target queue.
103 :param source_is_dynamic: Whether the source queue is managed by a dynamic source.
104 """
105 source_queue = PlayerQueue(
106 queue_id="src",
107 active=True,
108 display_name="Src",
109 available=True,
110 items=0,
111 shuffle_enabled=source_shuffle_enabled,
112 smart_shuffle_active=source_is_dynamic,
113 is_dynamic=source_is_dynamic,
114 )
115 target_queue = PlayerQueue(
116 queue_id="tgt", active=True, display_name="Tgt", available=True, items=0
117 )
118
119 fake = MagicMock()
120 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == "src" else target_queue)
121 fake._queue_data = {
122 "src": PlayerQueueData(queue=source_queue, shuffle_set_at=source_shuffle_set_at),
123 "tgt": PlayerQueueData(queue=target_queue, shuffle_set_at=target_shuffle_set_at),
124 }
125 fake.stop = AsyncMock()
126 fake.load = AsyncMock()
127 fake.resume = AsyncMock()
128 fake._clear = MagicMock()
129 fake.update_items = MagicMock()
130 fake.is_smart_shuffle_active = MagicMock(side_effect=lambda queue: queue.is_dynamic)
131 target_player = MagicMock()
132 target_player.state.synced_to = None
133 target_player.state.active_group = None
134 fake.mass.players.get_player = MagicMock(return_value=target_player)
135 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
136 return fake
137
138
139async def test_transfer_queue_drops_a_stale_shuffle_intent_on_the_target() -> None:
140 """
141 A shuffle the user switched on for the target earlier does not shuffle the queue moved onto it.
142
143 The transfer brings its own (off) shuffle state; leaving the target's own stamp behind would
144 make the media started next read it as a "shuffle this" gesture the user never made for it.
145 """
146 fake = _shuffle_controller(
147 source_shuffle_enabled=False,
148 source_shuffle_set_at=None,
149 target_shuffle_set_at=time.monotonic(),
150 )
151
152 await PlayerQueuesController.transfer_queue(
153 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
154 )
155
156 assert fake.get("tgt").shuffle_enabled is False
157 assert fake._queue_data["tgt"].shuffle_set_at is None
158
159
160async def test_transfer_queue_carries_the_source_shuffle_intent() -> None:
161 """A shuffle switched on moments before the transfer still counts for the media started next."""
162 switched_on_at = time.monotonic()
163 fake = _shuffle_controller(
164 source_shuffle_enabled=True,
165 source_shuffle_set_at=switched_on_at,
166 target_shuffle_set_at=None,
167 )
168
169 await PlayerQueuesController.transfer_queue(
170 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
171 )
172
173 assert fake.get("tgt").shuffle_enabled is True
174 assert fake._queue_data["tgt"].shuffle_set_at == switched_on_at
175 # the gesture is good for one play and it followed the queue, so it must not be left behind
176 # to shuffle whatever gets started on the player it was moved off
177 assert fake._queue_data["src"].shuffle_set_at is None
178
179
180async def test_transfer_queue_drops_dynamic_shuffle_from_source() -> None:
181 """The shuffle imposed by a dynamic source follows it to the target queue."""
182 fake = _shuffle_controller(
183 source_shuffle_enabled=True,
184 source_shuffle_set_at=None,
185 target_shuffle_set_at=None,
186 source_is_dynamic=True,
187 )
188 fake._clear.side_effect = lambda queue_id, skip_stop=False: PlayerQueuesController._clear(
189 cast("PlayerQueuesController", fake), queue_id, skip_stop
190 )
191
192 await PlayerQueuesController.transfer_queue(
193 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
194 )
195
196 assert fake.get("tgt").is_dynamic is True
197 assert fake.get("tgt").shuffle_enabled is True
198 assert fake.get("src").is_dynamic is False
199 assert fake.get("src").shuffle_enabled is False
200 assert fake.get("src").smart_shuffle_active is False
201