music-assistant-server

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
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    _register_players(mass, player)
103    return player
104
105
106def _make_universal_player(mass: MagicMock, protocol_player_ids: list[str]) -> UniversalPlayer:
107    provider = _make_universal_provider(mass)
108    base_cfg = MagicMock()
109    base_cfg.name = None
110    base_cfg.default_name = "Universal"
111    mass.config.get_base_player_config.return_value = base_cfg
112    player = UniversalPlayer(
113        provider=provider,
114        player_id="up_test",
115        name="Universal",
116        device_info=DeviceInfo(model="Universal Player", manufacturer="Music Assistant"),
117        protocol_player_ids=list(protocol_player_ids),
118    )
119    player._attr_available = True
120    player._attr_type = PlayerType.PLAYER
121    player._cache.clear()
122    player.set_initialized()
123    return player
124
125
126@pytest.fixture
127def setup() -> tuple[UniversalPlayer, MagicMock]:
128    """Universal player with a chromecast running Spotify Connect."""
129    mass = _make_mock_mass()
130    chromecast = _make_chromecast_player(
131        mass,
132        "cc_1",
133        active_source="spotify_connect",
134        features={
135            PlayerFeature.PAUSE,
136            PlayerFeature.SEEK,
137            PlayerFeature.NEXT_PREVIOUS,
138            PlayerFeature.VOLUME_SET,
139            PlayerFeature.VOLUME_MUTE,
140            PlayerFeature.POWER,
141            PlayerFeature.PLAY_MEDIA,
142        },
143    )
144    universal = _make_universal_player(mass, ["cc_1"])
145    return universal, chromecast
146
147
148def test_supported_features_forwards_only_transport(
149    setup: tuple[UniversalPlayer, MagicMock],
150) -> None:
151    """
152    Only transport features are forwarded; volume/mute/power/play_media are not.
153
154    Volume, mute and power resolve to the protocol player via the base Player's
155    *_control logic, so the universal player must not advertise them itself.
156    """
157    universal, _chromecast = setup
158    assert universal.supported_features == {
159        PlayerFeature.PAUSE,
160        PlayerFeature.SEEK,
161        PlayerFeature.NEXT_PREVIOUS,
162    }
163
164
165def test_volume_and_mute_not_captured_as_native(
166    setup: tuple[UniversalPlayer, MagicMock],
167) -> None:
168    """With an external source active, the player does not capture native volume/mute control."""
169    universal, _chromecast = setup
170    assert universal.volume_control != PLAYER_CONTROL_NATIVE
171    assert universal.mute_control != PLAYER_CONTROL_NATIVE
172
173
174async def test_transport_proxied_to_external_source(
175    setup: tuple[UniversalPlayer, MagicMock],
176) -> None:
177    """Transport commands delegate to the active external source protocol player."""
178    universal, chromecast = setup
179    await universal.pause()
180    chromecast.pause.assert_awaited_once_with()
181
182
183def test_no_features_without_external_source() -> None:
184    """Without an active external source the universal player advertises nothing of its own."""
185    mass = _make_mock_mass()
186    _make_chromecast_player(
187        mass,
188        "cc_1",
189        active_source=None,
190        features={PlayerFeature.PAUSE, PlayerFeature.VOLUME_SET},
191    )
192    universal = _make_universal_player(mass, ["cc_1"])
193    assert universal.supported_features == set()
194
195
196def test_native_feature_survives_external_source() -> None:
197    """
198    A protocol-backed subclass keeps its own native features during external playback.
199
200    The neutral base unions this player's native capabilities with the forwardable
201    transport controls of the active external source, so a subclass that owns e.g. native
202    grouping (SET_MEMBERS) still advertises it while a linked protocol plays Spotify.
203    """
204    mass = _make_mock_mass()
205    _make_chromecast_player(
206        mass,
207        "cc_1",
208        active_source="spotify_connect",
209        features={PlayerFeature.PAUSE, PlayerFeature.SEEK, PlayerFeature.VOLUME_SET},
210    )
211    provider = _make_universal_provider(mass)
212    base_cfg = MagicMock()
213    base_cfg.name = None
214    base_cfg.default_name = "Shell"
215    mass.config.get_base_player_config.return_value = base_cfg
216
217    class _NativeShell(ProtocolBackedPlayer):
218        def __init__(self) -> None:
219            super().__init__(provider, "shell_1")
220            self._attr_supported_features = {PlayerFeature.SET_MEMBERS}
221
222        def _backing_protocol_player_ids(self) -> list[str]:
223            return ["cc_1"]
224
225    shell = _NativeShell()
226    shell._cache.clear()
227    shell.set_initialized()
228    # native SET_MEMBERS survives, and the external source's forwardable transport is added
229    assert shell.supported_features == {
230        PlayerFeature.SET_MEMBERS,
231        PlayerFeature.PAUSE,
232        PlayerFeature.SEEK,
233    }
234
235
236def test_setup_needed_when_only_protocol_awaits_setup() -> None:
237    """A wrapper whose only protocol still needs setup reports it, including the reason."""
238    mass = _make_mock_mass()
239    airplay = _make_protocol_player(
240        "ap_1", "airplay", needs_setup=True, setup_reason="pairing_required"
241    )
242    _register_players(mass, airplay)
243    universal = _make_universal_player(mass, ["ap_1"])
244    assert universal.available is False
245    assert universal.needs_setup is True
246    assert universal.setup_reason == "pairing_required"
247
248
249def test_no_setup_needed_while_another_protocol_is_usable() -> None:
250    """A wrapper that can play through another protocol does not ask for setup."""
251    mass = _make_mock_mass()
252    airplay = _make_protocol_player(
253        "ap_1", "airplay", needs_setup=True, setup_reason="pairing_required"
254    )
255    chromecast = _make_protocol_player("cc_1", "chromecast")
256    _register_players(mass, airplay, chromecast)
257    universal = _make_universal_player(mass, ["ap_1", "cc_1"])
258    assert universal.available is True
259    assert universal.needs_setup is False
260    assert universal.setup_reason is None
261