/
/
/
1"""Unit tests for AirPlay provider helpers."""
2
3import plistlib
4from typing import Any
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8from aiohttp import ClientError
9
10from music_assistant.providers.airplay.helpers import (
11 _CLI_BINARY_CHECK_TIMEOUT,
12 default_buffer_depth,
13 get_cli_binary,
14 get_decoded_property,
15 probe_audio_formats,
16 serialize_txt_records,
17 supports_airplay2,
18 supports_companion_pairing,
19)
20
21ALAC_44100_24 = 1 << 19
22ALAC_48000_24 = 1 << 21
23
24
25def test_get_decoded_property_matches_case_insensitively() -> None:
26 """TXT record keys resolve regardless of the casing advertised on the wire."""
27 discovery_info = MagicMock()
28 discovery_info.decoded_properties = {"rpFl": "0x367A2", "SystemBuildVersion": "21K69"}
29
30 assert get_decoded_property(discovery_info, "rpFl") == "0x367A2"
31 assert get_decoded_property(discovery_info, "rpfl") == "0x367A2"
32 assert get_decoded_property(discovery_info, "systembuildversion") == "21K69"
33 assert get_decoded_property(discovery_info, "missing") is None
34
35
36@pytest.mark.parametrize(
37 ("properties", "expected"),
38 [
39 # Apple TV advertises its flags under the mixed-case wire key "rpFl"
40 ({"rpFl": "0x367A2"}, True),
41 # HomePod: PIN pairing not supported
42 ({"rpFl": "0x62792"}, False),
43 # pairing explicitly disabled
44 ({"rpFl": "0x367A6"}, False),
45 ({"rpFl": "invalid"}, False),
46 ({}, False),
47 ],
48)
49def test_supports_companion_pairing(properties: dict[str, str], expected: bool) -> None:
50 """Companion PIN pairing support is read from the wire-cased rpFl flags."""
51 discovery_info = MagicMock()
52 discovery_info.decoded_properties = properties
53 assert supports_companion_pairing(discovery_info) is expected
54
55
56def test_supports_companion_pairing_without_service() -> None:
57 """A device without a Companion service is never pairable."""
58 assert supports_companion_pairing(None) is False
59
60
61@pytest.mark.parametrize(
62 ("features_value", "expected"),
63 [
64 # Apple TV 4K: UnifiedMediaControl (38) and CoreUtils (48) set
65 ("0x4A7FDFD5,0x3C177FDE", True),
66 # Sonos: UnifiedMediaControl set
67 ("0x445F8A00,0x1C340", True),
68 # legacy RAOP-only receiver (single low field, bits 38/48 unreachable)
69 ("0x5A7FFFF7", False),
70 # high field present but neither AirPlay 2 bit set
71 ("0xFFFFFFFF,0x2", False),
72 # bit 48 (CoreUtils) alone is enough
73 ("0x0,0x10000", True),
74 # missing/garbage values never claim support
75 (None, False),
76 ("", False),
77 ("not-a-number", False),
78 ],
79)
80def test_supports_airplay2(features_value: str | None, expected: bool) -> None:
81 """AirPlay 2 support is detected from the UnifiedMediaControl/CoreUtils feature bits."""
82 assert supports_airplay2(features_value) is expected
83
84
85def test_serialize_txt_records() -> None:
86 """TXT records serialize to space-separated k=v pairs, skipping unsafe entries."""
87 discovery_info = MagicMock()
88 discovery_info.decoded_properties = {
89 "features": "0x5A7FFFF7,0x1E",
90 "flags": "0x4",
91 "deviceid": "AA:BB:CC:DD:EE:FF",
92 "manufacturer": "Acme, Inc.", # value contains a space: skipped
93 "odd key": "value", # key contains a space: skipped
94 "empty": None, # zeroconf may decode a valueless entry as None: skipped
95 }
96 txt = serialize_txt_records(discovery_info)
97 pairs = txt.split(" ")
98 assert "features=0x5A7FFFF7,0x1E" in pairs
99 assert "flags=0x4" in pairs
100 assert "deviceid=AA:BB:CC:DD:EE:FF" in pairs
101 assert len(pairs) == 3
102
103
104@pytest.mark.parametrize(
105 ("system", "machine", "expected_name"),
106 [
107 ("Linux", "arm64", "cliairplay-linux-aarch64"),
108 ("Linux", "amd64", "cliairplay-linux-x86_64"),
109 ("Darwin", "aarch64", "cliairplay-macos-arm64"),
110 ("Darwin", "x86_64", "cliairplay-macos-x86_64"),
111 ],
112)
113async def test_get_cli_binary_uses_release_asset_name(
114 system: str,
115 machine: str,
116 expected_name: str,
117 monkeypatch: pytest.MonkeyPatch,
118) -> None:
119 """
120 Resolve platform aliases to the filename installed by development setup.
121
122 The runtime lookup must use the exact release asset name for every supported alias.
123 """
124 check_output = AsyncMock(return_value=(0, b"cliairplay v0.1.0 check"))
125 monkeypatch.setattr("music_assistant.providers.airplay.helpers.platform.system", lambda: system)
126 monkeypatch.setattr(
127 "music_assistant.providers.airplay.helpers.platform.machine", lambda: machine
128 )
129 monkeypatch.setattr("music_assistant.providers.airplay.helpers.check_output", check_output)
130
131 result = await get_cli_binary()
132
133 assert result.endswith(f"/bin/{expected_name}")
134 check_output.assert_awaited_once_with(result, "--check", timeout=_CLI_BINARY_CHECK_TIMEOUT)
135
136
137async def test_get_cli_binary_raises_when_check_times_out(monkeypatch: pytest.MonkeyPatch) -> None:
138 """A binary that never answers --check surfaces as RuntimeError, not a hang."""
139 check_output = AsyncMock(side_effect=TimeoutError)
140 monkeypatch.setattr(
141 "music_assistant.providers.airplay.helpers.platform.system", lambda: "Darwin"
142 )
143 monkeypatch.setattr(
144 "music_assistant.providers.airplay.helpers.platform.machine", lambda: "aarch64"
145 )
146 monkeypatch.setattr("music_assistant.providers.airplay.helpers.check_output", check_output)
147
148 with pytest.raises(RuntimeError, match="did not respond to --check"):
149 await get_cli_binary()
150
151
152@pytest.mark.parametrize(
153 ("info", "expected_hires"),
154 [
155 # Apple TV: 24-bit listed for the buffered stream only
156 (
157 {
158 "supportedAudioFormatsExtended": {
159 "audioStream": [11, 18, 22, 24],
160 "bufferStream": [19, 21, 22, 23],
161 }
162 },
163 True,
164 ),
165 # Sonos: 16-bit only on both streams
166 (
167 {
168 "supportedAudioFormatsExtended": {
169 "audioStream": [18, 22],
170 "bufferStream": [18, 22],
171 }
172 },
173 False,
174 ),
175 # Mac: legacy mask for the realtime stream, extended table for the buffered one
176 (
177 {
178 "supportedFormats": {"audioStream": 0x1440800},
179 "supportedAudioFormatsExtended": {"bufferStream": [21, 22, 23]},
180 },
181 True,
182 ),
183 # receivers that publish no format tables at all
184 ({"deviceid": "AA:BB:CC:DD:EE:FF"}, False),
185 ],
186)
187async def test_probe_audio_formats(info: dict[str, Any], expected_hires: bool) -> None:
188 """The advertised formats are read from the union of both /info stream tables."""
189 mass = _mass_serving_info(info)
190
191 formats = await probe_audio_formats(mass, "10.0.0.5", 7000)
192
193 assert bool(formats & (ALAC_44100_24 | ALAC_48000_24)) is expected_hires
194 mass.http_session.get.assert_called_once()
195 assert mass.http_session.get.call_args.args[0] == "http://10.0.0.5:7000/info"
196
197
198async def test_probe_audio_formats_ipv6_address() -> None:
199 """An IPv6 receiver address is bracketed for the request URL."""
200 mass = _mass_serving_info({})
201
202 await probe_audio_formats(mass, "fe80::1", 7000)
203
204 assert mass.http_session.get.call_args.args[0] == "http://[fe80::1]:7000/info"
205
206
207@pytest.mark.parametrize(
208 "response",
209 [
210 ClientError("unreachable"),
211 TimeoutError,
212 b"not-a-plist",
213 404,
214 ],
215)
216async def test_probe_audio_formats_failures(response: Any) -> None:
217 """An unreachable, slow or nonsensical receiver reports no formats instead of raising."""
218 if isinstance(response, int):
219 mass = _mass_serving_info({}, status=response)
220 elif isinstance(response, bytes):
221 mass = _mass_serving_info({})
222 mass.http_session.get.return_value.__aenter__.return_value.read = AsyncMock(
223 return_value=response
224 )
225 else:
226 mass = MagicMock()
227 mass.http_session.get = MagicMock(side_effect=response)
228
229 assert await probe_audio_formats(mass, "10.0.0.5", 7000) == 0
230
231
232def _mass_serving_info(info: dict[str, Any], status: int = 200) -> MagicMock:
233 """Build a mass mock whose http session serves the given /info plist."""
234 response = MagicMock()
235 response.status = status
236 response.read = AsyncMock(return_value=plistlib.dumps(info, fmt=plistlib.FMT_BINARY))
237 context = MagicMock()
238 context.__aenter__ = AsyncMock(return_value=response)
239 context.__aexit__ = AsyncMock(return_value=False)
240 mass = MagicMock()
241 mass.http_session.get = MagicMock(return_value=context)
242 return mass
243
244
245def test_default_buffer_depth_family_table() -> None:
246 """
247 Every family defaults to Automatic since buffered became the auto-route.
248
249 The LinkPlay generations that used to need a deepened realtime queue
250 (WiiM 1750 ms, Edifier MS50A 2500 ms) manage their own buffer on the
251 buffered stream; the per-player depth setting remains as the manual
252 override for field reports.
253 """
254 assert (
255 default_buffer_depth("Linkplay Technology Inc.", "WiiM Pro Receiver", "p20.4.8.814756") == 0
256 )
257 assert default_buffer_depth("Edifier Inc", "Edifier MS50A", "p20.Linkplay.4.6.430230") == 0
258 assert default_buffer_depth("Sonos", "Era 100", "p20.96.0-79160") == 0
259 assert default_buffer_depth("Apple Inc.", "AppleTV11,1", None) == 0
260