/
/
/
1"""Tests for the Sendspin player interactive pairing setup flow (run_setup_flow)."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from collections import deque
8from types import SimpleNamespace
9from typing import TYPE_CHECKING, Any, cast
10from unittest import mock
11
12import pytest
13from aiosendspin.models.core import PairMethodDescriptor
14from aiosendspin.models.types import PairAbortReason, PairMethod
15from aiosendspin.noise.pairing import RemotePairingAbortError
16from aiosendspin.noise.trust_store import PskCategory
17from music_assistant_models.enums import ConfigEntryType, FlowStepType
18
19from music_assistant.models.setup_flow import (
20 FINISH_STEP_SILENT,
21 AbortFlow,
22 SetupFlowContext,
23 SetupSession,
24 StepExpiredError,
25)
26from music_assistant.providers.sendspin import player as player_module
27from music_assistant.providers.sendspin.constants import (
28 CONF_CONNECT_METHOD,
29 CONF_PAIRING_METHOD,
30 CONF_PAIRING_PIN,
31 CONF_SOURCE_APPROVAL_DISMISSED,
32 CONF_SOURCE_INPUT_ACTION,
33 CONNECT_METHOD_PAIR,
34 CONNECT_METHOD_UNPAIRED,
35 PAIR_METHOD_DYNAMIC_PIN,
36 PAIR_METHOD_PIN,
37 PAIR_METHOD_STATIC_PIN,
38 SOURCE_INPUT_DISMISS,
39 SOURCE_INPUT_PAIR,
40)
41from music_assistant.providers.sendspin.helpers import SecurityActionError
42from music_assistant.providers.sendspin.player import SendspinBasePlayer
43from tests.common import collect_loop_errors
44
45if TYPE_CHECKING:
46 from aiosendspin.server.client import SendspinClient
47 from music_assistant_models.setup_flow import SetupFlowStep
48
49 from music_assistant.providers.sendspin.provider import SendspinProvider
50
51
52def _desc(
53 method: PairMethod,
54 *,
55 locations: list[str] | None = None,
56 out_channels: list[str] | None = None,
57) -> PairMethodDescriptor:
58 return PairMethodDescriptor(method=method, locations=locations, out_channels=out_channels)
59
60
61class _FakePinSession:
62 """Minimal PinPairingSession stand-in the fake provider hands back to the flow."""
63
64 def __init__(
65 self,
66 *,
67 awaiting_gesture: bool = False,
68 verify: bool = False,
69 method: PairMethod = PairMethod.DYNAMIC_PIN,
70 ) -> None:
71 self.pin_request_event = asyncio.Event()
72 self.gesture_event = asyncio.Event()
73 if awaiting_gesture:
74 # A gesture-gated device reports pair-pending before asking for the PIN.
75 self.gesture_event.set()
76 else:
77 self.pin_request_event.set()
78 self.awaiting_pin = True
79 self.finished = False
80 self.error: Exception | None = None
81 self.can_retry = False
82 self.verify = verify
83 self.method = method
84 self.pin_length: int | None = 6 if method is PairMethod.DYNAMIC_PIN else None
85 # None so the flow's post-submit "confirming" wait is skipped in tests.
86 self.task: asyncio.Task[None] | None = None
87
88 @property
89 def awaiting_first_message(self) -> bool:
90 return not self.gesture_event.is_set() and not self.pin_request_event.is_set()
91
92 @property
93 def awaiting_gesture(self) -> bool:
94 return self.gesture_event.is_set() and not self.pin_request_event.is_set()
95
96 async def wait_first_message(self) -> None:
97 await self.pin_request_event.wait()
98
99 async def wait_pin_request(self) -> None:
100 await self.pin_request_event.wait()
101
102
103class _FakeApi:
104 """Fake SendspinClient exposing just what the flow reads (hello + security + roles)."""
105
106 def __init__(
107 self,
108 methods: list[PairMethodDescriptor],
109 *,
110 active_roles: tuple[str, ...] = (),
111 psk_category: PskCategory = PskCategory.SENTINEL,
112 unpaired_access: bool = False,
113 ):
114 self.info_or_none = SimpleNamespace(
115 supported_pair_methods=list(methods),
116 unpaired_access=SimpleNamespace(enabled=unpaired_access),
117 )
118 self.connection_security: Any = SimpleNamespace(psk_category=psk_category)
119 self.active_roles = active_roles
120 self.negotiated_role_ids: list[str] = []
121
122 def roles_by_family(self, family: str) -> list[str]:
123 return [role for role in self.active_roles if role.startswith(f"{family}@")]
124
125
126class _FakePairingStore:
127 """Pairing-store stand-in serving a scripted record and unpaired-trust state."""
128
129 def __init__(self, record: Any = None, trusted: Any = None) -> None:
130 self.record = record
131 self.trusted = trusted
132
133 async def record_by_client_id(self, client_id: str) -> Any:
134 return self.record
135
136 async def trusted_unpaired(self, client_id: str) -> Any:
137 return self.trusted
138
139
140class _FakeProvider:
141 """Scripts the pairing primitives run_setup_flow drives, recording every call."""
142
143 def __init__(
144 self,
145 api: _FakeApi,
146 *,
147 gesture: bool = False,
148 submit_outcomes: list[str] | None = None,
149 token_errors: list[Exception] | None = None,
150 record: Any = None,
151 trusted: Any = None,
152 ) -> None:
153 self.api = api
154 self.server_api = SimpleNamespace(pairing_store=_FakePairingStore(record, trusted))
155 self.session: _FakePinSession | None = None
156 self.start_calls = 0
157 self.static: bool | None = None
158 self.verify: bool | None = None
159 self.submitted_pins: list[str] = []
160 self.tokens: list[str] = []
161 self.cancel_calls = 0
162 self.clear_calls = 0
163 self.trust_calls: list[bool] = []
164 self._gesture = gesture
165 self._submit_outcomes = deque(submit_outcomes or [])
166 self._token_errors = deque(token_errors or [])
167
168 def pairing_config_snapshot(self, client_id: str) -> None:
169 return None
170
171 def get_pin_session(self, client_id: str) -> _FakePinSession | None:
172 return self.session
173
174 def clear_pin_session(self, client_id: str) -> None:
175 self.clear_calls += 1
176 if self.session is not None and self.session.finished:
177 self.session = None
178
179 async def start_pin_pairing(
180 self, client_id: str, *, verify: bool = False, static: bool = False
181 ) -> _FakePinSession:
182 self.start_calls += 1
183 self.static = static
184 self.verify = verify
185 if self.session is not None and self.session.can_retry:
186 # A retryable session resumes in place, past the gesture, awaiting a PIN again.
187 self.session.can_retry = False
188 self.session.error = None
189 self.session.awaiting_pin = True
190 self.session.pin_request_event.set()
191 return self.session
192 offered = {d.method for d in self.api.info_or_none.supported_pair_methods}
193 dynamic_offered = PairMethod.DYNAMIC_PIN in offered
194 self.session = _FakePinSession(
195 awaiting_gesture=self._gesture,
196 verify=verify,
197 method=(
198 PairMethod.DYNAMIC_PIN if dynamic_offered and not static else PairMethod.STATIC_PIN
199 ),
200 )
201 return self.session
202
203 def submit_pin(self, client_id: str, pin: str) -> None:
204 assert self.session is not None
205 self.submitted_pins.append(pin)
206 outcome = self._submit_outcomes.popleft() if self._submit_outcomes else "success"
207 if outcome == "success":
208 self.session.finished = True
209 self.session.error = None
210 self.session.awaiting_pin = False
211 self.api.active_roles = ("player",)
212 elif outcome == "retry":
213 self.session.can_retry = True
214 self.session.error = RemotePairingAbortError(PairAbortReason.PIN_MISMATCH)
215 elif outcome == "session_lost":
216 self.session = None
217 raise SecurityActionError("pairing_error_no_pin_session")
218
219 async def cancel_pin_pairing(self, client_id: str) -> None:
220 self.cancel_calls += 1
221 self.session = None
222
223 async def set_trusted_unpaired(self, client_id: str, enabled: bool) -> None:
224 self.trust_calls.append(enabled)
225 if enabled:
226 self.server_api.pairing_store.trusted = object()
227
228 async def pair_with_token(self, client_id: str, token: str) -> None:
229 self.tokens.append(token)
230 if self._token_errors:
231 raise self._token_errors.popleft()
232 self.api.active_roles = ("player",)
233
234
235def _make_player(api: _FakeApi, provider: _FakeProvider) -> SendspinBasePlayer:
236 player = SendspinBasePlayer.__new__(SendspinBasePlayer)
237 player._player_id = "client-1"
238 player._provider = cast("SendspinProvider", provider)
239 player.api = cast("SendspinClient", api)
240 return player
241
242
243def _make_session(finish_handler: Any) -> tuple[SetupSession, mock.Mock]:
244 mass = mock.Mock()
245 context = SetupFlowContext(kind="setup", reason="user", domain="sendspin", player_id="client-1")
246 return SetupSession(mass, "flow-test", context, finish_handler), mass
247
248
249async def _ok_finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
250 """Finish handler stand-in that accepts any values and reports the player id."""
251 return {"player_id": "client-1"}
252
253
254def _published_steps(mass: mock.Mock) -> list[Any]:
255 return [call.kwargs["data"] for call in mass.signal_event.call_args_list]
256
257
258async def _wait_for(predicate: Any, timeout: float = 5.0) -> Any:
259 deadline = time.monotonic() + timeout
260 while time.monotonic() < deadline:
261 if result := predicate():
262 return result
263 await asyncio.sleep(0.01)
264 raise AssertionError("condition not met within timeout")
265
266
267async def _wait_step(
268 session: SetupSession,
269 *,
270 step_type: FlowStepType | None = None,
271 step_id: str | None = None,
272 with_errors: bool = False,
273) -> SetupFlowStep:
274 def _match() -> SetupFlowStep | None:
275 step = session.current_step
276 if step is None:
277 return None
278 if step_type is not None and step.type != step_type:
279 return None
280 if step_id is not None and step.step_id != step_id:
281 return None
282 if with_errors and not step.errors:
283 return None
284 return step
285
286 return cast("SetupFlowStep", await _wait_for(_match))
287
288
289async def test_select_method_pin_gesture_submit_success() -> None:
290 """Select PIN, wait through the gesture, submit the PIN, succeed, and finish with {}."""
291 collected: dict[str, Any] = {}
292
293 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
294 collected["values"] = values
295 return {"player_id": "client-1"}
296
297 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.STATIC_PIN)])
298 provider = _FakeProvider(api, gesture=True)
299 session, mass = _make_session(finish)
300 player = _make_player(api, provider)
301
302 task = asyncio.create_task(player.run_setup_flow(session))
303 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="select_method")
304 assert {option.value for option in step.entries[0].options} == {
305 PAIR_METHOD_DYNAMIC_PIN,
306 PAIR_METHOD_STATIC_PIN,
307 }
308 # the method is rendered as an expanded (radio) list with nothing preselected
309 assert step.entries[0].expanded_options is True
310 assert step.entries[0].default_value is None
311 session.handle_submit({CONF_PAIRING_METHOD: PAIR_METHOD_DYNAMIC_PIN})
312
313 await _wait_step(session, step_type=FlowStepType.PROGRESS, step_id="awaiting_gesture")
314 assert provider.session is not None
315 provider.session.pin_request_event.set()
316
317 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
318 session.handle_submit({CONF_PAIRING_PIN: "123456"})
319
320 await _wait_for(lambda: session.finished)
321 await task
322
323 assert collected["values"] == {}
324 assert provider.submitted_pins == ["123456"]
325 assert provider.static is False
326 assert provider.verify is False
327 assert provider.cancel_calls == 0
328 assert provider.clear_calls == 1
329 steps = _published_steps(mass)
330 assert [s.step_id for s in steps if s.type == FlowStepType.PROGRESS] == ["awaiting_gesture"]
331 assert steps[-1].type == FlowStepType.FINISH
332
333
334async def test_confirming_wait_failure_after_deadline_logs_no_loop_error(
335 monkeypatch: pytest.MonkeyPatch,
336) -> None:
337 """A pairing attempt failing after the confirming step expired is not reported to the loop."""
338 release = asyncio.Event()
339
340 async def _failing_attempt() -> None:
341 await release.wait()
342 raise RuntimeError("refreshing the player failed")
343
344 monkeypatch.setattr(player_module, "PAIR_CONFIRM_TIMEOUT", 0.01)
345 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
346 provider = _FakeProvider(api)
347 session, _mass = _make_session(_ok_finish)
348 player = _make_player(api, provider)
349
350 with collect_loop_errors() as reported:
351 flow = asyncio.create_task(player.run_setup_flow(session))
352 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
353 assert provider.session is not None
354 attempt = asyncio.create_task(_failing_attempt())
355 provider.session.task = attempt
356 session.handle_submit({CONF_PAIRING_PIN: "123456"})
357
358 # let the attempt fail only once the confirming step has expired and the flow has
359 # moved on, so the failure reliably lands after the flow stopped waiting for it
360 await _wait_for(lambda: session.finished)
361 await flow
362 release.set()
363 with pytest.raises(RuntimeError, match="refreshing the player failed"):
364 await attempt
365
366 assert reported == []
367
368
369async def test_single_pin_method_skips_select() -> None:
370 """A device offering one usable PIN method goes straight to the PIN form, no method select."""
371 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
372 provider = _FakeProvider(api)
373 session, mass = _make_session(_ok_finish)
374 player = _make_player(api, provider)
375
376 task = asyncio.create_task(player.run_setup_flow(session))
377 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
378 assert not any(s.step_id == "select_method" for s in _published_steps(mass))
379 session.handle_submit({CONF_PAIRING_PIN: "123456"})
380 await _wait_for(lambda: session.finished)
381 await task
382 assert provider.submitted_pins == ["123456"]
383
384
385async def test_pin_mismatch_retries_in_place_then_succeeds() -> None:
386 """A mismatch re-renders the PIN form with a base error; the retry resumes and succeeds."""
387 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
388 provider = _FakeProvider(api, submit_outcomes=["retry", "success"])
389 session, _mass = _make_session(_ok_finish)
390 player = _make_player(api, provider)
391
392 task = asyncio.create_task(player.run_setup_flow(session))
393 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
394 session.handle_submit({CONF_PAIRING_PIN: "000000"})
395
396 error_step = await _wait_step(
397 session, step_type=FlowStepType.FORM, step_id="enter_pin", with_errors=True
398 )
399 assert error_step.errors == {"base": "pairing_error_pin_mismatch"}
400 session.handle_submit({CONF_PAIRING_PIN: "123456"})
401
402 await _wait_for(lambda: session.finished)
403 await task
404 assert provider.submitted_pins == ["000000", "123456"]
405 assert provider.start_calls == 2
406 assert provider.cancel_calls == 0
407
408
409async def test_trusted_unpaired_pin_mismatch_still_retries() -> None:
410 """With unpaired access already trusted, a mismatch must not be misreported as success."""
411 api = _FakeApi(
412 [_desc(PairMethod.DYNAMIC_PIN)],
413 active_roles=("player",),
414 unpaired_access=True,
415 )
416 provider = _FakeProvider(api, submit_outcomes=["retry", "success"], trusted=object())
417 session, mass = _make_session(_ok_finish)
418 player = _make_player(api, provider)
419
420 task = asyncio.create_task(player.run_setup_flow(session))
421 # trusted-unpaired devices are not offered the unpaired option again
422 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
423 assert not any(s.step_id == "select_method" for s in _published_steps(mass))
424 session.handle_submit({CONF_PAIRING_PIN: "000000"})
425
426 error_step = await _wait_step(
427 session, step_type=FlowStepType.FORM, step_id="enter_pin", with_errors=True
428 )
429 assert error_step.errors == {"base": "pairing_error_pin_mismatch"}
430 session.handle_submit({CONF_PAIRING_PIN: "123456"})
431
432 await _wait_for(lambda: session.finished)
433 await task
434 assert provider.submitted_pins == ["000000", "123456"]
435 assert provider.trust_calls == []
436
437
438async def test_consent_step_grants_trust() -> None:
439 """Submitting the consent step without opting into pairing allows unpaired playback."""
440 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)], unpaired_access=True)
441 provider = _FakeProvider(api)
442 session, _mass = _make_session(_ok_finish)
443 player = _make_player(api, provider)
444
445 task = asyncio.create_task(player.run_setup_flow(session))
446 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="approve_device")
447 assert [entry.key for entry in step.entries] == [CONF_CONNECT_METHOD]
448 session.handle_submit({CONF_CONNECT_METHOD: CONNECT_METHOD_UNPAIRED})
449
450 await _wait_for(lambda: session.finished)
451 await task
452 assert provider.trust_calls == [True]
453 assert provider.start_calls == 0
454 assert provider.tokens == []
455 # a one-click allow closes the dialog without a success screen
456 assert session.finish_step_id == FINISH_STEP_SILENT
457
458
459async def test_consent_without_pair_methods_still_asks() -> None:
460 """The unpaired grant is never automatic: a device without pair methods still asks."""
461 api = _FakeApi([], unpaired_access=True)
462 provider = _FakeProvider(api)
463 session, _mass = _make_session(_ok_finish)
464 player = _make_player(api, provider)
465
466 task = asyncio.create_task(player.run_setup_flow(session))
467 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="approve_device")
468 assert step.entries == []
469 session.handle_submit({})
470
471 await _wait_for(lambda: session.finished)
472 await task
473 assert provider.trust_calls == [True]
474
475
476async def test_consent_on_combo_declines_the_input_in_one_click() -> None:
477 """A plain allow on a combo also declines the pending audio input, one submit total."""
478 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)], unpaired_access=True)
479 api.negotiated_role_ids = ["player@v1", "source@v1"]
480 provider = _FakeProvider(api)
481 session, _mass = _make_session(_ok_finish)
482 player = _make_player(api, provider)
483 mass = _attach_mass(player)
484
485 task = asyncio.create_task(player.run_setup_flow(session))
486 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="approve_device_source")
487 assert [entry.key for entry in step.entries] == [CONF_CONNECT_METHOD]
488 assert step.last_step is True
489 session.handle_submit({CONF_CONNECT_METHOD: CONNECT_METHOD_UNPAIRED})
490
491 await _wait_for(lambda: session.finished)
492 await task
493 assert provider.trust_calls == [True]
494 mass.config.set_raw_player_config_value.assert_called_once_with(
495 "client-1", CONF_SOURCE_APPROVAL_DISMISSED, True
496 )
497
498
499async def test_consent_opting_into_pairing_pairs_instead() -> None:
500 """Ticking the pairing opt-in continues into the pair-method selection, granting nothing."""
501 api = _FakeApi(
502 [_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.STATIC_PIN)], unpaired_access=True
503 )
504 provider = _FakeProvider(api)
505 session, _mass = _make_session(_ok_finish)
506 player = _make_player(api, provider)
507
508 task = asyncio.create_task(player.run_setup_flow(session))
509 await _wait_step(session, step_type=FlowStepType.FORM, step_id="approve_device")
510 session.handle_submit({CONF_CONNECT_METHOD: CONNECT_METHOD_PAIR})
511
512 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="select_method")
513 assert {option.value for option in step.entries[0].options} == {
514 PAIR_METHOD_DYNAMIC_PIN,
515 PAIR_METHOD_STATIC_PIN,
516 }
517 session.handle_submit({CONF_PAIRING_METHOD: PAIR_METHOD_DYNAMIC_PIN})
518
519 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
520 session.handle_submit({CONF_PAIRING_PIN: "123456"})
521
522 await _wait_for(lambda: session.finished)
523 await task
524 assert provider.submitted_pins == ["123456"]
525 assert provider.trust_calls == []
526
527
528def _attach_mass(player: SendspinBasePlayer, *, dismissed: bool = False) -> mock.MagicMock:
529 """Give a bare test player the mass surface the approval paths touch."""
530 player.mass = mock.MagicMock()
531 player.mass.config.get_raw_player_config_value = mock.Mock(return_value=dismissed)
532 player.mass.config.save_player_config = mock.AsyncMock()
533 player.update_state = mock.Mock() # type: ignore[method-assign, misc]
534 return player.mass
535
536
537def _combo_api_with_pending_source() -> _FakeApi:
538 api = _FakeApi(
539 [_desc(PairMethod.DYNAMIC_PIN)], active_roles=("player@v1",), unpaired_access=True
540 )
541 api.negotiated_role_ids = ["player@v1", "source@v1"]
542 return api
543
544
545async def test_guest_device_with_an_input_consents_and_keeps_guest_access() -> None:
546 """
547 A guest device with a pending input consents on the approval step, not the input picker.
548
549 Guest access already carries playback, so the only choice left is the optional upgrade
550 to a pairing; finishing keeps guest access and leaves the input off.
551 """
552 api = _combo_api_with_pending_source()
553 provider = _FakeProvider(api)
554 session, _mass = _make_session(_ok_finish)
555 player = _make_player(api, provider)
556 mass = _attach_mass(player)
557
558 task = asyncio.create_task(player.run_setup_flow(session))
559 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="approve_device_source")
560 assert [entry.key for entry in step.entries] == [CONF_CONNECT_METHOD]
561 assert step.last_step is True
562 session.handle_submit({CONF_CONNECT_METHOD: CONNECT_METHOD_UNPAIRED})
563
564 await _wait_for(lambda: session.finished)
565 await task
566 mass.config.set_raw_player_config_value.assert_called_once_with(
567 "client-1", CONF_SOURCE_APPROVAL_DISMISSED, True
568 )
569 assert provider.trust_calls == [True]
570 assert provider.start_calls == 0
571 assert session.finish_step_id == FINISH_STEP_SILENT
572
573
574async def test_input_picker_serves_a_device_that_withdrew_guest_access() -> None:
575 """Without guest access on offer, a pending input still gets the pair-or-decline picker."""
576 api = _combo_api_with_pending_source()
577 api.info_or_none.unpaired_access = SimpleNamespace(enabled=False)
578 provider = _FakeProvider(api)
579 session, _mass = _make_session(_ok_finish)
580 player = _make_player(api, provider)
581 mass = _attach_mass(player)
582
583 task = asyncio.create_task(player.run_setup_flow(session))
584 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="source_input")
585 assert [option.value for option in step.entries[0].options] == [
586 SOURCE_INPUT_PAIR,
587 SOURCE_INPUT_DISMISS,
588 ]
589 session.handle_submit({CONF_SOURCE_INPUT_ACTION: SOURCE_INPUT_DISMISS})
590
591 await _wait_for(lambda: session.finished)
592 await task
593 mass.config.set_raw_player_config_value.assert_called_once_with(
594 "client-1", CONF_SOURCE_APPROVAL_DISMISSED, True
595 )
596 assert provider.trust_calls == []
597 assert session.finish_step_id == FINISH_STEP_SILENT
598
599
600async def test_opting_into_pairing_for_the_input_offers_only_pair_methods() -> None:
601 """Ticking the pairing box on the approval step never re-offers unpaired access or ignore."""
602 api = _combo_api_with_pending_source()
603 api.info_or_none.supported_pair_methods = [
604 _desc(PairMethod.DYNAMIC_PIN),
605 _desc(PairMethod.STATIC_PIN),
606 ]
607 provider = _FakeProvider(api)
608 session, _mass = _make_session(_ok_finish)
609 player = _make_player(api, provider)
610 _attach_mass(player)
611
612 task = asyncio.create_task(player.run_setup_flow(session))
613 await _wait_step(session, step_type=FlowStepType.FORM, step_id="approve_device_source")
614 session.handle_submit({CONF_CONNECT_METHOD: CONNECT_METHOD_PAIR})
615
616 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="select_method")
617 assert {option.value for option in step.entries[0].options} == {
618 PAIR_METHOD_DYNAMIC_PIN,
619 PAIR_METHOD_STATIC_PIN,
620 }
621 session.handle_submit({CONF_PAIRING_METHOD: PAIR_METHOD_DYNAMIC_PIN})
622
623 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
624 session.handle_submit({CONF_PAIRING_PIN: "123456"})
625
626 await _wait_for(lambda: session.finished)
627 await task
628 assert provider.submitted_pins == ["123456"]
629
630
631async def test_verify_presence_on_paired_device() -> None:
632 """Re-running the flow on a paired device runs the dynamic-PIN presence verification."""
633 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)], psk_category=PskCategory.LONG_TERM)
634 record = SimpleNamespace(pair_methods=[PairMethod.STATIC_PIN])
635 provider = _FakeProvider(api, record=record)
636 session, _mass = _make_session(_ok_finish)
637 player = _make_player(api, provider)
638
639 task = asyncio.create_task(player.run_setup_flow(session))
640 await _wait_step(session, step_type=FlowStepType.FORM, step_id="verify_pin")
641 assert provider.verify is True
642 assert provider.static is False
643 session.handle_submit({CONF_PAIRING_PIN: "123456"})
644
645 await _wait_for(lambda: session.finished)
646 await task
647 assert provider.submitted_pins == ["123456"]
648
649
650async def test_paired_device_without_verification_aborts() -> None:
651 """A paired device whose presence verification would add nothing aborts as already paired."""
652 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)], psk_category=PskCategory.LONG_TERM)
653 record = SimpleNamespace(pair_methods=[PairMethod.DYNAMIC_PIN])
654 provider = _FakeProvider(api, record=record)
655 session, _mass = _make_session(_ok_finish)
656 player = _make_player(api, provider)
657
658 with pytest.raises(AbortFlow) as excinfo:
659 await player.run_setup_flow(session)
660 assert excinfo.value.reason == "already_paired"
661 assert provider.start_calls == 0
662
663
664async def test_submit_pin_session_lost_rerenders() -> None:
665 """A session that ends underneath the submit re-renders the PIN form and starts afresh."""
666 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
667 provider = _FakeProvider(api, submit_outcomes=["session_lost", "success"])
668 session, _mass = _make_session(_ok_finish)
669 player = _make_player(api, provider)
670
671 task = asyncio.create_task(player.run_setup_flow(session))
672 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
673 session.handle_submit({CONF_PAIRING_PIN: "000000"})
674
675 error_step = await _wait_step(
676 session, step_type=FlowStepType.FORM, step_id="enter_pin", with_errors=True
677 )
678 assert error_step.errors == {"base": "pairing_error_no_pin_session"}
679 session.handle_submit({CONF_PAIRING_PIN: "123456"})
680
681 await _wait_for(lambda: session.finished)
682 await task
683 assert provider.submitted_pins == ["000000", "123456"]
684 assert provider.start_calls == 2
685
686
687async def test_gesture_timeout_propagates(monkeypatch: pytest.MonkeyPatch) -> None:
688 """An expired gesture wait propagates (timed_out abort) and tears the session down."""
689 monkeypatch.setattr("music_assistant.providers.sendspin.player.SERVER_GESTURE_TIMEOUT_S", 0.05)
690 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
691 provider = _FakeProvider(api, gesture=True)
692 session, _mass = _make_session(_ok_finish)
693 player = _make_player(api, provider)
694
695 with pytest.raises(StepExpiredError):
696 await player.run_setup_flow(session)
697 assert provider.cancel_calls == 1
698
699
700async def test_pin_form_expiry_retries_in_place(monkeypatch: pytest.MonkeyPatch) -> None:
701 """An unanswered PIN form re-renders with a timeout error rather than dropping the flow."""
702 monkeypatch.setattr("music_assistant.providers.sendspin.player.PAIR_PIN_ENTRY_TIMEOUT", 0.05)
703 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
704 provider = _FakeProvider(api)
705 session, _mass = _make_session(_ok_finish)
706 player = _make_player(api, provider)
707
708 task = asyncio.create_task(player.run_setup_flow(session))
709 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
710 retry = await _wait_step(
711 session, step_type=FlowStepType.FORM, step_id="enter_pin", with_errors=True
712 )
713 assert retry.errors == {"base": "pairing_error_timeout"}
714 session.handle_submit({CONF_PAIRING_PIN: "123456"})
715 await _wait_for(lambda: session.finished)
716 await task
717
718
719async def test_pin_form_encodes_the_negotiated_length() -> None:
720 """The PIN field renders as a pairing-code box matching the negotiated digit count."""
721 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
722 provider = _FakeProvider(api)
723 session, _mass = _make_session(_ok_finish)
724 player = _make_player(api, provider)
725
726 task = asyncio.create_task(player.run_setup_flow(session))
727 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
728 pin_entry = next(entry for entry in step.entries if entry.key == CONF_PAIRING_PIN)
729 assert pin_entry.type is ConfigEntryType.PAIRING_CODE
730 assert pin_entry.format == "###-###"
731 session.handle_submit({CONF_PAIRING_PIN: "123456"})
732 await _wait_for(lambda: session.finished)
733 await task
734
735
736async def test_pin_form_accepts_a_separator_in_the_submitted_pin() -> None:
737 """A PIN submitted with the format's separator still pairs (parse_value strips it)."""
738 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
739 provider = _FakeProvider(api)
740 session, _mass = _make_session(_ok_finish)
741 player = _make_player(api, provider)
742
743 task = asyncio.create_task(player.run_setup_flow(session))
744 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
745 session.handle_submit({CONF_PAIRING_PIN: "123-456"})
746
747 await _wait_for(lambda: session.finished)
748 await task
749 assert provider.submitted_pins == ["123456"]
750
751
752async def test_pin_form_rejects_a_short_pin() -> None:
753 """A PIN shorter than the negotiated length re-serves the form with a field error."""
754 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
755 provider = _FakeProvider(api)
756 session, _mass = _make_session(_ok_finish)
757 player = _make_player(api, provider)
758
759 task = asyncio.create_task(player.run_setup_flow(session))
760 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
761 step = session.handle_submit({CONF_PAIRING_PIN: "123"})
762 assert step is not None
763 assert step.errors == {CONF_PAIRING_PIN: "invalid_value"}
764 assert provider.submitted_pins == []
765
766 session.handle_submit({CONF_PAIRING_PIN: "123456"})
767 await _wait_for(lambda: session.finished)
768 await task
769
770
771async def test_static_pin_form_hints_where_the_pin_lives() -> None:
772 """A static-PIN form surfaces the device's own hint about where its PIN is printed."""
773 api = _FakeApi([_desc(PairMethod.STATIC_PIN, locations=["device", "bogus"])])
774 provider = _FakeProvider(api)
775 session, _mass = _make_session(_ok_finish)
776 player = _make_player(api, provider)
777
778 task = asyncio.create_task(player.run_setup_flow(session))
779 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
780 # The unknown location is ignored rather than rendered as a missing translation.
781 assert [entry.key for entry in step.entries] == [CONF_PAIRING_PIN]
782 assert step.entries[0].translation_key == "static_pin_location_device"
783 # A static PIN is always exactly 8 digits (enforced by aiosendspin).
784 pin_entry = next(entry for entry in step.entries if entry.key == CONF_PAIRING_PIN)
785 assert pin_entry.type is ConfigEntryType.PAIRING_CODE
786 assert pin_entry.format == "####-####"
787 session.handle_submit({CONF_PAIRING_PIN: "12345678"})
788 await _wait_for(lambda: session.finished)
789 await task
790
791
792async def test_dynamic_pin_form_hints_how_the_pin_arrives() -> None:
793 """A dynamic-PIN form surfaces the device's own hint about the channel carrying the PIN."""
794 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN, out_channels=["speaker", "other"])])
795 provider = _FakeProvider(api)
796 session, _mass = _make_session(_ok_finish)
797 player = _make_player(api, provider)
798
799 task = asyncio.create_task(player.run_setup_flow(session))
800 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
801 # "other" says nothing an operator can act on, so it renders no hint.
802 assert [entry.key for entry in step.entries] == [CONF_PAIRING_PIN]
803 assert step.entries[0].translation_key == "dynamic_pin_channel_speaker"
804 session.handle_submit({CONF_PAIRING_PIN: "123456"})
805 await _wait_for(lambda: session.finished)
806 await task
807
808
809async def test_dynamic_pin_form_names_both_channels_when_the_device_offers_both() -> None:
810 """A PIN carried on screen and aloud is labelled with both, since either one works."""
811 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN, out_channels=["display", "speaker"])])
812 provider = _FakeProvider(api)
813 session, _mass = _make_session(_ok_finish)
814 player = _make_player(api, provider)
815
816 task = asyncio.create_task(player.run_setup_flow(session))
817 step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
818 assert step.entries[0].translation_key == "dynamic_pin_channel_display_speaker"
819 session.handle_submit({CONF_PAIRING_PIN: "123456"})
820 await _wait_for(lambda: session.finished)
821 await task
822
823
824async def test_abort_mid_pairing_runs_cleanup() -> None:
825 """Cancelling the flow while a PIN session is in flight tears it down in the finally."""
826 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
827 provider = _FakeProvider(api)
828 session, _mass = _make_session(_ok_finish)
829 player = _make_player(api, provider)
830
831 task = asyncio.create_task(player.run_setup_flow(session))
832 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
833 assert provider.session is not None
834
835 task.cancel()
836 with pytest.raises(asyncio.CancelledError):
837 await task
838
839 assert provider.cancel_calls == 1
840 assert not session.finished
841
842
843async def test_token_only_device_aborts_as_unpairable() -> None:
844 """Token pairing is how a server enrols itself, so a token-only device cannot be paired."""
845 api = _FakeApi([_desc(PairMethod.PAIRING_PSK)])
846 provider = _FakeProvider(api)
847 session, _mass = _make_session(_ok_finish)
848 player = _make_player(api, provider)
849
850 with pytest.raises(AbortFlow) as excinfo:
851 await player.run_setup_flow(session)
852 assert excinfo.value.reason == "token_pairing_only"
853 assert provider.tokens == []
854
855
856async def test_token_hidden_when_the_device_can_pair_by_pin() -> None:
857 """A device offering both goes straight to its PIN, never showing the token as a choice."""
858 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.PAIRING_PSK)])
859 provider = _FakeProvider(api)
860 session, mass = _make_session(_ok_finish)
861 player = _make_player(api, provider)
862
863 task = asyncio.create_task(player.run_setup_flow(session))
864 await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
865 assert not any(s.step_id == "select_method" for s in _published_steps(mass))
866 session.handle_submit({CONF_PAIRING_PIN: "123456"})
867
868 await _wait_for(lambda: session.finished)
869 await task
870 assert provider.tokens == []
871
872
873async def test_no_pair_methods_aborts() -> None:
874 """A device offering nothing usable aborts with the no_pair_methods reason."""
875 api = _FakeApi([])
876 provider = _FakeProvider(api)
877 session, _mass = _make_session(_ok_finish)
878 player = _make_player(api, provider)
879
880 with pytest.raises(AbortFlow) as excinfo:
881 await player.run_setup_flow(session)
882 assert excinfo.value.reason == "no_pair_methods"
883 assert provider.start_calls == 0
884
885
886async def test_unencrypted_connection_aborts() -> None:
887 """An unencrypted (legacy) connection has nothing to pair and aborts."""
888 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
889 api.connection_security = None
890 provider = _FakeProvider(api)
891 session, _mass = _make_session(_ok_finish)
892 player = _make_player(api, provider)
893
894 with pytest.raises(AbortFlow) as excinfo:
895 await player.run_setup_flow(session)
896 assert excinfo.value.reason == "nothing_to_configure"
897
898
899def test_pairing_method_options_derivation() -> None:
900 """Static PIN needs both PIN methods usable; the token is never offered at all."""
901 api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.STATIC_PIN)])
902 provider = _FakeProvider(api)
903 player = _make_player(api, provider)
904 # Opposite the static option the generic "pin" gives way to the dynamic-specific value,
905 # so each option can describe itself.
906 assert player._pairing_method_options(cast("SendspinProvider", provider)) == [
907 PAIR_METHOD_DYNAMIC_PIN,
908 PAIR_METHOD_STATIC_PIN,
909 ]
910
911 api_single = _FakeApi(
912 [_desc(PairMethod.STATIC_PIN), _desc(PairMethod.PAIRING_PSK)], unpaired_access=True
913 )
914 provider_single = _FakeProvider(api_single)
915 player_single = _make_player(api_single, provider_single)
916 assert player_single._pairing_method_options(cast("SendspinProvider", provider_single)) == [
917 PAIR_METHOD_PIN
918 ]
919
920 # The token is never an operator-facing option, so a token-only device offers nothing.
921 api_token = _FakeApi([_desc(PairMethod.PAIRING_PSK)], unpaired_access=True)
922 provider_token = _FakeProvider(api_token)
923 player_token = _make_player(api_token, provider_token)
924 assert player_token._pairing_method_options(cast("SendspinProvider", provider_token)) == []
925