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