/
/
/
1"""Tests for the universal player."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from unittest.mock import AsyncMock, MagicMock
8
9import pytest
10from music_assistant_models.constants import PLAYER_CONTROL_NATIVE
11from music_assistant_models.enums import PlaybackState, PlayerFeature, PlayerType, RepeatMode
12
13from music_assistant.models.player import DeviceInfo, Player
14from music_assistant.models.protocol_backed_player import ProtocolBackedPlayer
15from music_assistant.providers.universal_player.player import UniversalPlayer
16from music_assistant.providers.universal_player.provider import UniversalPlayerProvider
17
18
19def _make_mock_mass() -> MagicMock:
20 mass = MagicMock()
21 mass.closing = False
22 mass.config = MagicMock()
23 mass.config.get = MagicMock(return_value=[])
24
25 def _get_raw_player_config_value(
26 _player_id: str, key: str, default: object | None = None
27 ) -> object | None:
28 if key == "min_volume":
29 return 0
30 if key == "max_volume":
31 return 100
32 return default
33
34 mass.config.get_raw_player_config_value = MagicMock(side_effect=_get_raw_player_config_value)
35 mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
36 mass.config.set = MagicMock()
37 mass.signal_event = MagicMock()
38 mass.get_providers = MagicMock(return_value=[])
39 return mass
40
41
42def _make_universal_provider(mock_mass: MagicMock) -> UniversalPlayerProvider:
43 manifest = MagicMock()
44 manifest.domain = "universal_player"
45 manifest.name = "Universal Player"
46 provider = UniversalPlayerProvider.__new__(UniversalPlayerProvider)
47 provider.mass = mock_mass
48 provider.manifest = manifest
49 provider.logger = logging.getLogger("test.universal_player")
50 config = MagicMock()
51 config.instance_id = "universal_player"
52 config.name = None
53 provider.config = config
54 provider._lock = asyncio.Lock()
55 return provider
56
57
58def _make_protocol_player(
59 player_id: str,
60 domain: str,
61 *,
62 needs_setup: bool = False,
63 setup_reason: str | None = None,
64) -> MagicMock:
65 """Return a reachable protocol player, optionally still awaiting its setup."""
66 player = MagicMock(spec=Player)
67 player.player_id = player_id
68 player.available = True
69 player.available_for_playback = not needs_setup
70 player.needs_setup = needs_setup
71 player.setup_reason = setup_reason
72 provider = MagicMock()
73 provider.domain = domain
74 player.provider = provider
75 return player
76
77
78def _register_players(mass: MagicMock, *players: MagicMock) -> None:
79 """Make the given players resolvable by id on the mass mock."""
80 registry = {player.player_id: player for player in players}
81 mass.players.get_player = MagicMock(side_effect=lambda pid, *_a, **_kw: registry.get(pid))
82
83
84def _make_chromecast_player(
85 mass: MagicMock,
86 player_id: str,
87 *,
88 active_source: str | None,
89 features: set[PlayerFeature],
90) -> MagicMock:
91 """Return a chromecast-domain protocol player with the given active source and features."""
92 player = _make_protocol_player(player_id, "chromecast")
93 player.active_source = active_source
94 player.playback_state = PlaybackState.PLAYING
95 player.supported_features = features
96 player.play = AsyncMock()
97 player.pause = AsyncMock()
98 player.stop = AsyncMock()
99 player.next_track = AsyncMock()
100 player.previous_track = AsyncMock()
101 player.seek = AsyncMock()
102 player.set_shuffle = AsyncMock()
103 player.set_repeat = AsyncMock()
104 _register_players(mass, player)
105 return player
106
107
108def _make_universal_player(mass: MagicMock, protocol_player_ids: list[str]) -> UniversalPlayer:
109 provider = _make_universal_provider(mass)
110 base_cfg = MagicMock()
111 base_cfg.name = None
112 base_cfg.default_name = "Universal"
113 mass.config.get_base_player_config.return_value = base_cfg
114 player = UniversalPlayer(
115 provider=provider,
116 player_id="up_test",
117 name="Universal",
118 device_info=DeviceInfo(model="Universal Player", manufacturer="Music Assistant"),
119 protocol_player_ids=list(protocol_player_ids),
120 )
121 player._attr_available = True
122 player._attr_type = PlayerType.PLAYER
123 player._cache.clear()
124 player.set_initialized()
125 return player
126
127
128@pytest.fixture
129def setup() -> tuple[UniversalPlayer, MagicMock]:
130 """Universal player with a chromecast running Spotify Connect."""
131 mass = _make_mock_mass()
132 chromecast = _make_chromecast_player(
133 mass,
134 "cc_1",
135 active_source="spotify_connect",
136 features={
137 PlayerFeature.PAUSE,
138 PlayerFeature.SEEK,
139 PlayerFeature.NEXT_PREVIOUS,
140 PlayerFeature.VOLUME_SET,
141 PlayerFeature.VOLUME_MUTE,
142 PlayerFeature.POWER,
143 PlayerFeature.PLAY_MEDIA,
144 },
145 )
146 universal = _make_universal_player(mass, ["cc_1"])
147 return universal, chromecast
148
149
150def test_supported_features_forwards_only_transport(
151 setup: tuple[UniversalPlayer, MagicMock],
152) -> None:
153 """
154 Only transport features are forwarded; volume/mute/power/play_media are not.
155
156 Volume, mute and power resolve to the protocol player via the base Player's
157 *_control logic, so the universal player must not advertise them itself.
158 """
159 universal, _chromecast = setup
160 assert universal.supported_features == {
161 PlayerFeature.PAUSE,
162 PlayerFeature.SEEK,
163 PlayerFeature.NEXT_PREVIOUS,
164 }
165
166
167def test_volume_and_mute_not_captured_as_native(
168 setup: tuple[UniversalPlayer, MagicMock],
169) -> None:
170 """With an external source active, the player does not capture native volume/mute control."""
171 universal, _chromecast = setup
172 assert universal.volume_control != PLAYER_CONTROL_NATIVE
173 assert universal.mute_control != PLAYER_CONTROL_NATIVE
174
175
176async def test_transport_proxied_to_external_source(
177 setup: tuple[UniversalPlayer, MagicMock],
178) -> None:
179 """Transport commands delegate to the active external source protocol player."""
180 universal, chromecast = setup
181 await universal.pause()
182 chromecast.pause.assert_awaited_once_with()
183
184
185async def test_ordering_proxied_to_external_source(
186 setup: tuple[UniversalPlayer, MagicMock],
187) -> None:
188 """
189 Shuffle and repeat reach the protocol player playing the source.
190
191 The source_list is taken from that player too, so the capability it declares and
192 the command that acts on it must arrive at the same place.
193 """
194 universal, chromecast = setup
195
196 await universal.set_shuffle(True)
197 await universal.set_repeat(RepeatMode.ALL)
198
199 chromecast.set_shuffle.assert_awaited_once_with(True)
200 chromecast.set_repeat.assert_awaited_once_with(RepeatMode.ALL)
201
202
203def test_no_features_without_external_source() -> None:
204 """Without an active external source the universal player advertises nothing of its own."""
205 mass = _make_mock_mass()
206 _make_chromecast_player(
207 mass,
208 "cc_1",
209 active_source=None,
210 features={PlayerFeature.PAUSE, PlayerFeature.VOLUME_SET},
211 )
212 universal = _make_universal_player(mass, ["cc_1"])
213 assert universal.supported_features == set()
214
215
216def test_native_feature_survives_external_source() -> None:
217 """
218 A protocol-backed subclass keeps its own native features during external playback.
219
220 The neutral base unions this player's native capabilities with the forwardable
221 transport controls of the active external source, so a subclass that owns e.g. native
222 grouping (SET_MEMBERS) still advertises it while a linked protocol plays Spotify.
223 """
224 mass = _make_mock_mass()
225 _make_chromecast_player(
226 mass,
227 "cc_1",
228 active_source="spotify_connect",
229 features={PlayerFeature.PAUSE, PlayerFeature.SEEK, PlayerFeature.VOLUME_SET},
230 )
231 provider = _make_universal_provider(mass)
232 base_cfg = MagicMock()
233 base_cfg.name = None
234 base_cfg.default_name = "Shell"
235 mass.config.get_base_player_config.return_value = base_cfg
236
237 class _NativeShell(ProtocolBackedPlayer):
238 def __init__(self) -> None:
239 super().__init__(provider, "shell_1")
240 self._attr_supported_features = {PlayerFeature.SET_MEMBERS}
241
242 def _backing_protocol_player_ids(self) -> list[str]:
243 return ["cc_1"]
244
245 shell = _NativeShell()
246 shell._cache.clear()
247 shell.set_initialized()
248 # native SET_MEMBERS survives, and the external source's forwardable transport is added
249 assert shell.supported_features == {
250 PlayerFeature.SET_MEMBERS,
251 PlayerFeature.PAUSE,
252 PlayerFeature.SEEK,
253 }
254
255
256def test_setup_needed_when_only_protocol_awaits_setup() -> None:
257 """A wrapper whose only protocol still needs setup reports it, including the reason."""
258 mass = _make_mock_mass()
259 airplay = _make_protocol_player(
260 "ap_1", "airplay", needs_setup=True, setup_reason="pairing_required"
261 )
262 _register_players(mass, airplay)
263 universal = _make_universal_player(mass, ["ap_1"])
264 assert universal.available is False
265 assert universal.needs_setup is True
266 assert universal.setup_reason == "pairing_required"
267
268
269def test_no_setup_needed_while_another_protocol_is_usable() -> None:
270 """A wrapper that can play through another protocol does not ask for setup."""
271 mass = _make_mock_mass()
272 airplay = _make_protocol_player(
273 "ap_1", "airplay", needs_setup=True, setup_reason="pairing_required"
274 )
275 chromecast = _make_protocol_player("cc_1", "chromecast")
276 _register_players(mass, airplay, chromecast)
277 universal = _make_universal_player(mass, ["ap_1", "cc_1"])
278 assert universal.available is True
279 assert universal.needs_setup is False
280 assert universal.setup_reason is None
281