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