/
/
/
1"""Tests for the in-place player refresh and player class selection."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import TYPE_CHECKING, cast
7from unittest.mock import Mock
8
9import pytest
10from music_assistant_models.enums import PlayerType
11
12import music_assistant.providers.sendspin.player as player_module
13import music_assistant.providers.sendspin.provider as provider_module
14from music_assistant.models.player import Player
15from music_assistant.providers.sendspin.player import (
16 SendspinBasePlayer,
17 SendspinPlayer,
18 SendspinSourcePlayer,
19 SendspinVisualizerPlayer,
20)
21from music_assistant.providers.sendspin.provider import SendspinProvider
22
23if TYPE_CHECKING:
24 from aiosendspin.server.client import SendspinClient
25
26 from music_assistant.mass import MusicAssistant
27 from music_assistant.models.player_provider import PlayerProvider
28
29
30def _make_provider(player: SendspinBasePlayer | None) -> SendspinProvider:
31 provider = SendspinProvider.__new__(SendspinProvider)
32 provider.mass = cast(
33 "MusicAssistant",
34 SimpleNamespace(
35 players=SimpleNamespace(get_player=lambda _client_id: player),
36 create_task=Mock(side_effect=AssertionError("refresh must not spawn tasks")),
37 ),
38 )
39 return provider
40
41
42def _mock_player(*, initialized: bool = True) -> Mock:
43 player = Mock(spec=SendspinBasePlayer)
44 player.initialized.is_set.return_value = initialized
45 return player
46
47
48async def test_refresh_player_updates_in_place() -> None:
49 """A refresh re-applies config and state on the registered player object."""
50 player = _mock_player()
51 provider = _make_provider(player)
52 await provider._refresh_player("c")
53 player.on_config_updated.assert_awaited_once()
54 player.update_state.assert_called_once()
55
56
57async def test_refresh_player_ignores_unknown_player() -> None:
58 """A refresh for a client without a registered player is a no-op."""
59 provider = _make_provider(None)
60 await provider._refresh_player("c")
61
62
63async def test_refresh_player_ignores_uninitialized_player() -> None:
64 """A refresh during initial registration is a no-op (registration applies config)."""
65 player = _mock_player(initialized=False)
66 provider = _make_provider(player)
67 await provider._refresh_player("c")
68 player.on_config_updated.assert_not_awaited()
69 player.update_state.assert_not_called()
70
71
72class _StubPlayer:
73 def __init__(
74 self,
75 provider: SendspinProvider,
76 client_id: str,
77 initial_hello: object | None = None,
78 ) -> None:
79 self.player_id = client_id
80 self._attr_type: PlayerType | None = None
81 self._attr_underlying_player_id: str | None = None
82
83 def preserve_control_features_from(self, other: object) -> None:
84 pass
85
86
87def _class_selection_provider(monkeypatch: pytest.MonkeyPatch) -> SendspinProvider:
88 provider = SendspinProvider.__new__(SendspinProvider)
89 provider._bridge_identifiers = {}
90 provider._bridge_player_types = {}
91 provider._bridge_underlying_players = {}
92 provider._bridge_static_delay_defaults = {}
93 monkeypatch.setattr(provider_module, "SendspinPlayer", _StubPlayer)
94 monkeypatch.setattr(provider_module, "SendspinVisualizerPlayer", _StubPlayer)
95 return provider
96
97
98def _client(negotiated_role_ids: list[str]) -> SendspinClient:
99 return cast(
100 "SendspinClient",
101 SimpleNamespace(
102 negotiated_role_ids=negotiated_role_ids,
103 roles_by_family=Mock(side_effect=AssertionError("must select on negotiated roles")),
104 ),
105 )
106
107
108@pytest.mark.parametrize(
109 ("negotiated_role_ids", "expected_type"),
110 [
111 (["player@v1", "metadata@v1"], None),
112 (["metadata@v1"], PlayerType.DISPLAY),
113 (["visualizer@v1"], PlayerType.VISUALIZER),
114 ([], None),
115 ],
116)
117def test_create_player_selects_class_on_negotiated_roles(
118 monkeypatch: pytest.MonkeyPatch,
119 negotiated_role_ids: list[str],
120 expected_type: PlayerType | None,
121) -> None:
122 """The player class/type follows negotiated roles, not the (trust-gated) active roles."""
123 provider = _class_selection_provider(monkeypatch)
124 player = provider._create_player("c", _client(negotiated_role_ids), None)
125 assert cast("_StubPlayer", player)._attr_type == expected_type
126
127
128class _StubSourcePlayer(_StubPlayer):
129 pass
130
131
132def test_create_player_uses_the_capture_only_class_for_source_only_clients(
133 monkeypatch: pytest.MonkeyPatch,
134) -> None:
135 """A source-only client must not become a player that advertises playback."""
136 provider = _class_selection_provider(monkeypatch)
137 monkeypatch.setattr(provider_module, "SendspinSourcePlayer", _StubSourcePlayer)
138 source_only = provider._create_player("c", _client(["source@v1"]), None)
139 assert isinstance(cast("object", source_only), _StubSourcePlayer)
140 # A device that also plays keeps the full player class.
141 combo = provider._create_player("c", _client(["source@v1", "player@v1"]), None)
142 assert not isinstance(cast("object", combo), _StubSourcePlayer)
143
144
145def test_player_classes_declare_their_visibility() -> None:
146 """
147 Sendspin device players are shared devices, so none of them are private.
148
149 Visualizers and lights stay hidden by default but keep offering themselves as
150 group members. Source clients are listed as audio inputs so the device stays
151 discoverable.
152 """
153 assert SendspinVisualizerPlayer._attr_hidden_by_default is True
154 assert SendspinVisualizerPlayer._attr_private is False
155 assert SendspinSourcePlayer._attr_private is False
156 assert SendspinSourcePlayer._attr_hidden_by_default is False
157 assert SendspinSourcePlayer._attr_type is PlayerType.SOURCE
158
159
160def test_player_classes_advertise_grouping_by_role(monkeypatch: pytest.MonkeyPatch) -> None:
161 """Only Sendspin clients with an output role advertise grouping capability."""
162
163 def _init_player(self: Player, provider: PlayerProvider, player_id: str) -> None:
164 self._provider = provider
165 self._player_id = player_id
166 self._attr_can_group_with = set()
167
168 client = SimpleNamespace(info=SimpleNamespace(player_support=None))
169 provider = cast(
170 "SendspinProvider",
171 SimpleNamespace(
172 instance_id="sendspin--test",
173 logger=Mock(),
174 server_api=SimpleNamespace(get_client=Mock(return_value=client)),
175 ),
176 )
177 monkeypatch.setattr(Player, "__init__", _init_player)
178 monkeypatch.setattr(SendspinBasePlayer, "_refresh_client_info", Mock())
179 monkeypatch.setattr(SendspinBasePlayer, "_subscribe_client_callbacks", Mock())
180 monkeypatch.setattr(SendspinPlayer, "_refresh_client_info", Mock())
181 monkeypatch.setattr(SendspinPlayer, "_subscribe_client_callbacks", Mock())
182 monkeypatch.setattr(player_module, "SendspinPlaybackSession", Mock())
183
184 audio_player = SendspinPlayer(provider, "audio")
185 visualizer_player = SendspinVisualizerPlayer(provider, "visualizer")
186 source_player = SendspinSourcePlayer(provider, "source")
187
188 assert audio_player.can_group_with == {provider.instance_id}
189 assert visualizer_player.can_group_with == {provider.instance_id}
190 assert source_player.can_group_with == set()
191