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