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