/
/
/
1"""Tests for automatically approving devices that offer guest access."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from types import SimpleNamespace
8from typing import TYPE_CHECKING, cast
9
10from aiosendspin.noise.keys import generate_psk, psk_id_for
11from aiosendspin.noise.trust_store import (
12 InMemoryServerPairingStore,
13 PskCategory,
14 ServerPairingRecord,
15 TrustedUnpairedClient,
16)
17
18from music_assistant.providers.sendspin.provider import SendspinProvider
19
20from .test_pin_session import _FakeMass
21
22if TYPE_CHECKING:
23 from aiosendspin.server import SendspinServer
24 from aiosendspin.server.client import SendspinClient
25
26 from music_assistant.mass import MusicAssistant
27
28
29class _ApprovalServerApi:
30 """Server stand-in exposing just the pairing store and the trust call."""
31
32 def __init__(self) -> None:
33 self.pairing_store = InMemoryServerPairingStore()
34 self.trusted: list[str] = []
35
36 async def trust_unpaired(self, client_id: str) -> None:
37 await self.pairing_store.add_trusted_unpaired(TrustedUnpairedClient(client_id=client_id))
38 self.trusted.append(client_id)
39
40
41def _client(
42 *,
43 guest: bool = True,
44 roles: tuple[str, ...] = ("player@v1",),
45 psk_category: PskCategory | None = PskCategory.SENTINEL,
46) -> SendspinClient:
47 security = None if psk_category is None else SimpleNamespace(psk_category=psk_category)
48 return cast(
49 "SendspinClient",
50 SimpleNamespace(
51 info_or_none=SimpleNamespace(unpaired_access=SimpleNamespace(enabled=guest)),
52 negotiated_role_ids=list(roles),
53 connection_security=security,
54 ),
55 )
56
57
58def _make_provider(api: _ApprovalServerApi) -> SendspinProvider:
59 provider = SendspinProvider.__new__(SendspinProvider)
60 provider.mass = cast("MusicAssistant", _FakeMass(asyncio.get_running_loop()))
61 provider.server_api = cast("SendspinServer", api)
62 provider.logger = logging.getLogger("test.sendspin.guest")
63 provider._client_event_versions = {"c1": 1}
64 return provider
65
66
67async def test_a_guest_capable_player_is_approved_on_connect() -> None:
68 """The device plays without any setup step, so nothing is left for the user to decide."""
69 api = _ApprovalServerApi()
70 provider = _make_provider(api)
71
72 await provider._auto_trust_guest_access("c1", _client(), 1)
73
74 assert api.trusted == ["c1"]
75
76
77async def test_a_combo_with_an_audio_input_is_approved_too() -> None:
78 """Its playback roles still benefit, even though the input itself needs pairing."""
79 api = _ApprovalServerApi()
80 provider = _make_provider(api)
81
82 await provider._auto_trust_guest_access("c1", _client(roles=("player@v1", "source@v1")), 1)
83
84 assert api.trusted == ["c1"]
85
86
87async def test_a_capture_only_device_is_left_to_pair() -> None:
88 """Guest access grants a device whose every role needs pairing precisely nothing."""
89 api = _ApprovalServerApi()
90 provider = _make_provider(api)
91
92 await provider._auto_trust_guest_access("c1", _client(roles=("source@v1",)), 1)
93
94 assert api.trusted == []
95
96
97async def test_a_device_not_offering_guest_access_is_left_alone() -> None:
98 """Approving a device that never offered it would grant access it did not consent to."""
99 api = _ApprovalServerApi()
100 provider = _make_provider(api)
101
102 await provider._auto_trust_guest_access("c1", _client(guest=False), 1)
103
104 assert api.trusted == []
105
106
107async def test_an_existing_approval_is_not_rewritten() -> None:
108 """A second connect must not churn the store for a device already approved."""
109 api = _ApprovalServerApi()
110 await api.pairing_store.add_trusted_unpaired(TrustedUnpairedClient(client_id="c1"))
111 provider = _make_provider(api)
112
113 await provider._auto_trust_guest_access("c1", _client(), 1)
114
115 assert api.trusted == []
116
117
118async def test_a_live_pairing_skips_the_approval() -> None:
119 """A paired device needs no guest access; its long-term handshake already carries it."""
120 api = _ApprovalServerApi()
121 provider = _make_provider(api)
122
123 await provider._auto_trust_guest_access("c1", _client(psk_category=PskCategory.LONG_TERM), 1)
124
125 assert api.trusted == []
126
127
128async def test_a_stale_record_does_not_block_a_guest_handshake() -> None:
129 """
130 A client that lost its half reconnects as a guest, and must still be approved.
131
132 The server-side record outlives that loss, so keying the decision on the record
133 would leave the device unusable with no way back.
134 """
135 api = _ApprovalServerApi()
136 psk = generate_psk()
137 await api.pairing_store.store_record(
138 ServerPairingRecord(
139 psk_id=psk_id_for(psk), psk=psk, client_id="c1", pair_methods=[], owner=None
140 )
141 )
142 provider = _make_provider(api)
143
144 await provider._auto_trust_guest_access("c1", _client(), 1)
145
146 assert api.trusted == ["c1"]
147
148
149async def test_a_superseded_event_grants_nothing() -> None:
150 """The store reads suspend, so a reconnect in between invalidates the hello behind this."""
151 api = _ApprovalServerApi()
152 provider = _make_provider(api)
153 provider._client_event_versions["c1"] = 2
154
155 await provider._auto_trust_guest_access("c1", _client(), 1)
156
157 assert api.trusted == []
158