music-assistant-server

1"""Tests for the universal player."""
2
3from __future__ import annotations
4
5import logging
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.constants import PLAYER_CONTROL_NATIVE
10from music_assistant_models.enums import PlaybackState, PlayerFeature, PlayerType
11
12from music_assistant.models.player import DeviceInfo, Player
13from music_assistant.providers.universal_player.player import UniversalPlayer
14from music_assistant.providers.universal_player.provider import UniversalPlayerProvider
15
16
17def _make_mock_mass() -> MagicMock:
18    mass = MagicMock()
19    mass.closing = False
20    mass.config = MagicMock()
21    mass.config.get = MagicMock(return_value=[])
22
23    def _get_raw_player_config_value(
24        _player_id: str, key: str, default: object | None = None
25    ) -> object | None:
26        if key == "min_volume":
27            return 0
28        if key == "max_volume":
29            return 100
30        return default
31
32    mass.config.get_raw_player_config_value = MagicMock(side_effect=_get_raw_player_config_value)
33    mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
34    mass.config.set = MagicMock()
35    mass.signal_event = MagicMock()
36    mass.get_providers = MagicMock(return_value=[])
37    return mass
38
39
40def _make_universal_provider(mock_mass: MagicMock) -> UniversalPlayerProvider:
41    manifest = MagicMock()
42    manifest.domain = "universal_player"
43    manifest.name = "Universal Player"
44    provider = UniversalPlayerProvider.__new__(UniversalPlayerProvider)
45    provider.mass = mock_mass
46    provider.manifest = manifest
47    provider.logger = logging.getLogger("test.universal_player")
48    config = MagicMock()
49    config.instance_id = "universal_player"
50    config.name = None
51    provider.config = config
52    provider._universal_player_locks = {}
53    return provider
54
55
56def _make_protocol_player(
57    player_id: str,
58    domain: str,
59    *,
60    needs_setup: bool = False,
61    setup_reason: str | None = None,
62) -> MagicMock:
63    """Return a reachable protocol player, optionally still awaiting its setup."""
64    player = MagicMock(spec=Player)
65    player.player_id = player_id
66    player.available = True
67    player.available_for_playback = not needs_setup
68    player.needs_setup = needs_setup
69    player.setup_reason = setup_reason
70    provider = MagicMock()
71    provider.domain = domain
72    player.provider = provider
73    return player
74
75
76def _register_players(mass: MagicMock, *players: MagicMock) -> None:
77    """Make the given players resolvable by id on the mass mock."""
78    registry = {player.player_id: player for player in players}
79    mass.players.get_player = MagicMock(side_effect=lambda pid, *_a, **_kw: registry.get(pid))
80
81
82def _make_chromecast_player(
83    mass: MagicMock,
84    player_id: str,
85    *,
86    active_source: str | None,
87    features: set[PlayerFeature],
88) -> MagicMock:
89    """Return a chromecast-domain protocol player with the given active source and features."""
90    player = _make_protocol_player(player_id, "chromecast")
91    player.active_source = active_source
92    player.playback_state = PlaybackState.PLAYING
93    player.supported_features = features
94    player.play = AsyncMock()
95    player.pause = AsyncMock()
96    player.stop = AsyncMock()
97    player.next_track = AsyncMock()
98    player.previous_track = AsyncMock()
99    player.seek = AsyncMock()
100    _register_players(mass, player)
101    return player
102
103
104def _make_universal_player(mass: MagicMock, protocol_player_ids: list[str]) -> UniversalPlayer:
105    provider = _make_universal_provider(mass)
106    base_cfg = MagicMock()
107    base_cfg.name = None
108    base_cfg.default_name = "Universal"
109    mass.config.get_base_player_config.return_value = base_cfg
110    player = UniversalPlayer(
111        provider=provider,
112        player_id="up_test",
113        name="Universal",
114        device_info=DeviceInfo(model="Universal Player", manufacturer="Music Assistant"),
115        protocol_player_ids=list(protocol_player_ids),
116    )
117    player._attr_available = True
118    player._attr_type = PlayerType.PLAYER
119    player._cache.clear()
120    player.set_initialized()
121    return player
122
123
124@pytest.fixture
125def setup() -> tuple[UniversalPlayer, MagicMock]:
126    """Universal player with a chromecast running Spotify Connect."""
127    mass = _make_mock_mass()
128    chromecast = _make_chromecast_player(
129        mass,
130        "cc_1",
131        active_source="spotify_connect",
132        features={
133            PlayerFeature.PAUSE,
134            PlayerFeature.SEEK,
135            PlayerFeature.NEXT_PREVIOUS,
136            PlayerFeature.VOLUME_SET,
137            PlayerFeature.VOLUME_MUTE,
138            PlayerFeature.POWER,
139            PlayerFeature.PLAY_MEDIA,
140        },
141    )
142    universal = _make_universal_player(mass, ["cc_1"])
143    return universal, chromecast
144
145
146def test_supported_features_forwards_only_transport(
147    setup: tuple[UniversalPlayer, MagicMock],
148) -> None:
149    """
150    Only transport features are forwarded; volume/mute/power/play_media are not.
151
152    Volume, mute and power resolve to the protocol player via the base Player's
153    *_control logic, so the universal player must not advertise them itself.
154    """
155    universal, _chromecast = setup
156    assert universal.supported_features == {
157        PlayerFeature.PAUSE,
158        PlayerFeature.SEEK,
159        PlayerFeature.NEXT_PREVIOUS,
160    }
161
162
163def test_volume_and_mute_not_captured_as_native(
164    setup: tuple[UniversalPlayer, MagicMock],
165) -> None:
166    """With an external source active, the player does not capture native volume/mute control."""
167    universal, _chromecast = setup
168    assert universal.volume_control != PLAYER_CONTROL_NATIVE
169    assert universal.mute_control != PLAYER_CONTROL_NATIVE
170
171
172async def test_transport_proxied_to_external_source(
173    setup: tuple[UniversalPlayer, MagicMock],
174) -> None:
175    """Transport commands delegate to the active external source protocol player."""
176    universal, chromecast = setup
177    await universal.pause()
178    chromecast.pause.assert_awaited_once_with()
179
180
181def test_no_features_without_external_source() -> None:
182    """Without an active external source the universal player advertises nothing of its own."""
183    mass = _make_mock_mass()
184    _make_chromecast_player(
185        mass,
186        "cc_1",
187        active_source=None,
188        features={PlayerFeature.PAUSE, PlayerFeature.VOLUME_SET},
189    )
190    universal = _make_universal_player(mass, ["cc_1"])
191    assert universal.supported_features == set()
192
193
194def test_setup_needed_when_only_protocol_awaits_setup() -> None:
195    """A wrapper whose only protocol still needs setup reports it, including the reason."""
196    mass = _make_mock_mass()
197    airplay = _make_protocol_player(
198        "ap_1", "airplay", needs_setup=True, setup_reason="pairing_required"
199    )
200    _register_players(mass, airplay)
201    universal = _make_universal_player(mass, ["ap_1"])
202    assert universal.available is False
203    assert universal.needs_setup is True
204    assert universal.setup_reason == "pairing_required"
205
206
207def test_no_setup_needed_while_another_protocol_is_usable() -> None:
208    """A wrapper that can play through another protocol does not ask for setup."""
209    mass = _make_mock_mass()
210    airplay = _make_protocol_player(
211        "ap_1", "airplay", needs_setup=True, setup_reason="pairing_required"
212    )
213    chromecast = _make_protocol_player("cc_1", "chromecast")
214    _register_players(mass, airplay, chromecast)
215    universal = _make_universal_player(mass, ["ap_1", "cc_1"])
216    assert universal.available is True
217    assert universal.needs_setup is False
218    assert universal.setup_reason is None
219