/
/
1"""Tests for PlayerQueuesController.transfer_queue protocol/ungroup handling."""
2
3from __future__ import annotations
4
5from typing import cast
6from unittest.mock import AsyncMock, MagicMock
7
8from music_assistant_models.enums import PlaybackState
9
10from music_assistant.controllers.player_queues import PlayerQueuesController
11
12
13class _DummyACM:
14 """Minimal async context manager stand-in for wait_for_player_update."""
15
16 async def __aenter__(self) -> None:
17 return None
18
19 async def __aexit__(self, *args: object) -> bool:
20 return False
21
22
23def _fake_controller(source_id: str, target_player: MagicMock) -> MagicMock:
24 """
25 Build a MagicMock standing in for the controller for transfer_queue tests.
26
27 The source queue is idle so the capture/stop/resume path is a no-op; the test
28 only exercises the pre-transfer ungroup branch.
29 """
30 source_queue = MagicMock()
31 source_queue.state = PlaybackState.IDLE
32 source_queue.resume_pos = 0
33 source_queue.elapsed_time = 0
34 source_queue.current_index = 0
35 source_queue.current_item = None
36 target_queue = MagicMock()
37
38 fake = MagicMock()
39 fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == source_id else target_queue)
40 fake.stop = AsyncMock()
41 fake.load = AsyncMock()
42 fake.resume = AsyncMock()
43 fake.clear = MagicMock()
44 fake.update_items = MagicMock()
45 fake.mass.players.get_player = MagicMock(return_value=target_player)
46 fake.mass.players.cmd_ungroup = AsyncMock()
47 fake.mass.players.wait_for_player_update = MagicMock(return_value=_DummyACM())
48 fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
49 return fake
50
51
52async def test_transfer_queue_ad_hoc_member_ungroups_target_not_leader() -> None:
53 """
54 Transferring onto an ad-hoc sync member frees the target itself, not its leader.
55
56 Ungrouping the leader would transfer leadership to a remaining member and
57 recurse back into transfer_queue, so only the target may be ungrouped.
58 """
59 target_player = MagicMock()
60 target_player.state.synced_to = "leaderA"
61 target_player.state.active_group = None
62 fake = _fake_controller("src", target_player)
63
64 await PlayerQueuesController.transfer_queue(
65 cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
66 )
67
68 fake.mass.players.cmd_ungroup.assert_awaited_once_with("memberB")
69
70
71async def test_transfer_queue_group_member_ungroups_group() -> None:
72 """Transferring onto a virtual-group member releases the group player itself."""
73 target_player = MagicMock()
74 target_player.state.synced_to = None
75 target_player.state.active_group = "groupP"
76 fake = _fake_controller("src", target_player)
77
78 await PlayerQueuesController.transfer_queue(
79 cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
80 )
81
82 fake.mass.players.cmd_ungroup.assert_awaited_once_with("groupP")
83