/
/
/
1"""Tests for how the open device-management section is rendered from the config snapshot."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import TYPE_CHECKING, cast
7
8from music_assistant.providers.sendspin.constants import (
9 CONF_ACTION_MANAGEMENT_ENTER,
10 CONF_ACTION_MANAGEMENT_EXIT,
11)
12from music_assistant.providers.sendspin.helpers import SecurityActionError
13from music_assistant.providers.sendspin.player import SendspinBasePlayer
14
15if TYPE_CHECKING:
16 from aiosendspin.models.management import ManagementResultData
17
18 from music_assistant.providers.sendspin.provider import SendspinProvider
19
20
21def _empty_config() -> ManagementResultData:
22 return cast(
23 "ManagementResultData",
24 SimpleNamespace(unpaired_access=None, static_pin=None, dynamic_pin=None),
25 )
26
27
28class _FakeProvider:
29 """Provider stand-in that records management-config fetches."""
30
31 def __init__(self, *, session: object | None, config: ManagementResultData | None) -> None:
32 self._session = session
33 self._config = config
34 self.fetch_calls = 0
35 self.exit_calls = 0
36
37 def get_management_session(self, client_id: str) -> object | None:
38 return self._session
39
40 async def management_get_pairing_config(self, client_id: str) -> ManagementResultData:
41 self.fetch_calls += 1
42 if self._config is None:
43 raise SecurityActionError("management_error_timeout")
44 return self._config
45
46 def exit_management(self, client_id: str) -> None:
47 self.exit_calls += 1
48
49
50def _player() -> SendspinBasePlayer:
51 player = SendspinBasePlayer.__new__(SendspinBasePlayer)
52 player._player_id = "client-1"
53 return player
54
55
56async def _paired(provider: _FakeProvider, snapshot: ManagementResultData | None) -> set[str]:
57 entries = await _player()._paired_entries(cast("SendspinProvider", provider), snapshot)
58 return {entry.key for entry in entries}
59
60
61async def test_renders_from_snapshot_without_fetch() -> None:
62 """An open session with a fresh snapshot renders from it, issuing no device round trip."""
63 provider = _FakeProvider(session=object(), config=None)
64 keys = await _paired(provider, _empty_config())
65 assert provider.fetch_calls == 0
66 assert CONF_ACTION_MANAGEMENT_EXIT in keys
67
68
69async def test_falls_back_to_fetch_when_snapshot_empty() -> None:
70 """A session open without a snapshot fetches once as a fallback."""
71 provider = _FakeProvider(session=object(), config=_empty_config())
72 keys = await _paired(provider, None)
73 assert provider.fetch_calls == 1
74 assert CONF_ACTION_MANAGEMENT_EXIT in keys
75
76
77async def test_fetch_failure_exits_and_offers_enter() -> None:
78 """A failed fallback fetch drops the session and falls through to the enter action."""
79 provider = _FakeProvider(session=object(), config=None)
80 keys = await _paired(provider, None)
81 assert provider.fetch_calls == 1
82 assert provider.exit_calls == 1
83 assert CONF_ACTION_MANAGEMENT_ENTER in keys
84 assert CONF_ACTION_MANAGEMENT_EXIT not in keys
85