music-assistant-server

1"""Tests for the Sendspin operator PIN pairing session state machine and orchestration."""
2
3from __future__ import annotations
4
5import asyncio
6import inspect
7import logging
8from collections import deque
9from types import SimpleNamespace
10from typing import TYPE_CHECKING, Any, cast
11
12import pytest
13from aiosendspin.models.core import PairMethodDescriptor
14from aiosendspin.models.types import PairAbortReason, PairMethod
15from aiosendspin.noise.driver import HandshakeAbortedError
16from aiosendspin.noise.pairing import (
17    LocalPairingAbortError,
18    PairingError,
19    RemotePairingAbortError,
20)
21
22import music_assistant.providers.sendspin.provider as provider_module
23from music_assistant.providers.sendspin.helpers import SecurityActionError
24from music_assistant.providers.sendspin.provider import (
25    PinPairingSession,
26    SendspinProvider,
27    _pin_idle_task_id,
28)
29
30if TYPE_CHECKING:
31    from aiosendspin.noise.pairing import PairingAttempt
32    from aiosendspin.server import SendspinServer
33    from aiosendspin.server.client import SendspinClient
34
35    from music_assistant.mass import MusicAssistant
36
37
38def _desc(method: PairMethod, *, locked_out: bool = False) -> PairMethodDescriptor:
39    return PairMethodDescriptor(method=method, locked_out=locked_out)
40
41
42async def _blocked() -> None:
43    await asyncio.Event().wait()
44
45
46async def _submit_and_settle(provider: SendspinProvider, pin: str, client_id: str = "c") -> None:
47    """Submit the PIN and wait for the attempt task to reach its outcome."""
48    provider.submit_pin(client_id, pin)
49    session = provider.get_pin_session(client_id)
50    assert session is not None
51    assert session.task is not None
52    await session.task
53
54
55class _FakeMass:
56    """MusicAssistant stand-in mirroring create_task / call_later timer bookkeeping."""
57
58    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
59        self.loop = loop
60        self.timers: dict[str, asyncio.TimerHandle] = {}
61        self.tasks: dict[str, asyncio.Task[Any]] = {}
62
63    def create_task(
64        self, coro: Any, *, task_id: str | None = None, abort_existing: bool = False
65    ) -> asyncio.Task[Any]:
66        if task_id and (existing := self.tasks.get(task_id)) and not existing.done():
67            if abort_existing:
68                existing.cancel()
69            else:
70                if asyncio.iscoroutine(coro):
71                    coro.close()
72                return existing
73        task = self.loop.create_task(coro)
74        if task_id is not None:
75            key = task_id
76            self.tasks[key] = task
77
78            def _discard(_task: asyncio.Task[Any]) -> None:
79                self.tasks.pop(key, None)
80
81            task.add_done_callback(_discard)
82        return task
83
84    def call_later(
85        self, delay: float, target: Any, *args: Any, task_id: str, **kwargs: Any
86    ) -> asyncio.TimerHandle:
87        if existing := self.timers.get(task_id):
88            existing.cancel()
89
90        def _fire() -> None:
91            self.timers.pop(task_id, None)
92            if inspect.iscoroutinefunction(target) or inspect.iscoroutine(target):
93                self.create_task(target(*args, **kwargs), task_id=task_id, abort_existing=True)
94            else:
95                target(*args, **kwargs)
96
97        handle = self.loop.call_later(delay, _fire)
98        self.timers[task_id] = handle
99        return handle
100
101    def cancel_timer(self, task_id: str) -> None:
102        if handle := self.timers.pop(task_id, None):
103            handle.cancel()
104
105    def cancel_task(self, task_id: str) -> None:
106        if task := self.tasks.pop(task_id, None):
107            task.cancel()
108
109
110def _timers(provider: SendspinProvider) -> dict[str, asyncio.TimerHandle]:
111    """Pending call_later timers on the provider's fake mass."""
112    return cast("_FakeMass", provider.mass).timers
113
114
115class _FakeServerApi:
116    """Scripts pairing outcomes and records end_pairing calls, mirroring the real wiring."""
117
118    def __init__(
119        self,
120        methods: list[PairMethodDescriptor],
121        *,
122        await_pin: bool = True,
123        gesture: asyncio.Event | None = None,
124    ) -> None:
125        self._client = cast(
126            "SendspinClient",
127            SimpleNamespace(info_or_none=SimpleNamespace(supported_pair_methods=methods)),
128        )
129        self._await_pin = await_pin
130        self._gesture = gesture
131        self._active_cancel: asyncio.Event | None = None
132        self._cancel_requested = False
133        self.outcomes: deque[BaseException | None] = deque()
134        self.attempts: list[PairingAttempt] = []
135        self.initiate_calls = 0
136        self.end_pairing_calls = 0
137
138    def get_client(self, client_id: str) -> SendspinClient | None:
139        return self._client
140
141    async def initiate_pairing(self, client_id: str, attempt: PairingAttempt) -> None:
142        self.initiate_calls += 1
143        self.attempts.append(attempt)
144        if self._gesture is not None:
145            await self._gesture.wait()
146        if self._await_pin and attempt.pin_provider is not None:
147            if not self._cancel_requested:
148                cancel = asyncio.Event()
149                self._active_cancel = cancel
150                cancel_task = asyncio.ensure_future(cancel.wait())
151                pin_task = asyncio.ensure_future(attempt.pin_provider())
152                try:
153                    await asyncio.wait(
154                        {pin_task, cancel_task},
155                        return_when=asyncio.FIRST_COMPLETED,
156                    )
157                finally:
158                    cancel_task.cancel()
159                    self._active_cancel = None
160            if self._cancel_requested:
161                # end_pairing cancels the attempt regardless of ordering, as the real one does.
162                raise LocalPairingAbortError(PairAbortReason.USER_CANCELLED)
163        outcome = self.outcomes.popleft() if self.outcomes else None
164        if isinstance(outcome, BaseException):
165            raise outcome
166
167    async def end_pairing(self, client_id: str) -> None:
168        self.end_pairing_calls += 1
169        self._cancel_requested = True
170        if self._active_cancel is not None:
171            self._active_cancel.set()
172
173
174def _make_provider(
175    server_api: _FakeServerApi, monkeypatch: pytest.MonkeyPatch
176) -> tuple[SendspinProvider, list[str]]:
177    provider = SendspinProvider.__new__(SendspinProvider)
178    provider.mass = cast("MusicAssistant", _FakeMass(asyncio.get_running_loop()))
179    provider.server_api = cast("SendspinServer", server_api)
180    provider._pin_sessions = {}
181    provider._pairing_config_snapshots = {}
182    provider.logger = logging.getLogger("test.sendspin.pin")
183    refreshed: list[str] = []
184
185    async def _record_refresh(client_id: str) -> None:
186        refreshed.append(client_id)
187
188    monkeypatch.setattr(provider, "_refresh_player", _record_refresh)
189    return provider, refreshed
190
191
192async def test_session_running_states() -> None:
193    """A running attempt tracks the gesture and PIN waits independently."""
194    loop = asyncio.get_running_loop()
195    running: asyncio.Task[None] = loop.create_task(_blocked())
196
197    awaiting_future: asyncio.Future[str] = loop.create_future()
198    awaiting = PinPairingSession(
199        client_id="c", method=PairMethod.DYNAMIC_PIN, pin_future=awaiting_future, task=running
200    )
201    assert awaiting.attempt_running
202    assert awaiting.awaiting_pin
203    assert awaiting.awaiting_gesture
204    assert not awaiting.can_retry
205    assert not awaiting.finished
206
207    submitted_future: asyncio.Future[str] = loop.create_future()
208    submitted_future.set_result("123456")
209    submitted_early = PinPairingSession(
210        client_id="c", method=PairMethod.DYNAMIC_PIN, pin_future=submitted_future, task=running
211    )
212    assert submitted_early.attempt_running
213    assert not submitted_early.awaiting_pin
214    assert submitted_early.awaiting_gesture
215
216    in_progress = PinPairingSession(
217        client_id="c", method=PairMethod.DYNAMIC_PIN, pin_future=submitted_future, task=running
218    )
219    in_progress.pin_request_event.set()
220    assert in_progress.attempt_running
221    assert not in_progress.awaiting_pin
222    assert not in_progress.awaiting_gesture
223    running.cancel()
224
225
226async def test_session_terminal_states() -> None:
227    """A completed attempt is retryable while retryable is set, terminal once cleared."""
228    loop = asyncio.get_running_loop()
229    done: asyncio.Task[None] = loop.create_task(asyncio.sleep(0))
230    await done
231    pin_future: asyncio.Future[str] = loop.create_future()
232    pin_future.cancel()
233
234    retryable = PinPairingSession(
235        client_id="c",
236        method=PairMethod.DYNAMIC_PIN,
237        pin_future=pin_future,
238        task=done,
239        retryable=True,
240    )
241    assert retryable.can_retry
242    assert not retryable.finished
243    assert not retryable.awaiting_pin
244    assert not retryable.awaiting_gesture
245
246    terminal = PinPairingSession(
247        client_id="c", method=PairMethod.DYNAMIC_PIN, pin_future=pin_future, task=done
248    )
249    assert terminal.finished
250    assert not terminal.can_retry
251
252
253async def test_pin_pairing_success(monkeypatch: pytest.MonkeyPatch) -> None:
254    """A submitted PIN that succeeds finishes the session and refreshes the player."""
255    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)])
256    provider, refreshed = _make_provider(api, monkeypatch)
257    session = await provider.start_pin_pairing("c")
258    assert session.awaiting_pin
259    await _submit_and_settle(provider, "123456")
260    assert session.finished
261    assert session.error is None
262    assert refreshed == ["c"]
263    assert api.end_pairing_calls == 0
264
265
266async def test_pin_submitted_before_gesture(monkeypatch: pytest.MonkeyPatch) -> None:
267    """A PIN submitted while the gesture is pending is consumed once the client enters pairing."""
268    gesture = asyncio.Event()
269    api = _FakeServerApi([_desc(PairMethod.STATIC_PIN)], gesture=gesture)
270    provider, refreshed = _make_provider(api, monkeypatch)
271    monkeypatch.setattr(provider_module, "PIN_REQUEST_FEEDBACK_TIMEOUT", 0)
272    session = await provider.start_pin_pairing("c")
273    await asyncio.sleep(0)
274    assert (session.awaiting_gesture, session.awaiting_pin) == (True, True)
275
276    provider.submit_pin("c", "12345678")
277    assert (session.awaiting_gesture, session.awaiting_pin) == (True, False)
278    assert session.attempt_running
279
280    gesture.set()
281    assert session.task is not None
282    await session.task
283    assert not session.awaiting_gesture
284    assert session.finished
285    assert session.error is None
286    assert refreshed == ["c"]
287
288
289async def test_default_prefers_dynamic_pin(monkeypatch: pytest.MonkeyPatch) -> None:
290    """When both PIN methods are offered, the default pick is dynamic."""
291    api = _FakeServerApi([_desc(PairMethod.STATIC_PIN), _desc(PairMethod.DYNAMIC_PIN)])
292    provider, _refreshed = _make_provider(api, monkeypatch)
293    session = await provider.start_pin_pairing("c")
294    assert session.method is PairMethod.DYNAMIC_PIN
295    await provider.cancel_pin_pairing("c")
296
297
298async def test_static_override_picks_static_pin(monkeypatch: pytest.MonkeyPatch) -> None:
299    """The static override pairs with the static PIN even when dynamic is offered."""
300    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.STATIC_PIN)])
301    provider, refreshed = _make_provider(api, monkeypatch)
302    session = await provider.start_pin_pairing("c", static=True)
303    assert session.method is PairMethod.STATIC_PIN
304    await _submit_and_settle(provider, "12345678")
305    assert session.finished
306    assert session.error is None
307    assert refreshed == ["c"]
308
309
310async def test_static_override_requires_static_offer(monkeypatch: pytest.MonkeyPatch) -> None:
311    """The static override fails when the device does not offer a usable static PIN."""
312    api = _FakeServerApi(
313        [_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.STATIC_PIN, locked_out=True)]
314    )
315    provider, _refreshed = _make_provider(api, monkeypatch)
316    with pytest.raises(SecurityActionError) as excinfo:
317        await provider.start_pin_pairing("c", static=True)
318    assert excinfo.value.alert_key == "pairing_error_locked_out"
319
320
321async def test_pin_mismatch_is_retryable(monkeypatch: pytest.MonkeyPatch) -> None:
322    """A PIN mismatch leaves the session retryable, parked, with an idle deadline armed."""
323    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)])
324    api.outcomes.append(RemotePairingAbortError(PairAbortReason.PIN_MISMATCH))
325    provider, refreshed = _make_provider(api, monkeypatch)
326    session = await provider.start_pin_pairing("c")
327    await _submit_and_settle(provider, "000000")
328    assert session.can_retry
329    assert isinstance(session.error, RemotePairingAbortError)
330    assert _pin_idle_task_id("c") in _timers(provider)
331    assert api.end_pairing_calls == 0
332    assert refreshed == []
333    provider._cancel_pin_idle_timeout("c")
334
335
336async def test_retry_resumes_in_place(monkeypatch: pytest.MonkeyPatch) -> None:
337    """A same-mode retry reuses the session and can then succeed."""
338    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)])
339    api.outcomes.append(RemotePairingAbortError(PairAbortReason.PIN_MISMATCH))
340    provider, refreshed = _make_provider(api, monkeypatch)
341    session = await provider.start_pin_pairing("c", verify=True)
342    await _submit_and_settle(provider, "000000")
343    assert session.can_retry
344
345    same = await provider.start_pin_pairing("c", verify=True)
346    assert same is session
347    assert session.verify is True
348    assert session.method is PairMethod.DYNAMIC_PIN
349    assert session.error is None
350    assert not session.retryable
351    assert _pin_idle_task_id("c") not in _timers(provider)
352    assert session.awaiting_pin
353    # start_pin_pairing waited for the immediate PIN request, so no gesture is claimed.
354    assert not session.awaiting_gesture
355
356    await _submit_and_settle(provider, "123456")
357    assert session.finished
358    assert api.initiate_calls == 2
359    assert refreshed == ["c"]
360
361
362async def test_parked_mode_mismatch_restarts(monkeypatch: pytest.MonkeyPatch) -> None:
363    """A stale parked session never resumes under a different mode."""
364    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)])
365    api.outcomes.append(RemotePairingAbortError(PairAbortReason.PIN_MISMATCH))
366    provider, _refreshed = _make_provider(api, monkeypatch)
367    session = await provider.start_pin_pairing("c", verify=True)
368    await _submit_and_settle(provider, "000000")
369    assert session.can_retry
370
371    fresh = await provider.start_pin_pairing("c")
372    assert fresh is not session
373    assert fresh.verify is False
374    assert api.end_pairing_calls == 1
375    assert api.initiate_calls == 2
376
377
378async def test_parked_static_session_not_resumed_by_default_mode(
379    monkeypatch: pytest.MonkeyPatch,
380) -> None:
381    """A parked static-PIN session is not resumed by a later dynamic-first request."""
382    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN), _desc(PairMethod.STATIC_PIN)])
383    api.outcomes.append(RemotePairingAbortError(PairAbortReason.PIN_MISMATCH))
384    provider, _refreshed = _make_provider(api, monkeypatch)
385    session = await provider.start_pin_pairing("c", static=True)
386    assert session.method is PairMethod.STATIC_PIN
387    await _submit_and_settle(provider, "00000000")
388    assert session.can_retry
389
390    fresh = await provider.start_pin_pairing("c")
391    assert fresh is not session
392    assert fresh.method is PairMethod.DYNAMIC_PIN
393    assert api.end_pairing_calls == 1
394
395
396async def test_running_mode_mismatch_raises_concurrent(monkeypatch: pytest.MonkeyPatch) -> None:
397    """An attempt in flight with a different mode cannot be co-opted."""
398    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)])
399    provider, _refreshed = _make_provider(api, monkeypatch)
400    session = await provider.start_pin_pairing("c")
401    assert session.attempt_running
402
403    with pytest.raises(SecurityActionError) as excinfo:
404        await provider.start_pin_pairing("c", verify=True)
405    assert excinfo.value.alert_key == "pairing_error_concurrent"
406    assert provider.get_pin_session("c") is session
407    await provider.cancel_pin_pairing("c")
408
409
410async def test_retry_awaits_gesture_again(monkeypatch: pytest.MonkeyPatch) -> None:
411    """A retried attempt waits for the client's pair-init anew."""
412    gesture = asyncio.Event()
413    gesture.set()
414    api = _FakeServerApi([_desc(PairMethod.STATIC_PIN)], gesture=gesture)
415    api.outcomes.append(RemotePairingAbortError(PairAbortReason.PIN_MISMATCH))
416    provider, _refreshed = _make_provider(api, monkeypatch)
417    monkeypatch.setattr(provider_module, "PIN_REQUEST_FEEDBACK_TIMEOUT", 0)
418    session = await provider.start_pin_pairing("c")
419    await _submit_and_settle(provider, "00000000")
420    assert session.can_retry
421
422    gesture.clear()
423    same = await provider.start_pin_pairing("c")
424    assert same is session
425    await asyncio.sleep(0)
426    assert session.awaiting_gesture
427
428    gesture.set()
429    await _submit_and_settle(provider, "12345678")
430    assert session.finished
431    assert session.error is None
432
433
434async def test_locked_out_is_terminal(monkeypatch: pytest.MonkeyPatch) -> None:
435    """A locked-out abort is terminal and unparks the connection immediately."""
436    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)], await_pin=False)
437    api.outcomes.append(RemotePairingAbortError(PairAbortReason.LOCKED_OUT))
438    provider, _refreshed = _make_provider(api, monkeypatch)
439    session = await provider.start_pin_pairing("c")
440    assert session.task is not None
441    await session.task
442    assert session.finished
443    assert not session.can_retry
444    assert isinstance(session.error, RemotePairingAbortError)
445    assert _pin_idle_task_id("c") not in _timers(provider)
446    assert api.end_pairing_calls == 1
447
448
449async def test_local_user_cancel_records_no_error(monkeypatch: pytest.MonkeyPatch) -> None:
450    """Our own end_pairing cancel is not surfaced as an error."""
451    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)], await_pin=False)
452    api.outcomes.append(LocalPairingAbortError(PairAbortReason.USER_CANCELLED))
453    provider, refreshed = _make_provider(api, monkeypatch)
454    session = await provider.start_pin_pairing("c")
455    assert session.task is not None
456    await session.task
457    assert session.error is None
458    assert not session.retryable
459    assert session.finished
460    assert refreshed == []
461
462
463async def test_remote_user_cancel_is_retryable(monkeypatch: pytest.MonkeyPatch) -> None:
464    """A cancel initiated on the device is retryable from the operator's side."""
465    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)], await_pin=False)
466    api.outcomes.append(RemotePairingAbortError(PairAbortReason.USER_CANCELLED))
467    provider, _refreshed = _make_provider(api, monkeypatch)
468    session = await provider.start_pin_pairing("c")
469    assert session.task is not None
470    await session.task
471    assert session.can_retry
472    assert isinstance(session.error, RemotePairingAbortError)
473    assert _pin_idle_task_id("c") in _timers(provider)
474    provider._cancel_pin_idle_timeout("c")
475
476
477async def test_non_abort_failure_is_terminal(monkeypatch: pytest.MonkeyPatch) -> None:
478    """A non-abort failure is terminal and does not call end_pairing (server disconnected)."""
479    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)], await_pin=False)
480    api.outcomes.append(PairingError("boom"))
481    provider, refreshed = _make_provider(api, monkeypatch)
482    session = await provider.start_pin_pairing("c")
483    assert session.task is not None
484    await session.task
485    assert session.finished
486    assert not session.can_retry
487    assert isinstance(session.error, PairingError)
488    assert api.end_pairing_calls == 0
489    assert refreshed == []
490
491
492async def test_cancel_pin_pairing_ends_and_pops(monkeypatch: pytest.MonkeyPatch) -> None:
493    """Cancelling ends pairing, drops the session, and refreshes the player."""
494    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)])
495    provider, refreshed = _make_provider(api, monkeypatch)
496    session = await provider.start_pin_pairing("c")
497    assert session.awaiting_pin
498    await provider.cancel_pin_pairing("c")
499    assert provider.get_pin_session("c") is None
500    assert api.end_pairing_calls == 1
501    assert refreshed == ["c"]
502
503
504async def test_pair_with_token_failure_unparks(monkeypatch: pytest.MonkeyPatch) -> None:
505    """A failed token pairing unparks the connection and re-raises."""
506    api = _FakeServerApi([], await_pin=False)
507    api.outcomes.append(RemotePairingAbortError(PairAbortReason.PIN_MISMATCH))
508    provider, refreshed = _make_provider(api, monkeypatch)
509    monkeypatch.setattr(
510        provider_module,
511        "decode_token",
512        lambda _value: SimpleNamespace(client_id="c", pairing_psk=b"\x00" * 32),
513    )
514    with pytest.raises(RemotePairingAbortError):
515        await provider.pair_with_token("c", "tok")
516    assert api.end_pairing_calls == 1
517    assert refreshed == []
518
519
520async def test_pair_with_token_rejected_maps_to_pairing_error(
521    monkeypatch: pytest.MonkeyPatch,
522) -> None:
523    """A token the client does not accept surfaces as a friendly PairingError."""
524    api = _FakeServerApi([], await_pin=False)
525    api.outcomes.append(HandshakeAbortedError("expected Noise message 2 (TEXT), got CLOSE"))
526    provider, refreshed = _make_provider(api, monkeypatch)
527    monkeypatch.setattr(
528        provider_module,
529        "decode_token",
530        lambda _value: SimpleNamespace(client_id="c", pairing_psk=b"\x00" * 32),
531    )
532    with pytest.raises(PairingError, match="the token was rejected by the device"):
533        await provider.pair_with_token("c", "tok")
534    # The server already disconnected the client on a non-abort failure; nothing to unpark.
535    assert api.end_pairing_calls == 0
536    assert refreshed == []
537
538
539async def test_pair_with_token_malformed_token(monkeypatch: pytest.MonkeyPatch) -> None:
540    """A token that fails to decode surfaces as an invalid-token alert without pairing."""
541    api = _FakeServerApi([], await_pin=False)
542    provider, refreshed = _make_provider(api, monkeypatch)
543    with pytest.raises(SecurityActionError) as excinfo:
544        await provider.pair_with_token("c", "not-a-token")
545    assert excinfo.value.alert_key == "pairing_error_token_invalid"
546    assert api.initiate_calls == 0
547    assert refreshed == []
548
549
550async def test_pair_with_token_rejected_while_pin_attempt_runs(
551    monkeypatch: pytest.MonkeyPatch,
552) -> None:
553    """A token submitted while a PIN attempt is running is rejected without a second attempt."""
554    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)])
555    provider, refreshed = _make_provider(api, monkeypatch)
556    session = await provider.start_pin_pairing("c")
557    assert session.attempt_running
558    with pytest.raises(SecurityActionError) as excinfo:
559        await provider.pair_with_token("c", "tok")
560    assert excinfo.value.alert_key == "pairing_error_concurrent"
561    assert api.initiate_calls == 1
562    assert refreshed == []
563    await provider.cancel_pin_pairing("c")
564
565
566async def test_pair_with_token_success_refreshes(monkeypatch: pytest.MonkeyPatch) -> None:
567    """A successful token pairing refreshes the player without unparking."""
568    api = _FakeServerApi([], await_pin=False)
569    provider, refreshed = _make_provider(api, monkeypatch)
570    monkeypatch.setattr(
571        provider_module,
572        "decode_token",
573        lambda _value: SimpleNamespace(client_id="c", pairing_psk=b"\x00" * 32),
574    )
575    await provider.pair_with_token("c", "tok")
576    assert api.end_pairing_calls == 0
577    assert refreshed == ["c"]
578
579
580async def test_idle_timeout_restores_connection(monkeypatch: pytest.MonkeyPatch) -> None:
581    """An abandoned retryable session is terminated and the connection restored."""
582    api = _FakeServerApi([_desc(PairMethod.DYNAMIC_PIN)], await_pin=False)
583    provider, refreshed = _make_provider(api, monkeypatch)
584    monkeypatch.setattr(provider_module, "PIN_RETRY_IDLE_TIMEOUT", 0)
585    session = PinPairingSession(
586        client_id="c",
587        method=PairMethod.DYNAMIC_PIN,
588        pin_future=asyncio.get_running_loop().create_future(),
589        retryable=True,
590    )
591    provider._pin_sessions["c"] = session
592    provider._arm_pin_idle_timeout(session)
593    assert _pin_idle_task_id("c") in _timers(provider)
594    # The zero-delay timer fires on a later loop iteration, then spawns the idle task.
595    for _ in range(20):
596        if api.end_pairing_calls:
597            break
598        await asyncio.sleep(0)
599    assert api.end_pairing_calls == 1
600    assert isinstance(session.error, TimeoutError)
601    assert not session.retryable
602    assert refreshed == ["c"]
603    session.pin_future.cancel()
604