/
/
/
1"""Tests for pairing the built-in web player from its own pairing token."""
2
3from __future__ import annotations
4
5import asyncio
6from types import SimpleNamespace
7from typing import TYPE_CHECKING, Any, cast
8
9import pytest
10from aiosendspin.models.types import PairMethod
11from aiosendspin.noise.keys import PSK_SIZE, b64url_encode
12from aiosendspin.noise.pairing import PairingError
13from aiosendspin.noise.pairing_token import PSKPairingToken, encode_token
14from aiosendspin.noise.trust_store import PskCategory
15from music_assistant_models.auth import User, UserRole
16from music_assistant_models.errors import InvalidCommand
17
18import music_assistant.providers.sendspin.provider as provider_module
19from music_assistant.controllers.webserver.helpers.auth_middleware import set_current_user
20from music_assistant.providers.sendspin.player import SendspinBasePlayer
21
22from .test_pin_session import _FakeServerApi, _make_provider
23
24if TYPE_CHECKING:
25 from aiosendspin.server.client import SendspinClient
26
27 from music_assistant.providers.sendspin.provider import SendspinProvider
28
29_CLIENT_ID = b64url_encode(bytes(range(32)))
30_PAIRING_PSK = bytes(range(100, 100 + PSK_SIZE))
31_TOKEN = encode_token(PSKPairingToken(client_id=_CLIENT_ID, pairing_psk=_PAIRING_PSK))
32
33
34class _FakePairingStore:
35 """Pairing store stand-in answering only the pairing-record lookup."""
36
37 def __init__(self, *, paired: bool, record_owner: str | None = None) -> None:
38 self._record = SimpleNamespace(owner=record_owner) if paired else None
39
40 async def record_by_client_id(self, client_id: str) -> Any:
41 return self._record
42
43
44def _user_lookup(*, found: bool) -> Any:
45 """Return an auth get_user stand-in answering as if the account exists or is gone."""
46
47 async def _get_user(user_id: str) -> Any:
48 return SimpleNamespace(user_id=user_id) if found else None
49
50 return _get_user
51
52
53class _WebPlayerServerApi(_FakeServerApi):
54 """_FakeServerApi with the pairing surface the web-player entry point uses."""
55
56 def __init__(
57 self, *, connected: bool = True, paired: bool = False, record_owner: str | None = None
58 ) -> None:
59 super().__init__([], await_pin=False, connected=connected)
60 self.pairing_store = _FakePairingStore(paired=paired, record_owner=record_owner)
61
62
63def _make_web_provider(
64 api: _WebPlayerServerApi,
65 monkeypatch: pytest.MonkeyPatch,
66 *,
67 is_web_player: bool = True,
68 registered: bool = True,
69 initialized: bool = True,
70 psk_category: PskCategory | None = PskCategory.SENTINEL,
71 caller_has_access: bool = True,
72) -> tuple[SendspinProvider, list[str]]:
73 provider, refreshed = _make_provider(api, monkeypatch)
74 cast("Any", provider.mass).webserver = SimpleNamespace(
75 auth=SimpleNamespace(get_user=_user_lookup(found=caller_has_access))
76 )
77 # the provider resolves its translation namespace through the manifest
78 cast("Any", provider).manifest = SimpleNamespace(domain="sendspin")
79 player = SendspinBasePlayer.__new__(SendspinBasePlayer)
80 player.is_web_player = is_web_player
81 security = None if psk_category is None else SimpleNamespace(psk_category=psk_category)
82 player.api = cast("SendspinClient", SimpleNamespace(connection_security=security))
83 # initialized is a read-only property over a private Event, so seed it directly
84 player._Player__initialized = asyncio.Event() # type: ignore[attr-defined]
85 if initialized:
86 player.set_initialized()
87 cast("Any", provider.mass).players = SimpleNamespace(
88 get_player=lambda _client_id: player if registered else None
89 )
90 monkeypatch.setattr(provider_module, "WEB_PLAYER_CONNECT_TIMEOUT", 0.3)
91 return provider, refreshed
92
93
94async def test_pair_web_player_pairs_with_the_token_psk(monkeypatch: pytest.MonkeyPatch) -> None:
95 """The token's PSK is handed to a pairing attempt for the client the token names."""
96 api = _WebPlayerServerApi()
97 provider, refreshed = _make_web_provider(api, monkeypatch)
98 await provider.pair_web_player(_TOKEN)
99 assert [(a.method, a.pairing_psk) for a in api.attempts] == [
100 (PairMethod.PAIRING_PSK, _PAIRING_PSK)
101 ]
102 assert refreshed == [_CLIENT_ID]
103
104
105async def test_pair_web_player_binds_a_guest_pairing_to_the_guest(
106 monkeypatch: pytest.MonkeyPatch,
107) -> None:
108 """A guest's pairing is bound as session-scoped, so it ends with their session."""
109 api = _WebPlayerServerApi()
110 provider, _refreshed = _make_web_provider(api, monkeypatch)
111 set_current_user(User(user_id="g1", username="party_guest", role=UserRole.GUEST))
112 try:
113 await provider.pair_web_player(_TOKEN)
114 finally:
115 set_current_user(None)
116 assert [a.owner for a in api.attempts] == ["guest-g1"]
117
118
119async def test_pair_web_player_binds_a_full_user_pairing_to_their_account(
120 monkeypatch: pytest.MonkeyPatch,
121) -> None:
122 """A logged-in user's pairing is account-bound, so deleting the account removes it."""
123 api = _WebPlayerServerApi()
124 provider, _refreshed = _make_web_provider(api, monkeypatch)
125 set_current_user(User(user_id="u1", username="maxim", role=UserRole.USER))
126 try:
127 await provider.pair_web_player(_TOKEN)
128 finally:
129 set_current_user(None)
130 assert [a.owner for a in api.attempts] == ["user-u1"]
131
132
133async def test_pair_web_player_rejects_a_malformed_token(monkeypatch: pytest.MonkeyPatch) -> None:
134 """A token that does not decode names no client, so it is refused before any lookup."""
135 api = _WebPlayerServerApi()
136 provider, refreshed = _make_web_provider(api, monkeypatch)
137 with pytest.raises(InvalidCommand) as excinfo:
138 await provider.pair_web_player("SP:0NOTATOKEN")
139 assert excinfo.value.translation_key == "pairing_error_token_invalid"
140 assert api.attempts == []
141 assert refreshed == []
142
143
144async def test_pair_web_player_gives_up_on_a_client_that_never_connects(
145 monkeypatch: pytest.MonkeyPatch,
146) -> None:
147 """A client that only left its hello behind is refused rather than pairing blind."""
148 api = _WebPlayerServerApi(connected=False)
149 provider, refreshed = _make_web_provider(api, monkeypatch)
150 with pytest.raises(InvalidCommand, match="did not register"):
151 await provider.pair_web_player(_TOKEN)
152 assert api.attempts == []
153 assert refreshed == []
154
155
156async def test_pair_web_player_waits_for_a_connection_still_landing(
157 monkeypatch: pytest.MonkeyPatch,
158) -> None:
159 """A web player asking before its handshake arrives is paired once the client lands."""
160 api = _WebPlayerServerApi(connected=False)
161 provider, refreshed = _make_web_provider(api, monkeypatch)
162 client = cast("Any", api._client)
163 lookups = 0
164
165 def _get_client(_client_id: str) -> Any:
166 nonlocal lookups
167 lookups += 1
168 client.is_connected = lookups > 1
169 return client
170
171 monkeypatch.setattr(api, "get_client", _get_client)
172 await provider.pair_web_player(_TOKEN)
173 assert lookups > 1
174 assert [a.pairing_psk for a in api.attempts] == [_PAIRING_PSK]
175 assert refreshed == [_CLIENT_ID]
176
177
178async def test_pair_web_player_refuses_a_non_web_player(monkeypatch: pytest.MonkeyPatch) -> None:
179 """A client whose hello does not classify it as a web player keeps the operator gesture."""
180 api = _WebPlayerServerApi()
181 provider, refreshed = _make_web_provider(api, monkeypatch, is_web_player=False)
182 with pytest.raises(InvalidCommand, match="not a built-in web player"):
183 await provider.pair_web_player(_TOKEN)
184 assert api.attempts == []
185 assert refreshed == []
186
187
188async def test_pair_web_player_refuses_an_unregistered_client(
189 monkeypatch: pytest.MonkeyPatch,
190) -> None:
191 """A connected client with no Sendspin player behind it is never paired."""
192 api = _WebPlayerServerApi()
193 provider, refreshed = _make_web_provider(api, monkeypatch, registered=False)
194 with pytest.raises(InvalidCommand, match="did not register"):
195 await provider.pair_web_player(_TOKEN)
196 assert api.attempts == []
197 assert refreshed == []
198
199
200async def test_pair_web_player_waits_out_a_half_registered_player(
201 monkeypatch: pytest.MonkeyPatch,
202) -> None:
203 """A player still registering is not paired, so the pairing refresh cannot be dropped."""
204 api = _WebPlayerServerApi()
205 provider, refreshed = _make_web_provider(api, monkeypatch, initialized=False)
206 with pytest.raises(InvalidCommand, match="did not register"):
207 await provider.pair_web_player(_TOKEN)
208 assert api.attempts == []
209 assert refreshed == []
210
211
212async def test_pair_web_player_skips_an_unencrypted_client(
213 monkeypatch: pytest.MonkeyPatch,
214) -> None:
215 """An unencrypted client cannot hold a pairing, so it is left alone instead of failing."""
216 api = _WebPlayerServerApi()
217 provider, refreshed = _make_web_provider(api, monkeypatch, psk_category=None)
218 await provider.pair_web_player(_TOKEN)
219 assert api.attempts == []
220 assert refreshed == []
221
222
223async def test_pair_web_player_reports_a_pairing_failure_without_the_token(
224 monkeypatch: pytest.MonkeyPatch,
225) -> None:
226 """A failed attempt surfaces as a localized error, keeping the token out of the message."""
227 api = _WebPlayerServerApi()
228 provider, _refreshed = _make_web_provider(api, monkeypatch)
229 api.outcomes.append(PairingError("device refused the PSK"))
230 with pytest.raises(InvalidCommand) as excinfo:
231 await provider.pair_web_player(_TOKEN)
232 assert excinfo.value.translation_key == "pairing_error_failed"
233 assert excinfo.value.translation_owner == "provider.sendspin"
234 assert _TOKEN not in str(excinfo.value)
235
236
237async def test_pair_web_player_is_a_no_op_when_already_paired(
238 monkeypatch: pytest.MonkeyPatch,
239) -> None:
240 """A web player calling on every mount does not re-handshake an existing pairing."""
241 api = _WebPlayerServerApi(paired=True)
242 provider, refreshed = _make_web_provider(api, monkeypatch, psk_category=PskCategory.LONG_TERM)
243 await provider.pair_web_player(_TOKEN)
244 assert api.attempts == []
245 assert refreshed == []
246
247
248async def test_pair_web_player_restamps_a_pairing_owned_by_another_user(
249 monkeypatch: pytest.MonkeyPatch,
250) -> None:
251 """A browser keeps its identity across logins, so the pairing follows the current caller."""
252 api = _WebPlayerServerApi(paired=True, record_owner="user-someone-else")
253 provider, refreshed = _make_web_provider(api, monkeypatch, psk_category=PskCategory.LONG_TERM)
254 set_current_user(User(user_id="g1", username="party_guest", role=UserRole.GUEST))
255 try:
256 await provider.pair_web_player(_TOKEN)
257 finally:
258 set_current_user(None)
259 assert [a.owner for a in api.attempts] == ["guest-g1"]
260 assert refreshed == [_CLIENT_ID]
261
262
263async def test_pair_web_player_is_a_no_op_for_the_pairing_owner(
264 monkeypatch: pytest.MonkeyPatch,
265) -> None:
266 """The caller that owns the existing pairing never re-handshakes it."""
267 api = _WebPlayerServerApi(paired=True, record_owner="guest-g1")
268 provider, refreshed = _make_web_provider(api, monkeypatch, psk_category=PskCategory.LONG_TERM)
269 set_current_user(User(user_id="g1", username="party_guest", role=UserRole.GUEST))
270 try:
271 await provider.pair_web_player(_TOKEN)
272 finally:
273 set_current_user(None)
274 assert api.attempts == []
275 assert refreshed == []
276
277
278async def test_pair_web_player_repairs_a_client_that_lost_its_pairing(
279 monkeypatch: pytest.MonkeyPatch,
280) -> None:
281 """A record the client can no longer authenticate is re-paired instead of stranding it."""
282 api = _WebPlayerServerApi(paired=True)
283 provider, refreshed = _make_web_provider(api, monkeypatch, psk_category=PskCategory.SENTINEL)
284 await provider.pair_web_player(_TOKEN)
285 assert [a.pairing_psk for a in api.attempts] == [_PAIRING_PSK]
286 assert refreshed == [_CLIENT_ID]
287
288
289async def test_pair_web_player_withdraws_a_pairing_the_caller_just_lost(
290 monkeypatch: pytest.MonkeyPatch,
291) -> None:
292 """Access withdrawn while the handshake ran leaves no pairing behind."""
293 api = _WebPlayerServerApi()
294 provider, _refreshed = _make_web_provider(api, monkeypatch, caller_has_access=False)
295 evicted: list[str] = []
296
297 async def _record_eviction(owner: str) -> None:
298 evicted.append(owner)
299
300 monkeypatch.setattr(provider, "_evict_pairings_for_owner", _record_eviction)
301 set_current_user(User(user_id="u1", username="maxim", role=UserRole.USER))
302 try:
303 await provider.pair_web_player(_TOKEN)
304 finally:
305 set_current_user(None)
306
307 assert [a.owner for a in api.attempts] == ["user-u1"]
308 assert evicted == ["user-u1"]
309
310
311async def test_pair_web_player_keeps_the_pairing_of_a_live_account(
312 monkeypatch: pytest.MonkeyPatch,
313) -> None:
314 """The revalidation after pairing leaves an unaffected caller's pairing in place."""
315 api = _WebPlayerServerApi()
316 provider, _refreshed = _make_web_provider(api, monkeypatch)
317 evicted: list[str] = []
318
319 async def _record_eviction(owner: str) -> None:
320 evicted.append(owner)
321
322 monkeypatch.setattr(provider, "_evict_pairings_for_owner", _record_eviction)
323 set_current_user(User(user_id="u1", username="maxim", role=UserRole.USER))
324 try:
325 await provider.pair_web_player(_TOKEN)
326 finally:
327 set_current_user(None)
328
329 assert [a.owner for a in api.attempts] == ["user-u1"]
330 assert evicted == []
331