/
/
/
1"""Tests for utility/helper functions."""
2
3import logging
4import os
5import signal
6import subprocess
7import sys
8from ipaddress import IPv4Address, IPv6Address
9from pathlib import Path
10from unittest.mock import AsyncMock, MagicMock, patch
11
12import pytest
13from aiohttp.test_utils import make_mocked_request
14from music_assistant_models.enums import MediaType
15from music_assistant_models.errors import (
16 MusicAssistantError,
17 SetupFailedError,
18 UnsupportedSystemError,
19)
20from yarl import URL
21from zeroconf import InterfaceChoice, IPVersion
22
23from music_assistant.helpers import _ml_inference_probe, uri, util
24from music_assistant.helpers.aiohttp_client import encoded_request_url
25from music_assistant.helpers.webserver import Webserver, redact_sensitive_headers
26
27
28def test_redact_sensitive_headers() -> None:
29 """Credential-bearing request headers are redacted without hiding diagnostics."""
30 headers = {
31 "Accept": "application/json",
32 "Authorization": "Bearer secret-token",
33 "aUtHoRiZaTiOn-Extra": "secret-extra",
34 "PROXY-AUTHORIZATION": "Basic secret-proxy",
35 }
36
37 assert redact_sensitive_headers(headers) == {
38 "Accept": "application/json",
39 "Authorization": "<redacted>",
40 "aUtHoRiZaTiOn-Extra": "<redacted>",
41 "PROXY-AUTHORIZATION": "<redacted>",
42 }
43 assert "secret-token" not in str(redact_sensitive_headers(headers))
44 assert "secret-proxy" not in str(redact_sensitive_headers(headers))
45
46
47async def test_unhandled_request_log_redacts_sensitive_headers(
48 caplog: pytest.LogCaptureFixture,
49) -> None:
50 """The catch-all request log never includes an Authorization value."""
51 logger = logging.getLogger("test_webserver")
52 webserver = Webserver(logger, enable_dynamic_routes=True)
53 request = make_mocked_request(
54 "GET",
55 "/unknown",
56 headers={"Authorization": "Bearer secret-token", "Accept": "application/json"},
57 )
58
59 with caplog.at_level(logging.WARNING, logger=logger.name):
60 response = await webserver._handle_catch_all(request)
61
62 assert response.status == 404
63 assert "secret-token" not in caplog.text
64 assert "<redacted>" in caplog.text
65 assert "application/json" in caplog.text
66
67
68def test_version_extract() -> None:
69 """Test the extraction of version from title."""
70 test_str = "Bam Bam (feat. Ed Sheeran)"
71 title, version = util.parse_title_and_version(test_str)
72 assert title == "Bam Bam"
73 assert version == ""
74 test_str = "Bam Bam (feat. Ed Sheeran) - Karaoke Version"
75 title, version = util.parse_title_and_version(test_str)
76 assert title == "Bam Bam"
77 assert version == "Karaoke Version"
78 test_str = "Bam Bam (feat. Ed Sheeran) [Karaoke Version]"
79 title, version = util.parse_title_and_version(test_str)
80 assert title == "Bam Bam"
81 assert version == "Karaoke Version"
82 test_str = "SuperSong (2011 Remaster)"
83 title, version = util.parse_title_and_version(test_str)
84 assert title == "SuperSong"
85 assert version == "2011 Remaster"
86 test_str = "SuperSong (Live at Wembley)"
87 title, version = util.parse_title_and_version(test_str)
88 assert title == "SuperSong"
89 assert version == "Live at Wembley"
90 test_str = "SuperSong (Instrumental)"
91 title, version = util.parse_title_and_version(test_str)
92 assert title == "SuperSong"
93 assert version == "Instrumental"
94 test_str = "SuperSong (Explicit)"
95 title, version = util.parse_title_and_version(test_str)
96 assert title == "SuperSong"
97 assert version == ""
98 # Version keywords in main title should NOT be stripped (only in parentheses)
99 test_str = "Great live unplugged song"
100 title, version = util.parse_title_and_version(test_str)
101 assert title == "Great live unplugged song"
102 assert version == ""
103 test_str = "I Do (featuring Sonny of P.O.D.) (Album Version)"
104 title, version = util.parse_title_and_version(test_str)
105 assert title == "I Do"
106 assert version == "Album Version"
107 test_str = "Get Up Stand Up (Phunk Investigation instrumental club mix)"
108 title, version = util.parse_title_and_version(test_str)
109 assert title == "Get Up Stand Up"
110 assert version == "Phunk Investigation instrumental club mix"
111 # Complex case: non-version part + version part with 'mix' keyword
112 test_str = "Lovin' You More (That Big Track) (Mosquito Chillout mix)"
113 title, version = util.parse_title_and_version(test_str)
114 assert title == "Lovin' You More (That Big Track)"
115 assert version == "Mosquito Chillout mix"
116 # Nested parentheses inside the version should be preserved
117 test_str = "Fiji (Oliver Smith Remix (Mixed))"
118 title, version = util.parse_title_and_version(test_str)
119 assert title == "Fiji"
120 assert version == "Oliver Smith Remix (Mixed)"
121
122
123def test_version_extracts_multiple_qualifiers() -> None:
124 """All recognized title qualifiers contribute to the version."""
125 title, version = util.parse_title_and_version("( ) [Deluxe] [2022 Remaster]")
126
127 assert title == "( )"
128 assert version == "Deluxe 2022 Remaster"
129
130
131@pytest.mark.parametrize(
132 ("test_str", "expected"),
133 [
134 ("Barcelona (Special Edition - Deluxe)", ("Barcelona", "Special Edition - Deluxe")),
135 (
136 "All of Me (Tiësto's Birthday Treatment Remix - Radio Edit)",
137 ("All of Me", "Tiësto's Birthday Treatment Remix - Radio Edit"),
138 ),
139 (
140 "Crime Of The Century (2014 - HD Remaster)",
141 ("Crime Of The Century", "2014 - HD Remaster"),
142 ),
143 ("Song (Live) - Remastered 2011", ("Song", "Live Remastered 2011")),
144 ("Song (Remastered) (Remastered)", ("Song", "Remastered")),
145 ("Allejoppa - Extended [Extended]", ("Allejoppa", "Extended")),
146 ("Song - Single (Deluxe)", ("Song", "Deluxe Single")),
147 ],
148)
149def test_version_extract_sequential_passes(test_str: str, expected: tuple[str, str]) -> None:
150 """Later parsing passes see the title as reduced by earlier passes."""
151 assert util.parse_title_and_version(test_str) == expected
152
153
154def test_with_handling_in_titles() -> None:
155 """Test 'with' handling - preserved in title, stripped as featuring credit."""
156 # 'with you' (preserved as title word)
157 test_str = "CCF (I'm Gonna Stay with You)"
158 title, version = util.parse_title_and_version(test_str)
159 assert title == "CCF (I'm Gonna Stay with You)"
160 assert version == ""
161 # 'with someone' (preserved as title word)
162 test_str = "Ever Fallen in Love (With Someone You Shouldn't've)"
163 title, version = util.parse_title_and_version(test_str)
164 assert title == "Ever Fallen in Love (With Someone You Shouldn't've)"
165 assert version == ""
166 # 'with u' (preserved as title word)
167 test_str = "Dance (With U)"
168 title, version = util.parse_title_and_version(test_str)
169 assert title == "Dance (With U)"
170 assert version == ""
171 # 'with the' (preserved as title word)
172 test_str = "Girl (With the Patent Leather Face)"
173 title, version = util.parse_title_and_version(test_str)
174 assert title == "Girl (With the Patent Leather Face)"
175 assert version == ""
176 # 'with you' - different phrasing (preserved as title word)
177 test_str = "Rockin' Around (With You)"
178 title, version = util.parse_title_and_version(test_str)
179 assert title == "Rockin' Around (With You)"
180 assert version == ""
181 # 'with no' (preserved as title word)
182 test_str = "Ain't Gonna Bump No More (With No Big Fat Woman)"
183 title, version = util.parse_title_and_version(test_str)
184 assert title == "Ain't Gonna Bump No More (With No Big Fat Woman)"
185 assert version == ""
186 # 'with that' - not in WITH_TITLE_WORDS but not stripped because it doesn't start with "with "
187 test_str = "The Catastrophe (Good Luck with That Man)"
188 title, version = util.parse_title_and_version(test_str)
189 assert title == "The Catastrophe (Good Luck with That Man)"
190 assert version == ""
191 # 'with [artist name]' - should still be stripped (not a title word)
192 test_str = "Great Song (with John Smith)"
193 title, version = util.parse_title_and_version(test_str)
194 assert title == "Great Song"
195 assert version == ""
196 # 'with [artist name]' in brackets - should still be stripped
197 test_str = "Great Song [with Jane Doe]"
198 title, version = util.parse_title_and_version(test_str)
199 assert title == "Great Song"
200 assert version == ""
201 # Title word preserved + version extracted from dash notation
202 test_str = "CCF (I'm Gonna Stay with You) - Live Version"
203 title, version = util.parse_title_and_version(test_str)
204 assert title == "CCF (I'm Gonna Stay with You)"
205 assert version == "Live Version"
206 # Title word preserved + version extracted from brackets
207 test_str = "Dance (With U) [Remix]"
208 title, version = util.parse_title_and_version(test_str)
209 assert title == "Dance (With U)"
210 assert version == "Remix"
211
212
213async def test_uri_parsing() -> None:
214 """Test parsing of URI."""
215 # test regular uri
216 test_uri = "spotify://track/123456789"
217 media_type, provider, item_id = await uri.parse_uri(test_uri)
218 assert media_type == MediaType.TRACK
219 assert provider == "spotify"
220 assert item_id == "123456789"
221 # test spotify uri
222 test_uri = "spotify:track:123456789"
223 media_type, provider, item_id = await uri.parse_uri(test_uri)
224 assert media_type == MediaType.TRACK
225 assert provider == "spotify"
226 assert item_id == "123456789"
227 # test public play/open url
228 test_uri = "https://open.spotify.com/playlist/5lH9NjOeJvctAO92ZrKQNB?si=04a63c8234ac413e"
229 media_type, provider, item_id = await uri.parse_uri(test_uri)
230 assert media_type == MediaType.PLAYLIST
231 assert provider == "spotify"
232 assert item_id == "5lH9NjOeJvctAO92ZrKQNB"
233 # test filename with slashes as item_id
234 test_uri = "filesystem://track/Artist/Album/Track.flac"
235 media_type, provider, item_id = await uri.parse_uri(test_uri)
236 assert media_type == MediaType.TRACK
237 assert provider == "filesystem"
238 assert item_id == "Artist/Album/Track.flac"
239 # test regular url to builtin provider
240 test_uri = "http://radiostream.io/stream.mp3"
241 media_type, provider, item_id = await uri.parse_uri(test_uri)
242 assert media_type == MediaType.UNKNOWN
243 assert provider == "builtin"
244 assert item_id == "http://radiostream.io/stream.mp3"
245 # test local file to builtin provider
246 test_uri = __file__
247 media_type, provider, item_id = await uri.parse_uri(test_uri)
248 assert media_type == MediaType.UNKNOWN
249 assert provider == "builtin"
250 assert item_id == __file__
251 # test invalid uri
252 with pytest.raises(MusicAssistantError):
253 await uri.parse_uri("invalid://blah")
254
255
256async def test_apple_music_uri_parsing() -> None:
257 """Test parsing of Apple Music share URLs."""
258 # station â should resolve as PLAYLIST (is_dynamic)
259 media_type, provider, item_id = await uri.parse_uri(
260 "https://music.apple.com/de/station/dead-sara-essentials/ra.331701075"
261 )
262 assert media_type == MediaType.PLAYLIST
263 assert provider == "apple_music"
264 assert item_id == "ra.331701075"
265 # playlist
266 media_type, provider, item_id = await uri.parse_uri(
267 "https://music.apple.com/de/playlist/disturbed-essentials/pl.5d641aa29c5d4cc49b474d7d100996ec"
268 )
269 assert media_type == MediaType.PLAYLIST
270 assert provider == "apple_music"
271 assert item_id == "pl.5d641aa29c5d4cc49b474d7d100996ec"
272 # album
273 media_type, provider, item_id = await uri.parse_uri(
274 "https://music.apple.com/de/album/some-album/1234567890"
275 )
276 assert media_type == MediaType.ALBUM
277 assert provider == "apple_music"
278 assert item_id == "1234567890"
279 # artist
280 media_type, provider, item_id = await uri.parse_uri(
281 "https://music.apple.com/de/artist/dead-sara/123456789"
282 )
283 assert media_type == MediaType.ARTIST
284 assert provider == "apple_music"
285 assert item_id == "123456789"
286 # song
287 media_type, provider, item_id = await uri.parse_uri(
288 "https://music.apple.com/de/song/my-song/987654321"
289 )
290 assert media_type == MediaType.TRACK
291 assert provider == "apple_music"
292 assert item_id == "987654321"
293 # trailing slash stripped
294 media_type, provider, item_id = await uri.parse_uri(
295 "https://music.apple.com/de/station/some-station/ra.111222333/"
296 )
297 assert media_type == MediaType.PLAYLIST
298 assert item_id == "ra.111222333"
299 # query string stripped (non-track query params)
300 media_type, provider, item_id = await uri.parse_uri(
301 "https://music.apple.com/de/album/some-album/1234567890?itsct=music_box"
302 )
303 assert media_type == MediaType.ALBUM
304 assert item_id == "1234567890"
305 # track share link: album URL with ?i=<track_id>
306 media_type, provider, item_id = await uri.parse_uri(
307 "https://music.apple.com/de/album/some-album/1234567890?i=987654321"
308 )
309 assert media_type == MediaType.TRACK
310 assert provider == "apple_music"
311 assert item_id == "987654321"
312 # track share link with additional query params
313 media_type, _, item_id = await uri.parse_uri(
314 "https://music.apple.com/de/album/some-album/1234567890?itsct=music_box&i=111222333"
315 )
316 assert media_type == MediaType.TRACK
317 assert item_id == "111222333"
318
319
320def test_format_ip_for_url() -> None:
321 """Test IPv6 bracket wrapping for URLs (RFC 2732)."""
322 # IPv4 should pass through unchanged
323 assert util.format_ip_for_url("192.168.1.1") == "192.168.1.1"
324 assert util.format_ip_for_url("10.0.0.1") == "10.0.0.1"
325 assert util.format_ip_for_url("0.0.0.0") == "0.0.0.0"
326 # IPv6 should be wrapped in brackets
327 assert util.format_ip_for_url("::1") == "[::1]"
328 assert util.format_ip_for_url("fe80::1") == "[fe80::1]"
329 assert util.format_ip_for_url("2001:db8::1") == "[2001:db8::1]"
330 assert util.format_ip_for_url("fd00::cafe:1") == "[fd00::cafe:1]"
331
332
333def _mock_service_info(ipv4_addrs: list[str], ipv6_addrs: list[str]) -> MagicMock:
334 """Create a mock AsyncServiceInfo with ip_addresses_by_version."""
335 mock_info = MagicMock()
336
337 def ip_addresses_by_version(version: IPVersion) -> list[IPv4Address | IPv6Address]:
338 if version == IPVersion.V4Only:
339 return [IPv4Address(a) for a in ipv4_addrs]
340 if version == IPVersion.V6Only:
341 return [IPv6Address(a) for a in ipv6_addrs]
342 return [IPv4Address(a) for a in ipv4_addrs] + [IPv6Address(a) for a in ipv6_addrs]
343
344 mock_info.ip_addresses_by_version = ip_addresses_by_version
345 return mock_info
346
347
348def test_get_primary_ip_address_from_zeroconf_prefer_ipv4() -> None:
349 """Test zeroconf IP extraction preferring IPv4 (default)."""
350 mock_info = _mock_service_info(["192.168.1.100"], ["fd00::1"])
351 result = util.get_primary_ip_address_from_zeroconf(mock_info, prefer_ipv6=False)
352 assert result == "192.168.1.100"
353
354
355def test_get_primary_ip_address_from_zeroconf_prefer_ipv6() -> None:
356 """Test zeroconf IP extraction preferring IPv6."""
357 mock_info = _mock_service_info(["192.168.1.100"], ["fd00::1"])
358 result = util.get_primary_ip_address_from_zeroconf(mock_info, prefer_ipv6=True)
359 assert result == "fd00::1"
360
361
362def test_get_primary_ip_address_from_zeroconf_ipv6_fallback() -> None:
363 """Test zeroconf IP extraction falls back to IPv6 when no IPv4 available."""
364 mock_info = _mock_service_info([], ["fd00::1"])
365 result = util.get_primary_ip_address_from_zeroconf(mock_info, prefer_ipv6=False)
366 assert result == "fd00::1"
367
368
369def test_get_primary_ip_address_from_zeroconf_ipv4_fallback() -> None:
370 """Test zeroconf IP extraction falls back to IPv4 when no IPv6 available."""
371 mock_info = _mock_service_info(["192.168.1.100"], [])
372 result = util.get_primary_ip_address_from_zeroconf(mock_info, prefer_ipv6=True)
373 assert result == "192.168.1.100"
374
375
376def test_get_primary_ip_address_from_zeroconf_skips_link_local() -> None:
377 """Test zeroconf IP extraction skips loopback and link-local addresses."""
378 mock_info = _mock_service_info(
379 ["127.0.0.1", "169.254.1.1", "192.168.1.100"],
380 ["::1", "fe80::1", "fd00::1"],
381 )
382 # IPv4 preferred: should skip 127.x and 169.254.x
383 assert (
384 util.get_primary_ip_address_from_zeroconf(mock_info, prefer_ipv6=False) == "192.168.1.100"
385 )
386 # IPv6 preferred: should skip ::1 and fe80::
387 assert util.get_primary_ip_address_from_zeroconf(mock_info, prefer_ipv6=True) == "fd00::1"
388
389
390def test_get_primary_ip_address_from_zeroconf_no_addresses() -> None:
391 """Test zeroconf IP extraction returns None when no addresses available."""
392 mock_info = _mock_service_info([], [])
393 assert util.get_primary_ip_address_from_zeroconf(mock_info) is None
394 assert util.get_primary_ip_address_from_zeroconf(mock_info, prefer_ipv6=True) is None
395
396
397def _make_mock_adapter(
398 name: str,
399 ipv4_addrs: list[str] | None = None,
400 ipv6_addrs: list[tuple[str, int, int]] | None = None,
401) -> MagicMock:
402 """
403 Create a mock ifaddr.Adapter.
404
405 :param name: Adapter name.
406 :param ipv4_addrs: List of IPv4 address strings.
407 :param ipv6_addrs: List of (address, flowinfo, scope_id) tuples for IPv6.
408 """
409 adapter = MagicMock()
410 adapter.name = name
411 adapter.nice_name = name
412 ips = []
413 for addr in ipv4_addrs or []:
414 ip_mock = MagicMock()
415 ip_mock.is_IPv6 = False
416 ip_mock.ip = addr
417 ips.append(ip_mock)
418 for addr_tuple in ipv6_addrs or []:
419 ip_mock = MagicMock()
420 ip_mock.is_IPv6 = True
421 ip_mock.ip = addr_tuple
422 ips.append(ip_mock)
423 adapter.ips = ips
424 return adapter
425
426
427def test_get_zeroconf_args_dual_stack() -> None:
428 """Test zeroconf args on a dual-stack host."""
429 adapters = [
430 _make_mock_adapter("eth0", ["192.168.1.10"], [("fd00::1", 0, 2)]),
431 ]
432 with (
433 patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters),
434 patch("music_assistant.helpers.util.sys.platform", "linux"),
435 ):
436 result = util.get_zeroconf_args(use_all_interfaces=False)
437 assert result["ip_version"] == IPVersion.All
438 assert isinstance(result["interfaces"], list)
439 assert "192.168.1.10" in result["interfaces"]
440
441
442@pytest.mark.parametrize("platform", ["darwin", "freebsd14"])
443def test_get_zeroconf_args_dual_stack_ipv4_fallback(platform: str) -> None:
444 """Test that a dual-stack host falls back to IPv4-only on macOS/FreeBSD."""
445 adapters = [
446 _make_mock_adapter("eth0", ["192.168.1.10"], [("fd00::1", 0, 2)]),
447 ]
448 with (
449 patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters),
450 patch("music_assistant.helpers.util.sys.platform", platform),
451 ):
452 result = util.get_zeroconf_args(use_all_interfaces=False)
453 assert result["ip_version"] == IPVersion.V4Only
454 assert result["interfaces"] == InterfaceChoice.Default
455
456
457def test_get_zeroconf_args_ipv4_only() -> None:
458 """Test zeroconf args on an IPv4-only host."""
459 adapters = [
460 _make_mock_adapter("eth0", ["192.168.1.10"]),
461 ]
462 with patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters):
463 result = util.get_zeroconf_args(use_all_interfaces=False)
464 assert result["ip_version"] == IPVersion.V4Only
465 assert result["interfaces"] == InterfaceChoice.Default
466
467
468def test_get_zeroconf_args_ipv6_only() -> None:
469 """Test zeroconf args on an IPv6-only host."""
470 adapters = [
471 _make_mock_adapter("eth0", ipv6_addrs=[("fd00::1", 0, 2)]),
472 ]
473 with patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters):
474 result = util.get_zeroconf_args(use_all_interfaces=False)
475 assert result["ip_version"] == IPVersion.V6Only
476 assert isinstance(result["interfaces"], list)
477
478
479def test_get_zeroconf_args_skips_loopback() -> None:
480 """Test that loopback addresses are excluded from interface detection."""
481 adapters = [
482 _make_mock_adapter("lo", ["127.0.0.1"], [("::1", 0, 0)]),
483 _make_mock_adapter("eth0", ["192.168.1.10"]),
484 ]
485 with patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters):
486 result = util.get_zeroconf_args(use_all_interfaces=False)
487 # Should be IPv4-only (only loopback IPv6 found, which is excluded)
488 assert result["ip_version"] == IPVersion.V4Only
489
490
491def test_get_zeroconf_args_all_interfaces() -> None:
492 """Test zeroconf args with use_all_interfaces=True."""
493 adapters = [
494 _make_mock_adapter("eth0", ["192.168.1.10"], [("fd00::1", 0, 2)]),
495 ]
496 with (
497 patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters),
498 patch("music_assistant.helpers.util.sys.platform", "linux"),
499 ):
500 result = util.get_zeroconf_args(use_all_interfaces=True)
501 assert result["ip_version"] == IPVersion.All
502 assert isinstance(result["interfaces"], list)
503 assert "192.168.1.10" in result["interfaces"]
504
505
506def test_interface_name_for_ip_ipv4_match() -> None:
507 """An IPv4 address returns the name of the interface that holds it."""
508 adapters = [
509 _make_mock_adapter("lo", ["127.0.0.1"]),
510 _make_mock_adapter("eth0", ["192.168.1.10"]),
511 ]
512 with patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters):
513 assert util.interface_name_for_ip("192.168.1.10") == "eth0"
514
515
516def test_interface_name_for_ip_ipv6_match() -> None:
517 """An IPv6 address (stored as an (addr, flowinfo, scope_id) tuple) resolves by its address."""
518 adapters = [
519 _make_mock_adapter("eth0", ipv6_addrs=[("fd00::1", 0, 2)]),
520 ]
521 with patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters):
522 assert util.interface_name_for_ip("fd00::1") == "eth0"
523
524
525def test_interface_name_for_ip_no_match() -> None:
526 """An address that no interface holds returns None."""
527 adapters = [
528 _make_mock_adapter("eth0", ["192.168.1.10"]),
529 ]
530 with patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=adapters):
531 assert util.interface_name_for_ip("10.0.0.1") is None
532
533
534@pytest.mark.parametrize("capability", ["DEFAULT", "NO AVX"])
535def test_ml_inference_probe_rejects_no_avx2(capability: str) -> None:
536 """A no-AVX2 CPU is rejected before any kernel runs (safe to check in-process)."""
537 with patch("torch.backends.cpu.get_cpu_capability", return_value=capability):
538 assert _ml_inference_probe.run_probe() == _ml_inference_probe.PROBE_NO_AVX2
539
540
541def test_ml_inference_probe_subprocess_runs_to_a_clean_verdict() -> None:
542 """
543 End-to-end: the probe runs out-of-process and exits with a defined verdict, never a crash.
544
545 Spawning a subprocess (rather than calling run_probe() inline) is the same isolation the
546 production check relies on, so a hypothetical native crash on a misconfigured host fails
547 this one test instead of taking down the session. On an x86 host with AVX2 this exercises
548 the kernels and returns PROBE_CAPABLE; elsewhere it returns PROBE_NO_AVX2.
549 """
550 result = subprocess.run( # noqa: S603
551 [sys.executable, "-m", _ml_inference_probe.__name__],
552 capture_output=True,
553 timeout=120,
554 check=False,
555 )
556 assert result.returncode in (
557 _ml_inference_probe.PROBE_CAPABLE,
558 _ml_inference_probe.PROBE_NO_AVX2,
559 ), result.stderr.decode()
560
561
562@pytest.mark.parametrize(
563 ("returncode", "translation_key"),
564 [
565 (_ml_inference_probe.PROBE_CAPABLE, None),
566 (_ml_inference_probe.PROBE_NO_AVX2, "unsupported_system_avx2"),
567 (-signal.SIGILL, "unsupported_system_ml_inference_failed"),
568 (-signal.SIGSEGV, "unsupported_system_ml_inference_failed"),
569 (-signal.SIGABRT, "unsupported_system_ml_inference_failed"),
570 (-signal.SIGKILL, None), # external/OOM kill is not a CPU fault -> fail open
571 (1, None), # unexpected clean exit -> fail open
572 (None, None), # spawn failure or timeout -> fail open
573 ],
574)
575async def test_verify_cpu_supports_ml_inference_x86(
576 returncode: int | None, translation_key: str | None
577) -> None:
578 """On x86 the probe verdict vetoes only on no-AVX2 or a fatal signal; anything else fails open."""
579 with (
580 patch("music_assistant.helpers.util.platform.machine", return_value="x86_64"),
581 patch(
582 "music_assistant.helpers.util._run_ml_inference_probe",
583 AsyncMock(return_value=returncode),
584 ),
585 ):
586 if translation_key is not None:
587 with pytest.raises(UnsupportedSystemError) as err:
588 await util.verify_cpu_supports_ml_inference()
589 assert err.value.translation_key == translation_key
590 assert err.value.translation_args == []
591 else:
592 await util.verify_cpu_supports_ml_inference()
593
594
595async def test_verify_cpu_supports_ml_inference_arm() -> None:
596 """ARM machines pass without spawning the probe (QNNPACK backend works there)."""
597 with (
598 patch("music_assistant.helpers.util.platform.machine", return_value="aarch64"),
599 patch("music_assistant.helpers.util._run_ml_inference_probe", AsyncMock()) as probe,
600 ):
601 await util.verify_cpu_supports_ml_inference()
602 probe.assert_not_called()
603
604
605@pytest.mark.parametrize(
606 ("returncode", "expected"),
607 [
608 (-signal.SIGILL, -signal.SIGILL),
609 (0, 0),
610 ],
611)
612async def test_run_ml_inference_probe_returncode(returncode: int, expected: int) -> None:
613 """The probe runner reports the subprocess exit code (negative when a signal killed it)."""
614 proc = AsyncMock()
615 proc.wait = AsyncMock(return_value=returncode)
616 proc.returncode = returncode
617 with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
618 assert await util._run_ml_inference_probe() == expected
619
620
621async def test_run_ml_inference_probe_spawn_failure() -> None:
622 """A spawn failure is reported as None so the caller fails open."""
623 with patch("asyncio.create_subprocess_exec", AsyncMock(side_effect=OSError("boom"))):
624 assert await util._run_ml_inference_probe() is None
625
626
627async def test_run_ml_inference_probe_timeout() -> None:
628 """A probe that overruns the timeout is killed and reported as None."""
629 proc = AsyncMock()
630 proc.kill = MagicMock()
631 with (
632 patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)),
633 patch("asyncio.wait_for", AsyncMock(side_effect=TimeoutError)),
634 ):
635 assert await util._run_ml_inference_probe() is None
636 proc.kill.assert_called_once()
637
638
639def test_unsupported_system_error_is_setup_failed() -> None:
640 """UnsupportedSystemError must subclass SetupFailedError so existing handling applies."""
641 assert issubclass(UnsupportedSystemError, SetupFailedError)
642
643
644@pytest.mark.parametrize(
645 ("cpu_cores", "min_cpu_cores", "should_raise"),
646 [
647 (4, 4, False),
648 (8, 4, False),
649 (2, 4, True),
650 (1, 4, True),
651 (1, 0, False), # 0 disables the check
652 ],
653)
654async def test_verify_system_meets_requirements_cpu(
655 cpu_cores: int, min_cpu_cores: int, should_raise: bool
656) -> None:
657 """The CPU-core gate raises UnsupportedSystemError below the minimum."""
658 with (
659 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=cpu_cores),
660 patch("music_assistant.helpers.util.get_total_system_memory", return_value=64.0),
661 ):
662 if should_raise:
663 with pytest.raises(UnsupportedSystemError):
664 await util.verify_system_meets_requirements(
665 feature_name="X", min_cpu_cores=min_cpu_cores
666 )
667 else:
668 await util.verify_system_meets_requirements(
669 feature_name="X", min_cpu_cores=min_cpu_cores
670 )
671
672
673@pytest.mark.parametrize(
674 ("total_gb", "min_memory_gb", "should_raise"),
675 [
676 (8.0, 8.0, False),
677 (16.0, 8.0, False),
678 (4.0, 6.0, True),
679 (3.5, 6.0, True),
680 # the gate applies the reporting tolerance: 3.8GB clears a 4GB minimum (within 8%),
681 # 3.6GB does not (below the 3.68GB floor) -- guards against reverting to strict `<`
682 (3.8, 4.0, False),
683 (3.6, 4.0, True),
684 (0.0, 8.0, False), # 0.0 == unknown memory -> fail open, never block
685 (2.0, 0.0, False), # 0 disables the check
686 ],
687)
688async def test_verify_system_meets_requirements_memory(
689 total_gb: float, min_memory_gb: float, should_raise: bool
690) -> None:
691 """The RAM gate raises below the minimum (within tolerance) but fails open when unknown (0.0)."""
692 with (
693 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=16),
694 patch("music_assistant.helpers.util.get_total_system_memory", return_value=total_gb),
695 ):
696 if should_raise:
697 with pytest.raises(UnsupportedSystemError):
698 await util.verify_system_meets_requirements(
699 feature_name="X", min_memory_gb=min_memory_gb
700 )
701 else:
702 await util.verify_system_meets_requirements(
703 feature_name="X", min_memory_gb=min_memory_gb
704 )
705
706
707@pytest.mark.parametrize(
708 ("total_gb", "target_gb", "expected"),
709 [
710 (4.0, 4.0, True),
711 (3.8, 4.0, True), # a "4GB" host reports ~3.8GB -> still meets a 4GB target
712 (3.7, 4.0, True), # just above the 8% tolerance floor (3.68GB)
713 (3.6, 4.0, False), # below the tolerance floor
714 (7.7, 8.0, True), # an "8GB" host reporting ~7.7GB meets an 8GB target
715 (7.3, 8.0, False), # below the 8GB tolerance floor (7.36GB)
716 (0.0, 4.0, True), # unknown memory -> fail open
717 (2.0, 0.0, True), # no requirement -> always met
718 ],
719)
720def test_meets_memory_target(total_gb: float, target_gb: float, expected: bool) -> None:
721 """A nominal RAM target is met within the reporting tolerance; unknown/zero fail open."""
722 assert util.meets_memory_target(total_gb, target_gb) is expected
723
724
725async def test_verify_system_meets_requirements_ml_inference() -> None:
726 """require_ml_inference runs the capability probe after the RAM/CPU checks."""
727 with (
728 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=16),
729 patch("music_assistant.helpers.util.get_total_system_memory", return_value=64.0),
730 patch("music_assistant.helpers.util.platform.machine", return_value="x86_64"),
731 patch(
732 "music_assistant.helpers.util._run_ml_inference_probe",
733 AsyncMock(return_value=_ml_inference_probe.PROBE_NO_AVX2),
734 ),
735 ):
736 # capable RAM/CPU but the probe rejects: only raises when the ML check is requested
737 with pytest.raises(UnsupportedSystemError):
738 await util.verify_system_meets_requirements(
739 feature_name="X", min_cpu_cores=4, min_memory_gb=8.0, require_ml_inference=True
740 )
741 await util.verify_system_meets_requirements(
742 feature_name="X", min_cpu_cores=4, min_memory_gb=8.0
743 )
744
745
746async def test_unsupported_system_error_translation() -> None:
747 """Each raise path carries the right translation key + ordered args (feature name first)."""
748 with (
749 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=2),
750 patch("music_assistant.helpers.util.get_total_system_memory", return_value=64.0),
751 pytest.raises(UnsupportedSystemError) as cpu_err,
752 ):
753 await util.verify_system_meets_requirements(feature_name="Smart Fades", min_cpu_cores=4)
754 assert cpu_err.value.translation_key == "unsupported_system_cpu_cores"
755 assert cpu_err.value.translation_args == ["Smart Fades", 4, 2]
756
757 with (
758 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=16),
759 patch("music_assistant.helpers.util.get_total_system_memory", return_value=2.0),
760 pytest.raises(UnsupportedSystemError) as mem_err,
761 ):
762 await util.verify_system_meets_requirements(feature_name="Smart Fades", min_memory_gb=8.0)
763 assert mem_err.value.translation_key == "unsupported_system_memory"
764 assert mem_err.value.translation_args == ["Smart Fades", "8", "2.0"]
765
766 with (
767 patch("music_assistant.helpers.util.platform.machine", return_value="x86_64"),
768 patch(
769 "music_assistant.helpers.util._run_ml_inference_probe",
770 AsyncMock(return_value=_ml_inference_probe.PROBE_NO_AVX2),
771 ),
772 pytest.raises(UnsupportedSystemError) as avx_err,
773 ):
774 await util.verify_cpu_supports_ml_inference()
775 assert avx_err.value.translation_key == "unsupported_system_avx2"
776 assert avx_err.value.translation_args == []
777
778
779@pytest.mark.parametrize(
780 ("cpu_cores", "total_gb", "expected"),
781 [
782 (4, 6.0, True), # meets both recommended thresholds
783 (8, 16.0, True),
784 (2, 6.0, False), # below recommended cores
785 (4, 4.0, False), # below recommended RAM
786 (4, 0.0, True), # unknown memory -> fail open, same as the gate
787 ],
788)
789def test_system_meets_requirements(cpu_cores: int, total_gb: float, expected: bool) -> None:
790 """The non-raising predicate mirrors the gate's RAM/CPU checks, failing open on unknown RAM."""
791 with (
792 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=cpu_cores),
793 patch("music_assistant.helpers.util.get_total_system_memory", return_value=total_gb),
794 ):
795 assert util.system_meets_requirements(min_memory_gb=6.0, min_cpu_cores=4) is expected
796
797
798@pytest.mark.parametrize(
799 ("machine", "expected"),
800 [
801 ("aarch64", True),
802 ("arm64", True),
803 ("armv7l", True),
804 ("x86_64", False),
805 ("AMD64", False),
806 ],
807)
808def test_is_arm(machine: str, expected: bool) -> None:
809 """is_arm recognizes 32/64-bit ARM and rejects x86."""
810 with patch("music_assistant.helpers.util.platform.machine", return_value=machine):
811 assert util.is_arm() is expected
812
813
814@pytest.mark.parametrize(
815 ("cpu_count", "expected"),
816 [(1, 1), (2, 1), (4, 1), (8, 2), (12, 3), (32, 8)],
817)
818def test_inference_thread_budget(cpu_count: int, expected: int) -> None:
819 """The inference thread budget is a quarter of the cores, never below one."""
820 with (
821 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=cpu_count),
822 patch.dict(os.environ, {}, clear=False),
823 ):
824 os.environ.pop("OMP_NUM_THREADS", None)
825 assert util.inference_thread_budget() == expected
826
827
828def test_inference_thread_budget_follows_operator_override() -> None:
829 """An operator-supplied OMP_NUM_THREADS becomes the torch budget too, so the two agree."""
830 with (
831 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=32),
832 patch.dict(os.environ, {"OMP_NUM_THREADS": "2"}, clear=False),
833 ):
834 assert util.inference_thread_budget() == 2
835
836
837def test_cap_native_thread_pools_sets_env() -> None:
838 """The native pool caps are published to the environment for load-time pickup."""
839 env_vars = (
840 "OMP_NUM_THREADS",
841 "OPENBLAS_NUM_THREADS",
842 "MKL_NUM_THREADS",
843 "NUMEXPR_NUM_THREADS",
844 "VECLIB_MAXIMUM_THREADS",
845 )
846 with (
847 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=8),
848 patch.dict(os.environ, dict.fromkeys(env_vars, ""), clear=False),
849 ):
850 for env_var in env_vars:
851 del os.environ[env_var]
852 assert util.cap_native_thread_pools() == 2
853 for env_var in env_vars:
854 assert os.environ[env_var] == "2"
855
856
857def test_cap_native_thread_pools_respects_operator_value() -> None:
858 """An operator-supplied cap is kept, reported back, and applied to the other pools."""
859 with (
860 patch("music_assistant.helpers.util.os.process_cpu_count", return_value=8),
861 patch.dict(os.environ, {"OMP_NUM_THREADS": "1"}, clear=False),
862 ):
863 os.environ.pop("OPENBLAS_NUM_THREADS", None)
864 assert util.cap_native_thread_pools() == 1
865 assert os.environ["OMP_NUM_THREADS"] == "1"
866 assert os.environ["OPENBLAS_NUM_THREADS"] == "1"
867
868
869# 4/8/2 GiB expressed in bytes, for cgroup fixture files.
870_GIB = 1024**3
871
872
873@pytest.mark.parametrize(
874 ("raw", "expected"),
875 [
876 (str(4 * _GIB), 4.0),
877 (str(8 * _GIB), 8.0),
878 ("max", None), # v2 unlimited sentinel
879 ("", None), # empty file
880 (str(1 << 62), None), # v1 unlimited sentinel
881 ("0", None), # zero is not a real limit
882 ("-1", None), # negative is not a real limit
883 ("not-a-number", None),
884 ],
885)
886def test_read_cgroup_limit_file(tmp_path: Path, raw: str, expected: float | None) -> None:
887 """A cgroup limit file parses to GB, treating max/sentinel/garbage as no limit."""
888 limit_file = tmp_path / "memory.max"
889 limit_file.write_text(raw)
890 assert util._read_cgroup_limit_file(str(limit_file)) == expected
891
892
893def test_read_cgroup_limit_file_missing(tmp_path: Path) -> None:
894 """A missing cgroup limit file yields None rather than raising."""
895 assert util._read_cgroup_limit_file(str(tmp_path / "absent")) is None
896
897
898def test_cgroup_limit_v2(tmp_path: Path) -> None:
899 """Cgroup v2 memory.max at the mount root is read (namespaced container case)."""
900 (tmp_path / "memory.max").write_text(str(4 * _GIB))
901 # proc file absent -> rel is None -> falls back to the mount root.
902 limit = util._get_cgroup_memory_limit_gb(
903 cgroup_root=str(tmp_path), proc_cgroup=str(tmp_path / "absent")
904 )
905 assert limit == 4.0
906
907
908def test_cgroup_limit_v2_uses_min_across_hierarchy(tmp_path: Path) -> None:
909 """The effective v2 limit is the smallest memory.max across the cgroup and its ancestors."""
910 (tmp_path / "memory.max").write_text(str(8 * _GIB)) # root cap
911 leaf = tmp_path / "leaf"
912 leaf.mkdir()
913 (leaf / "memory.max").write_text(str(2 * _GIB)) # tighter leaf cap wins
914 proc = tmp_path / "proc_cgroup"
915 proc.write_text("0::/leaf\n")
916 limit = util._get_cgroup_memory_limit_gb(cgroup_root=str(tmp_path), proc_cgroup=str(proc))
917 assert limit == 2.0
918
919
920def test_cgroup_limit_v2_walks_ancestors(tmp_path: Path) -> None:
921 """A parent slice's memory.max caps the limit even when the leaf cgroup is unlimited."""
922 leaf = tmp_path / "system.slice" / "ma.service"
923 leaf.mkdir(parents=True)
924 (leaf / "memory.max").write_text("max") # leaf is unlimited...
925 (tmp_path / "system.slice" / "memory.max").write_text(str(4 * _GIB)) # ...ancestor caps it
926 (tmp_path / "memory.max").write_text("max")
927 proc = tmp_path / "proc_cgroup"
928 proc.write_text("0::/system.slice/ma.service\n")
929 limit = util._get_cgroup_memory_limit_gb(cgroup_root=str(tmp_path), proc_cgroup=str(proc))
930 assert limit == 4.0
931
932
933def test_cgroup_limit_v1_fallback(tmp_path: Path) -> None:
934 """With no v2 file, the v1 memory controller limit is used."""
935 mem = tmp_path / "memory"
936 mem.mkdir()
937 (mem / "memory.limit_in_bytes").write_text(str(4 * _GIB))
938 proc = tmp_path / "proc_cgroup"
939 proc.write_text("3:memory:/\n")
940 limit = util._get_cgroup_memory_limit_gb(cgroup_root=str(tmp_path), proc_cgroup=str(proc))
941 assert limit == 4.0
942
943
944def test_cgroup_limit_none_when_unset(tmp_path: Path) -> None:
945 """No cgroup files present -> no limit detected."""
946 assert (
947 util._get_cgroup_memory_limit_gb(
948 cgroup_root=str(tmp_path), proc_cgroup=str(tmp_path / "absent")
949 )
950 is None
951 )
952
953
954@pytest.mark.parametrize(
955 ("host_gb", "cgroup_gb", "platform", "expected"),
956 [
957 (16.0, 4.0, "linux", 4.0), # container limit below host -> use the limit
958 (8.0, 16.0, "linux", 8.0), # limit above host -> host wins
959 (8.0, None, "linux", 8.0), # no limit -> host RAM
960 (0.0, 4.0, "linux", 0.0), # host unknown -> unknown (fail open)
961 (8.0, 4.0, "darwin", 8.0), # non-linux never consults cgroups
962 ],
963)
964def test_get_total_system_memory(
965 host_gb: float, cgroup_gb: float | None, platform: str, expected: float
966) -> None:
967 """Total memory is min(host RAM, cgroup limit) on Linux; host RAM elsewhere."""
968 with (
969 patch("music_assistant.helpers.util._get_host_memory_gb", return_value=host_gb),
970 patch("music_assistant.helpers.util._get_cgroup_memory_limit_gb", return_value=cgroup_gb),
971 patch("music_assistant.helpers.util.sys.platform", platform),
972 ):
973 assert util.get_total_system_memory() == expected
974
975
976@pytest.mark.parametrize(
977 ("url", "expected"),
978 [
979 # plain URLs are left as strings for yarl to normalise
980 ("http://host/path", "http://host/path"),
981 ("http://host/path?a=1&b=2", "http://host/path?a=1&b=2"),
982 # already-encoded URLs are wrapped so yarl keeps the escapes verbatim
983 ("http://host/stream?token=ab%2Fcd", URL("http://host/stream?token=ab%2Fcd", encoded=True)),
984 ("http://host/with%20space", URL("http://host/with%20space", encoded=True)),
985 ],
986)
987def test_encoded_request_url(url: str, expected: str | URL) -> None:
988 """A pre-encoded URL is preserved as-is; a plain URL is left untouched."""
989 result = encoded_request_url(url)
990 assert result == expected
991 assert type(result) is type(expected)
992 # the percent-escapes must survive intact for auth-bearing stream URLs
993 assert str(result) == url
994