music-assistant-server

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.mass.players.get_player = MagicMock(return_value=target_player)
48    fake.mass.players.cmd_ungroup = AsyncMock()
49    fake.mass.players.wait_for_player_update = MagicMock(return_value=_DummyACM())
50    fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
51    return fake
52
53
54async def test_transfer_queue_ad_hoc_member_ungroups_target_not_leader() -> None:
55    """
56    Transferring onto an ad-hoc sync member frees the target itself, not its leader.
57
58    Ungrouping the leader would transfer leadership to a remaining member and
59    recurse back into transfer_queue, so only the target may be ungrouped.
60    """
61    target_player = MagicMock()
62    target_player.state.synced_to = "leaderA"
63    target_player.state.active_group = None
64    fake = _fake_controller("src", target_player)
65
66    await PlayerQueuesController.transfer_queue(
67        cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
68    )
69
70    fake.mass.players.cmd_ungroup.assert_awaited_once_with("memberB")
71
72
73async def test_transfer_queue_group_member_ungroups_group() -> None:
74    """Transferring onto a virtual-group member releases the group player itself."""
75    target_player = MagicMock()
76    target_player.state.synced_to = None
77    target_player.state.active_group = "groupP"
78    fake = _fake_controller("src", target_player)
79
80    await PlayerQueuesController.transfer_queue(
81        cast("PlayerQueuesController", fake), "src", "memberB", auto_play=False
82    )
83
84    fake.mass.players.cmd_ungroup.assert_awaited_once_with("groupP")
85
86
87def _shuffle_controller(
88    source_shuffle_enabled: bool,
89    target_shuffle_enabled: bool = False,
90    source_is_dynamic: bool = False,
91) -> MagicMock:
92    """
93    Build a controller stand-in whose two queues carry real state records.
94
95    The shuffle flag lives on the per-queue state record, so unlike the ungroup tests above these
96    need real queues rather than one shared mock.
97
98    :param source_shuffle_enabled: Whether shuffle is on for the queue being handed over.
99    :param target_shuffle_enabled: Whether shuffle is on for the queue being handed to.
100    :param source_is_dynamic: Whether the source queue is managed by a dynamic source.
101    """
102    source_queue = PlayerQueue(
103        queue_id="src",
104        active=True,
105        display_name="Src",
106        available=True,
107        items=0,
108        shuffle_enabled=source_shuffle_enabled,
109        smart_shuffle_active=source_is_dynamic,
110        is_dynamic=source_is_dynamic,
111    )
112    target_queue = PlayerQueue(
113        queue_id="tgt",
114        active=True,
115        display_name="Tgt",
116        available=True,
117        items=0,
118        shuffle_enabled=target_shuffle_enabled,
119    )
120
121    fake = MagicMock()
122    fake.get = MagicMock(side_effect=lambda qid: source_queue if qid == "src" else target_queue)
123    fake._queue_data = {
124        "src": PlayerQueueData(queue=source_queue),
125        "tgt": PlayerQueueData(queue=target_queue),
126    }
127    fake.stop = AsyncMock()
128    fake.load = AsyncMock()
129    fake.resume = AsyncMock()
130    fake._clear = MagicMock()
131    fake.update_items = MagicMock()
132    fake.is_smart_shuffle_active = MagicMock(side_effect=lambda queue: queue.is_dynamic)
133    target_player = MagicMock()
134    target_player.state.synced_to = None
135    target_player.state.active_group = None
136    fake.mass.players.get_player = MagicMock(return_value=target_player)
137    fake.mass.streams.is_smart_fades_active = MagicMock(return_value=False)
138    return fake
139
140
141async def test_transfer_queue_overwrites_the_targets_own_shuffle() -> None:
142    """The queue brings its own shuffle state, so the target's previous one does not survive."""
143    fake = _shuffle_controller(source_shuffle_enabled=False, target_shuffle_enabled=True)
144
145    await PlayerQueuesController.transfer_queue(
146        cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
147    )
148
149    assert fake.get("tgt").shuffle_enabled is False
150
151
152async def test_transfer_queue_carries_the_source_shuffle() -> None:
153    """A shuffled queue handed to another player stays shuffled there."""
154    fake = _shuffle_controller(source_shuffle_enabled=True)
155
156    await PlayerQueuesController.transfer_queue(
157        cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
158    )
159
160    assert fake.get("tgt").shuffle_enabled is True
161
162
163async def test_transfer_queue_drops_dynamic_shuffle_from_source() -> None:
164    """The shuffle imposed by a dynamic source follows it to the target queue."""
165    fake = _shuffle_controller(source_shuffle_enabled=True, source_is_dynamic=True)
166    fake._clear.side_effect = lambda queue_id, skip_stop=False: PlayerQueuesController._clear(
167        cast("PlayerQueuesController", fake), queue_id, skip_stop
168    )
169
170    await PlayerQueuesController.transfer_queue(
171        cast("PlayerQueuesController", fake), "src", "tgt", auto_play=False
172    )
173
174    assert fake.get("tgt").is_dynamic is True
175    assert fake.get("tgt").shuffle_enabled is True
176    assert fake.get("src").is_dynamic is False
177    assert fake.get("src").shuffle_enabled is False
178    assert fake.get("src").smart_shuffle_active is False
179