music-assistant-server

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 FlowStepType
18
19from music_assistant.models.setup_flow import (
20    AbortFlow,
21    SetupFlowContext,
22    SetupSession,
23    StepExpiredError,
24)
25from music_assistant.providers.sendspin import player as player_module
26from music_assistant.providers.sendspin.constants import (
27    CONF_PAIRING_METHOD,
28    CONF_PAIRING_PIN,
29    CONF_PAIRING_TOKEN,
30    PAIR_METHOD_DYNAMIC_PIN,
31    PAIR_METHOD_PIN,
32    PAIR_METHOD_STATIC_PIN,
33    PAIR_METHOD_TOKEN,
34    PAIR_METHOD_UNPAIRED,
35)
36from music_assistant.providers.sendspin.helpers import SecurityActionError
37from music_assistant.providers.sendspin.player import SendspinBasePlayer
38from tests.common import collect_loop_errors
39
40if TYPE_CHECKING:
41    from aiosendspin.server.client import SendspinClient
42    from music_assistant_models.setup_flow import SetupFlowStep
43
44    from music_assistant.providers.sendspin.provider import SendspinProvider
45
46
47def _desc(
48    method: PairMethod,
49    *,
50    locations: list[str] | None = None,
51    out_channels: list[str] | None = None,
52) -> PairMethodDescriptor:
53    return PairMethodDescriptor(method=method, locations=locations, out_channels=out_channels)
54
55
56class _FakePinSession:
57    """Minimal PinPairingSession stand-in the fake provider hands back to the flow."""
58
59    def __init__(
60        self,
61        *,
62        awaiting_gesture: bool = False,
63        verify: bool = False,
64        method: PairMethod = PairMethod.DYNAMIC_PIN,
65    ) -> None:
66        self.pin_request_event = asyncio.Event()
67        self.gesture_event = asyncio.Event()
68        if awaiting_gesture:
69            # A gesture-gated device reports pair-pending before asking for the PIN.
70            self.gesture_event.set()
71        else:
72            self.pin_request_event.set()
73        self.awaiting_pin = True
74        self.finished = False
75        self.error: Exception | None = None
76        self.can_retry = False
77        self.verify = verify
78        self.method = method
79        self.pin_length: int | None = 6 if method is PairMethod.DYNAMIC_PIN else None
80        # None so the flow's post-submit "confirming" wait is skipped in tests.
81        self.task: asyncio.Task[None] | None = None
82
83    @property
84    def awaiting_first_message(self) -> bool:
85        return not self.gesture_event.is_set() and not self.pin_request_event.is_set()
86
87    @property
88    def awaiting_gesture(self) -> bool:
89        return self.gesture_event.is_set() and not self.pin_request_event.is_set()
90
91    async def wait_first_message(self) -> None:
92        await self.pin_request_event.wait()
93
94    async def wait_pin_request(self) -> None:
95        await self.pin_request_event.wait()
96
97
98class _FakeApi:
99    """Fake SendspinClient exposing just what the flow reads (hello + security + roles)."""
100
101    def __init__(
102        self,
103        methods: list[PairMethodDescriptor],
104        *,
105        active_roles: tuple[str, ...] = (),
106        psk_category: PskCategory = PskCategory.SENTINEL,
107        unpaired_access: bool = False,
108    ):
109        self.info_or_none = SimpleNamespace(
110            supported_pair_methods=list(methods),
111            unpaired_access=SimpleNamespace(enabled=unpaired_access),
112        )
113        self.connection_security: Any = SimpleNamespace(psk_category=psk_category)
114        self.active_roles = active_roles
115
116
117class _FakePairingStore:
118    """Pairing-store stand-in serving a scripted record and unpaired-trust state."""
119
120    def __init__(self, record: Any = None, trusted: Any = None) -> None:
121        self.record = record
122        self.trusted = trusted
123
124    async def record_by_client_id(self, client_id: str) -> Any:
125        return self.record
126
127    async def trusted_unpaired(self, client_id: str) -> Any:
128        return self.trusted
129
130
131class _FakeProvider:
132    """Scripts the pairing primitives run_setup_flow drives, recording every call."""
133
134    def __init__(
135        self,
136        api: _FakeApi,
137        *,
138        gesture: bool = False,
139        submit_outcomes: list[str] | None = None,
140        token_errors: list[Exception] | None = None,
141        record: Any = None,
142        trusted: Any = None,
143    ) -> None:
144        self.api = api
145        self.server_api = SimpleNamespace(pairing_store=_FakePairingStore(record, trusted))
146        self.session: _FakePinSession | None = None
147        self.start_calls = 0
148        self.static: bool | None = None
149        self.verify: bool | None = None
150        self.submitted_pins: list[str] = []
151        self.tokens: list[str] = []
152        self.cancel_calls = 0
153        self.clear_calls = 0
154        self.trust_calls: list[bool] = []
155        self._gesture = gesture
156        self._submit_outcomes = deque(submit_outcomes or [])
157        self._token_errors = deque(token_errors or [])
158
159    def pairing_config_snapshot(self, client_id: str) -> None:
160        return None
161
162    def get_pin_session(self, client_id: str) -> _FakePinSession | None:
163        return self.session
164
165    def clear_pin_session(self, client_id: str) -> None:
166        self.clear_calls += 1
167        if self.session is not None and self.session.finished:
168            self.session = None
169
170    async def start_pin_pairing(
171        self, client_id: str, *, verify: bool = False, static: bool = False
172    ) -> _FakePinSession:
173        self.start_calls += 1
174        self.static = static
175        self.verify = verify
176        if self.session is not None and self.session.can_retry:
177            # A retryable session resumes in place, past the gesture, awaiting a PIN again.
178            self.session.can_retry = False
179            self.session.error = None
180            self.session.awaiting_pin = True
181            self.session.pin_request_event.set()
182            return self.session
183        offered = {d.method for d in self.api.info_or_none.supported_pair_methods}
184        dynamic_offered = PairMethod.DYNAMIC_PIN in offered
185        self.session = _FakePinSession(
186            awaiting_gesture=self._gesture,
187            verify=verify,
188            method=(
189                PairMethod.DYNAMIC_PIN if dynamic_offered and not static else PairMethod.STATIC_PIN
190            ),
191        )
192        return self.session
193
194    def submit_pin(self, client_id: str, pin: str) -> None:
195        assert self.session is not None
196        self.submitted_pins.append(pin)
197        outcome = self._submit_outcomes.popleft() if self._submit_outcomes else "success"
198        if outcome == "success":
199            self.session.finished = True
200            self.session.error = None
201            self.session.awaiting_pin = False
202            self.api.active_roles = ("player",)
203        elif outcome == "retry":
204            self.session.can_retry = True
205            self.session.error = RemotePairingAbortError(PairAbortReason.PIN_MISMATCH)
206        elif outcome == "session_lost":
207            self.session = None
208            raise SecurityActionError("pairing_error_no_pin_session")
209
210    async def cancel_pin_pairing(self, client_id: str) -> None:
211        self.cancel_calls += 1
212        self.session = None
213
214    async def set_trusted_unpaired(self, client_id: str, enabled: bool) -> None:
215        self.trust_calls.append(enabled)
216        if enabled:
217            self.server_api.pairing_store.trusted = object()
218
219    async def pair_with_token(self, client_id: str, token: str) -> None:
220        self.tokens.append(token)
221        if self._token_errors:
222            raise self._token_errors.popleft()
223        self.api.active_roles = ("player",)
224
225
226def _make_player(api: _FakeApi, provider: _FakeProvider) -> SendspinBasePlayer:
227    player = SendspinBasePlayer.__new__(SendspinBasePlayer)
228    player._player_id = "client-1"
229    player._provider = cast("SendspinProvider", provider)
230    player.api = cast("SendspinClient", api)
231    return player
232
233
234def _make_session(finish_handler: Any) -> tuple[SetupSession, mock.Mock]:
235    mass = mock.Mock()
236    context = SetupFlowContext(kind="setup", reason="user", domain="sendspin", player_id="client-1")
237    return SetupSession(mass, "flow-test", context, finish_handler), mass
238
239
240async def _ok_finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
241    """Finish handler stand-in that accepts any values and reports the player id."""
242    return {"player_id": "client-1"}
243
244
245def _published_steps(mass: mock.Mock) -> list[Any]:
246    return [call.kwargs["data"] for call in mass.signal_event.call_args_list]
247
248
249async def _wait_for(predicate: Any, timeout: float = 5.0) -> Any:
250    deadline = time.monotonic() + timeout
251    while time.monotonic() < deadline:
252        if result := predicate():
253            return result
254        await asyncio.sleep(0.01)
255    raise AssertionError("condition not met within timeout")
256
257
258async def _wait_step(
259    session: SetupSession,
260    *,
261    step_type: FlowStepType | None = None,
262    step_id: str | None = None,
263    with_errors: bool = False,
264) -> SetupFlowStep:
265    def _match() -> SetupFlowStep | None:
266        step = session.current_step
267        if step is None:
268            return None
269        if step_type is not None and step.type != step_type:
270            return None
271        if step_id is not None and step.step_id != step_id:
272            return None
273        if with_errors and not step.errors:
274            return None
275        return step
276
277    return cast("SetupFlowStep", await _wait_for(_match))
278
279
280async def test_select_method_pin_gesture_submit_success() -> None:
281    """Select PIN, wait through the gesture, submit the PIN, succeed, and finish with {}."""
282    collected: dict[str, Any] = {}
283
284    async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
285        collected["values"] = values
286        return {"player_id": "client-1"}
287
288    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.STATIC_PIN)])
289    provider = _FakeProvider(api, gesture=True)
290    session, mass = _make_session(finish)
291    player = _make_player(api, provider)
292
293    task = asyncio.create_task(player.run_setup_flow(session))
294    step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="select_method")
295    assert {option.value for option in step.entries[0].options} == {
296        PAIR_METHOD_DYNAMIC_PIN,
297        PAIR_METHOD_STATIC_PIN,
298    }
299    # the method is rendered as an expanded (radio) list with nothing preselected
300    assert step.entries[0].expanded_options is True
301    assert step.entries[0].default_value is None
302    session.handle_submit({CONF_PAIRING_METHOD: PAIR_METHOD_DYNAMIC_PIN})
303
304    await _wait_step(session, step_type=FlowStepType.PROGRESS, step_id="awaiting_gesture")
305    assert provider.session is not None
306    provider.session.pin_request_event.set()
307
308    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
309    session.handle_submit({CONF_PAIRING_PIN: "123456"})
310
311    await _wait_for(lambda: session.finished)
312    await task
313
314    assert collected["values"] == {}
315    assert provider.submitted_pins == ["123456"]
316    assert provider.static is False
317    assert provider.verify is False
318    assert provider.cancel_calls == 0
319    assert provider.clear_calls == 1
320    steps = _published_steps(mass)
321    assert [s.step_id for s in steps if s.type == FlowStepType.PROGRESS] == ["awaiting_gesture"]
322    assert steps[-1].type == FlowStepType.FINISH
323
324
325async def test_confirming_wait_failure_after_deadline_logs_no_loop_error(
326    monkeypatch: pytest.MonkeyPatch,
327) -> None:
328    """A pairing attempt failing after the confirming step expired is not reported to the loop."""
329    release = asyncio.Event()
330
331    async def _failing_attempt() -> None:
332        await release.wait()
333        raise RuntimeError("refreshing the player failed")
334
335    monkeypatch.setattr(player_module, "PAIR_CONFIRM_TIMEOUT", 0.01)
336    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
337    provider = _FakeProvider(api)
338    session, _mass = _make_session(_ok_finish)
339    player = _make_player(api, provider)
340
341    with collect_loop_errors() as reported:
342        flow = asyncio.create_task(player.run_setup_flow(session))
343        await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
344        assert provider.session is not None
345        attempt = asyncio.create_task(_failing_attempt())
346        provider.session.task = attempt
347        session.handle_submit({CONF_PAIRING_PIN: "123456"})
348
349        # let the attempt fail only once the confirming step has expired and the flow has
350        # moved on, so the failure reliably lands after the flow stopped waiting for it
351        await _wait_for(lambda: session.finished)
352        await flow
353        release.set()
354        with pytest.raises(RuntimeError, match="refreshing the player failed"):
355            await attempt
356
357    assert reported == []
358
359
360async def test_single_pin_method_skips_select() -> None:
361    """A device offering one usable PIN method goes straight to the PIN form, no method select."""
362    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
363    provider = _FakeProvider(api)
364    session, mass = _make_session(_ok_finish)
365    player = _make_player(api, provider)
366
367    task = asyncio.create_task(player.run_setup_flow(session))
368    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
369    assert not any(s.step_id == "select_method" for s in _published_steps(mass))
370    session.handle_submit({CONF_PAIRING_PIN: "123456"})
371    await _wait_for(lambda: session.finished)
372    await task
373    assert provider.submitted_pins == ["123456"]
374
375
376async def test_pin_mismatch_retries_in_place_then_succeeds() -> None:
377    """A mismatch re-renders the PIN form with a base error; the retry resumes and succeeds."""
378    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
379    provider = _FakeProvider(api, submit_outcomes=["retry", "success"])
380    session, _mass = _make_session(_ok_finish)
381    player = _make_player(api, provider)
382
383    task = asyncio.create_task(player.run_setup_flow(session))
384    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
385    session.handle_submit({CONF_PAIRING_PIN: "000000"})
386
387    error_step = await _wait_step(
388        session, step_type=FlowStepType.FORM, step_id="enter_pin", with_errors=True
389    )
390    assert error_step.errors == {"base": "pairing_error_pin_mismatch"}
391    session.handle_submit({CONF_PAIRING_PIN: "123456"})
392
393    await _wait_for(lambda: session.finished)
394    await task
395    assert provider.submitted_pins == ["000000", "123456"]
396    assert provider.start_calls == 2
397    assert provider.cancel_calls == 0
398
399
400async def test_trusted_unpaired_pin_mismatch_still_retries() -> None:
401    """With unpaired access already trusted, a mismatch must not be misreported as success."""
402    api = _FakeApi(
403        [_desc(PairMethod.DYNAMIC_PIN)],
404        active_roles=("player",),
405        unpaired_access=True,
406    )
407    provider = _FakeProvider(api, submit_outcomes=["retry", "success"], trusted=object())
408    session, mass = _make_session(_ok_finish)
409    player = _make_player(api, provider)
410
411    task = asyncio.create_task(player.run_setup_flow(session))
412    # trusted-unpaired devices are not offered the unpaired option again
413    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
414    assert not any(s.step_id == "select_method" for s in _published_steps(mass))
415    session.handle_submit({CONF_PAIRING_PIN: "000000"})
416
417    error_step = await _wait_step(
418        session, step_type=FlowStepType.FORM, step_id="enter_pin", with_errors=True
419    )
420    assert error_step.errors == {"base": "pairing_error_pin_mismatch"}
421    session.handle_submit({CONF_PAIRING_PIN: "123456"})
422
423    await _wait_for(lambda: session.finished)
424    await task
425    assert provider.submitted_pins == ["000000", "123456"]
426    assert provider.trust_calls == []
427
428
429async def test_unpaired_option_grants_trust() -> None:
430    """Picking the unpaired option grants unpaired playback and finishes without pairing."""
431    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)], unpaired_access=True)
432    provider = _FakeProvider(api)
433    session, _mass = _make_session(_ok_finish)
434    player = _make_player(api, provider)
435
436    task = asyncio.create_task(player.run_setup_flow(session))
437    step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="select_method")
438    assert {option.value for option in step.entries[0].options} == {
439        PAIR_METHOD_PIN,
440        PAIR_METHOD_UNPAIRED,
441    }
442    session.handle_submit({CONF_PAIRING_METHOD: PAIR_METHOD_UNPAIRED})
443
444    await _wait_for(lambda: session.finished)
445    await task
446    assert provider.trust_calls == [True]
447    assert provider.start_calls == 0
448    assert provider.tokens == []
449
450
451async def test_lone_unpaired_option_still_asks() -> None:
452    """A lone unpaired-access option is never auto-picked: the grant needs an explicit choice."""
453    api = _FakeApi([], unpaired_access=True)
454    provider = _FakeProvider(api)
455    session, _mass = _make_session(_ok_finish)
456    player = _make_player(api, provider)
457
458    task = asyncio.create_task(player.run_setup_flow(session))
459    step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="select_method")
460    assert [option.value for option in step.entries[0].options] == [PAIR_METHOD_UNPAIRED]
461    session.handle_submit({CONF_PAIRING_METHOD: PAIR_METHOD_UNPAIRED})
462
463    await _wait_for(lambda: session.finished)
464    await task
465    assert provider.trust_calls == [True]
466
467
468async def test_verify_presence_on_paired_device() -> None:
469    """Re-running the flow on a paired device runs the dynamic-PIN presence verification."""
470    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)], psk_category=PskCategory.LONG_TERM)
471    record = SimpleNamespace(pair_methods=[PairMethod.STATIC_PIN])
472    provider = _FakeProvider(api, record=record)
473    session, _mass = _make_session(_ok_finish)
474    player = _make_player(api, provider)
475
476    task = asyncio.create_task(player.run_setup_flow(session))
477    await _wait_step(session, step_type=FlowStepType.FORM, step_id="verify_pin")
478    assert provider.verify is True
479    assert provider.static is False
480    session.handle_submit({CONF_PAIRING_PIN: "123456"})
481
482    await _wait_for(lambda: session.finished)
483    await task
484    assert provider.submitted_pins == ["123456"]
485
486
487async def test_paired_device_without_verification_aborts() -> None:
488    """A paired device whose presence verification would add nothing aborts as already paired."""
489    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)], psk_category=PskCategory.LONG_TERM)
490    record = SimpleNamespace(pair_methods=[PairMethod.DYNAMIC_PIN])
491    provider = _FakeProvider(api, record=record)
492    session, _mass = _make_session(_ok_finish)
493    player = _make_player(api, provider)
494
495    with pytest.raises(AbortFlow) as excinfo:
496        await player.run_setup_flow(session)
497    assert excinfo.value.reason == "already_paired"
498    assert provider.start_calls == 0
499
500
501async def test_submit_pin_session_lost_rerenders() -> None:
502    """A session that ends underneath the submit re-renders the PIN form and starts afresh."""
503    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
504    provider = _FakeProvider(api, submit_outcomes=["session_lost", "success"])
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="enter_pin")
510    session.handle_submit({CONF_PAIRING_PIN: "000000"})
511
512    error_step = await _wait_step(
513        session, step_type=FlowStepType.FORM, step_id="enter_pin", with_errors=True
514    )
515    assert error_step.errors == {"base": "pairing_error_no_pin_session"}
516    session.handle_submit({CONF_PAIRING_PIN: "123456"})
517
518    await _wait_for(lambda: session.finished)
519    await task
520    assert provider.submitted_pins == ["000000", "123456"]
521    assert provider.start_calls == 2
522
523
524async def test_gesture_timeout_propagates(monkeypatch: pytest.MonkeyPatch) -> None:
525    """An expired gesture wait propagates (timed_out abort) and tears the session down."""
526    monkeypatch.setattr("music_assistant.providers.sendspin.player.SERVER_GESTURE_TIMEOUT_S", 0.05)
527    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
528    provider = _FakeProvider(api, gesture=True)
529    session, _mass = _make_session(_ok_finish)
530    player = _make_player(api, provider)
531
532    with pytest.raises(StepExpiredError):
533        await player.run_setup_flow(session)
534    assert provider.cancel_calls == 1
535
536
537async def test_pin_form_expiry_retries_in_place(monkeypatch: pytest.MonkeyPatch) -> None:
538    """An unanswered PIN form re-renders with a timeout error rather than dropping the flow."""
539    monkeypatch.setattr("music_assistant.providers.sendspin.player.PAIR_PIN_ENTRY_TIMEOUT", 0.05)
540    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
541    provider = _FakeProvider(api)
542    session, _mass = _make_session(_ok_finish)
543    player = _make_player(api, provider)
544
545    task = asyncio.create_task(player.run_setup_flow(session))
546    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
547    retry = await _wait_step(
548        session, step_type=FlowStepType.FORM, step_id="enter_pin", with_errors=True
549    )
550    assert retry.errors == {"base": "pairing_error_timeout"}
551    session.handle_submit({CONF_PAIRING_PIN: "123456"})
552    await _wait_for(lambda: session.finished)
553    await task
554
555
556async def test_pin_form_names_the_pin_length() -> None:
557    """The PIN form states the negotiated digit count."""
558    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
559    provider = _FakeProvider(api)
560    session, _mass = _make_session(_ok_finish)
561    player = _make_player(api, provider)
562
563    task = asyncio.create_task(player.run_setup_flow(session))
564    step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
565    digits = next(entry for entry in step.entries if entry.key == "dynamic_pin_digits")
566    assert digits.translation_params == ["6"]
567    session.handle_submit({CONF_PAIRING_PIN: "123456"})
568    await _wait_for(lambda: session.finished)
569    await task
570
571
572async def test_static_pin_form_hints_where_the_pin_lives() -> None:
573    """A static-PIN form surfaces the device's own hint about where its PIN is printed."""
574    api = _FakeApi([_desc(PairMethod.STATIC_PIN, locations=["device", "bogus"])])
575    provider = _FakeProvider(api)
576    session, _mass = _make_session(_ok_finish)
577    player = _make_player(api, provider)
578
579    task = asyncio.create_task(player.run_setup_flow(session))
580    step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
581    # The unknown location is ignored rather than rendered as a missing translation.
582    # A static PIN has no negotiated length, so nothing names a digit count.
583    assert [entry.key for entry in step.entries] == [
584        "static_pin_location_device",
585        CONF_PAIRING_PIN,
586    ]
587    session.handle_submit({CONF_PAIRING_PIN: "12345678"})
588    await _wait_for(lambda: session.finished)
589    await task
590
591
592async def test_dynamic_pin_form_hints_how_the_pin_arrives() -> None:
593    """A dynamic-PIN form surfaces the device's own hint about the channel carrying the PIN."""
594    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN, out_channels=["speaker", "other"])])
595    provider = _FakeProvider(api)
596    session, _mass = _make_session(_ok_finish)
597    player = _make_player(api, provider)
598
599    task = asyncio.create_task(player.run_setup_flow(session))
600    step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
601    # "other" says nothing an operator can act on, so it renders no hint.
602    assert [entry.key for entry in step.entries] == [
603        "dynamic_pin_channel_speaker",
604        "dynamic_pin_digits",
605        CONF_PAIRING_PIN,
606    ]
607    session.handle_submit({CONF_PAIRING_PIN: "123456"})
608    await _wait_for(lambda: session.finished)
609    await task
610
611
612async def test_token_form_hints_where_the_token_lives() -> None:
613    """A token form surfaces the device's own hint about where its pairing secret is printed."""
614    api = _FakeApi([_desc(PairMethod.PAIRING_PSK, locations=["leaflet"])])
615    provider = _FakeProvider(api)
616    session, _mass = _make_session(_ok_finish)
617    player = _make_player(api, provider)
618
619    task = asyncio.create_task(player.run_setup_flow(session))
620    step = await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_token")
621    assert [entry.key for entry in step.entries] == [
622        "pairing_psk_location_leaflet",
623        CONF_PAIRING_TOKEN,
624    ]
625    session.handle_submit({CONF_PAIRING_TOKEN: "tok-1"})
626    await _wait_for(lambda: session.finished)
627    await task
628
629
630async def test_abort_mid_pairing_runs_cleanup() -> None:
631    """Cancelling the flow while a PIN session is in flight tears it down in the finally."""
632    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
633    provider = _FakeProvider(api)
634    session, _mass = _make_session(_ok_finish)
635    player = _make_player(api, provider)
636
637    task = asyncio.create_task(player.run_setup_flow(session))
638    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
639    assert provider.session is not None
640
641    task.cancel()
642    with pytest.raises(asyncio.CancelledError):
643        await task
644
645    assert provider.cancel_calls == 1
646    assert not session.finished
647
648
649async def test_token_hidden_when_the_device_can_pair_by_pin() -> None:
650    """A device offering both goes straight to its PIN, never showing the token as a choice."""
651    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.PAIRING_PSK)])
652    provider = _FakeProvider(api)
653    session, mass = _make_session(_ok_finish)
654    player = _make_player(api, provider)
655
656    task = asyncio.create_task(player.run_setup_flow(session))
657    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_pin")
658    assert not any(s.step_id == "select_method" for s in _published_steps(mass))
659    session.handle_submit({CONF_PAIRING_PIN: "123456"})
660
661    await _wait_for(lambda: session.finished)
662    await task
663    assert provider.tokens == []
664
665
666async def test_token_pairing_success() -> None:
667    """A token-only device drives the token form and pairs on submit."""
668    collected: dict[str, Any] = {}
669
670    async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
671        collected["values"] = values
672        return {"player_id": "client-1"}
673
674    api = _FakeApi([_desc(PairMethod.PAIRING_PSK)])
675    provider = _FakeProvider(api)
676    session, mass = _make_session(finish)
677    player = _make_player(api, provider)
678
679    task = asyncio.create_task(player.run_setup_flow(session))
680    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_token")
681    assert not any(s.step_id == "select_method" for s in _published_steps(mass))
682    session.handle_submit({CONF_PAIRING_TOKEN: "tok-123"})
683
684    await _wait_for(lambda: session.finished)
685    await task
686    assert provider.tokens == ["tok-123"]
687    assert collected["values"] == {}
688
689
690async def test_token_invalid_re_renders_then_succeeds() -> None:
691    """An invalid token re-renders the token form with a base error, then pairs on retry."""
692    api = _FakeApi([_desc(PairMethod.PAIRING_PSK)])
693    provider = _FakeProvider(api, token_errors=[SecurityActionError("pairing_error_token_invalid")])
694    session, _mass = _make_session(_ok_finish)
695    player = _make_player(api, provider)
696
697    task = asyncio.create_task(player.run_setup_flow(session))
698    await _wait_step(session, step_type=FlowStepType.FORM, step_id="enter_token")
699    session.handle_submit({CONF_PAIRING_TOKEN: "bad"})
700
701    error_step = await _wait_step(
702        session, step_type=FlowStepType.FORM, step_id="enter_token", with_errors=True
703    )
704    assert error_step.errors == {"base": "pairing_error_token_invalid"}
705    session.handle_submit({CONF_PAIRING_TOKEN: "good"})
706
707    await _wait_for(lambda: session.finished)
708    await task
709    assert provider.tokens == ["bad", "good"]
710
711
712async def test_no_pair_methods_aborts() -> None:
713    """A device offering nothing usable aborts with the no_pair_methods reason."""
714    api = _FakeApi([])
715    provider = _FakeProvider(api)
716    session, _mass = _make_session(_ok_finish)
717    player = _make_player(api, provider)
718
719    with pytest.raises(AbortFlow) as excinfo:
720        await player.run_setup_flow(session)
721    assert excinfo.value.reason == "no_pair_methods"
722    assert provider.start_calls == 0
723
724
725async def test_unencrypted_connection_aborts() -> None:
726    """An unencrypted (legacy) connection has nothing to pair and aborts."""
727    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN)])
728    api.connection_security = None
729    provider = _FakeProvider(api)
730    session, _mass = _make_session(_ok_finish)
731    player = _make_player(api, provider)
732
733    with pytest.raises(AbortFlow) as excinfo:
734        await player.run_setup_flow(session)
735    assert excinfo.value.reason == "nothing_to_configure"
736
737
738def test_pairing_method_options_derivation() -> None:
739    """Static PIN needs both PIN methods usable; the token yields to any usable PIN."""
740    api = _FakeApi([_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.STATIC_PIN)])
741    provider = _FakeProvider(api)
742    player = _make_player(api, provider)
743    # Opposite the static option the generic "pin" gives way to the dynamic-specific value,
744    # so each option can describe itself.
745    assert player._pairing_method_options(
746        cast("SendspinProvider", provider), offer_unpaired=True
747    ) == [PAIR_METHOD_DYNAMIC_PIN, PAIR_METHOD_STATIC_PIN]
748
749    api_single = _FakeApi(
750        [_desc(PairMethod.STATIC_PIN), _desc(PairMethod.PAIRING_PSK)], unpaired_access=True
751    )
752    provider_single = _FakeProvider(api_single)
753    player_single = _make_player(api_single, provider_single)
754    assert player_single._pairing_method_options(
755        cast("SendspinProvider", provider_single), offer_unpaired=True
756    ) == [PAIR_METHOD_PIN, PAIR_METHOD_UNPAIRED]
757    assert player_single._pairing_method_options(
758        cast("SendspinProvider", provider_single), offer_unpaired=False
759    ) == [PAIR_METHOD_PIN]
760
761    # Without a PIN to fall back on the token is the only way in, so it returns to the list.
762    api_token = _FakeApi([_desc(PairMethod.PAIRING_PSK)], unpaired_access=True)
763    provider_token = _FakeProvider(api_token)
764    player_token = _make_player(api_token, provider_token)
765    assert player_token._pairing_method_options(
766        cast("SendspinProvider", provider_token), offer_unpaired=True
767    ) == [PAIR_METHOD_TOKEN, PAIR_METHOD_UNPAIRED]
768