/
/
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 PlaybackState
9from music_assistant_models.player_queue import PlayerQueue
10
11from music_assistant.controllers.player_queues import PlayerQueuesController
12from music_assistant.controllers.player_queues.state import PlayerQueueData
13
14
15class _DummyACM:
16 """Minimal async context manager stand-in for wait_for_player_update."""
17
18 async def __aenter__(self) -> None:
19 return None
20
21 async def __aexit__(self, *args: object) -> bool:
22 return False
23
24
25def _fake_controller(source_id: str, target_player: MagicMock) -> MagicMock:
26 """
27 Build a MagicMock standing in for the controller for transfer_queue tests.
28
29 The source queue is idle so the capture/stop/resume path is a no-op; the test
30 only exercises the pre-transfer ungroup branch.
31 """
32 source_queue = MagicMock()
33 source_queue.state = PlaybackState.IDLE
34 source_queue.resume_pos = 0
35 source_queue.elapsed_time = 0
36 source_queue.current_index = 0
37 source_queue.current_item = None
38 target_queue = MagicMock()
39
40 fake = MagicMock()
41 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == source_id else target_queue)
42 fake.stop = AsyncMock()
43 fake.load = AsyncMock()
44 fake.resume = AsyncMock()
45 fake.clear = MagicMock()
46 fake.update_items = MagicMock()
47 fake._notify_audio_source_transferred = AsyncMock()
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 target_shuffle_enabled: bool = False,
91 source_is_dynamic: bool = False,
92) -> MagicMock:
93 """
94 Build a controller stand-in whose two queues carry real state records.
95
96 The shuffle flag lives on the per-queue state record, so unlike the ungroup tests above these
97 need real queues rather than one shared mock.
98
99 :param source_shuffle_enabled: Whether shuffle is on for the queue being handed over.
100 :param target_shuffle_enabled: Whether shuffle is on for the queue being handed to.
101 :param source_is_dynamic: Whether the source queue is managed by a dynamic source.
102 """
103 source_queue = PlayerQueue(
104 queue_id="src",
105 active=True,
106 display_name="Src",
107 available=True,
108 items=0,
109 shuffle_enabled=source_shuffle_enabled,
110 smart_shuffle_active=source_is_dynamic,
111 is_dynamic=source_is_dynamic,
112 )
113 target_queue = PlayerQueue(
114 queue_id="tgt",
115 active=True,
116 display_name="Tgt",
117 available=True,
118 items=0,
119 shuffle_enabled=target_shuffle_enabled,
120 )
121
122 fake = MagicMock()
123 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == "src" else target_queue)
124 fake._queue_data = {
125 "src": PlayerQueueData(queue=source_queue),
126 "tgt": PlayerQueueData(queue=target_queue),
127 }
128 fake.stop = AsyncMock()
129 fake.load = AsyncMock()
130 fake.resume = AsyncMock()
131 fake._clear = MagicMock()
132 fake.update_items = MagicMock()
133 fake._notify_audio_source_transferred = AsyncMock()
134 fake.is_smart_shuffle_active = MagicMock(side_effect=lambda queue: queue.is_dynamic)
135 target_player = MagicMock()
136 target_player.state.synced_to = None
137 target_player.state.active_group = None
138 fake.mass.players.get_player = MagicMock(return_value=target_player)
139 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
140 return fake
141
142
143async def test_transfer_queue_overwrites_the_targets_own_shuffle() -> None:
144 """The queue brings its own shuffle state, so the target's previous one does not survive."""
145 fake = _shuffle_controller(source_shuffle_enabled=False, target_shuffle_enabled=True)
146
147 await PlayerQueuesController.transfer_queue(
148 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
149 )
150
151 assert fake.get("tgt").shuffle_enabled is False
152
153
154async def test_transfer_queue_carries_the_source_shuffle() -> None:
155 """A shuffled queue handed to another player stays shuffled there."""
156 fake = _shuffle_controller(source_shuffle_enabled=True)
157
158 await PlayerQueuesController.transfer_queue(
159 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
160 )
161
162 assert fake.get("tgt").shuffle_enabled is True
163
164
165async def test_transfer_queue_drops_dynamic_shuffle_from_source() -> None:
166 """The shuffle imposed by a dynamic source follows it to the target queue."""
167 fake = _shuffle_controller(source_shuffle_enabled=True, source_is_dynamic=True)
168 fake._clear.side_effect = lambda queue_id, skip_stop=False: PlayerQueuesController._clear(
169 cast("PlayerQueuesController", fake), queue_id, skip_stop
170 )
171
172 await PlayerQueuesController.transfer_queue(
173 cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
174 )
175
176 assert fake.get("tgt").is_dynamic is True
177 assert fake.get("tgt").shuffle_enabled is True
178 assert fake.get("src").is_dynamic is False
179 assert fake.get("src").shuffle_enabled is False
180 assert fake.get("src").smart_shuffle_active is False
181