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