music-assistant-server
41.6 KB•PY
test_util.py
41.6 KB • 1,017 lines • python
1"""Tests for music_assistant.helpers.util helpers."""
2
3import asyncio
4import codecs
5import contextlib
6import gc
7import logging
8import socket
9import threading
10import time
11from collections.abc import Iterator
12from unittest.mock import MagicMock, patch
13
14import ifaddr
15import pytest
16from music_assistant_models.enums import MediaType
17from music_assistant_models.errors import ProviderUnavailableError
18from music_assistant_models.media_items import Album, ItemMapping, ProviderMapping, Track
19
20from music_assistant.helpers import util
21from music_assistant.helpers.util import (
22 detect_charset,
23 get_source_ip_for_target,
24 guard_single_request,
25 import_module_in_thread,
26 is_port_in_use,
27 join_task,
28 load_provider_module,
29 sanitize_http_header_value,
30 select_free_port,
31)
32from music_assistant.mass import MusicAssistant
33from music_assistant.models.music_provider import MusicProvider
34from tests.common import collect_loop_errors
35
36GUARDED_PROVIDER_ID = "test_guarded_prov"
37
38# a CUE sheet as Russian rips ship them: ASCII keywords with only the titles in
39# the local ANSI codepage (support #6093)
40CYRILLIC_CUE = """REM GENRE "Punk Rock"
41REM DATE 2002
42PERFORMER "ÐоÑÐ¾Ð»Ñ Ð¸ ШÑÑ"
43TITLE "Ðак в ÑÑаÑой Ñказке"
44FILE "CDImage.ape" WAVE
45 TRACK 01 AUDIO
46 TITLE "ÐÑоклÑÑÑй ÑÑаÑÑй дом"
47 INDEX 01 00:00:00
48"""
49
50
51def _cue_sheet(performer: str, title: str, track: str) -> str:
52 """Build a CUE sheet with the same ASCII-heavy shape as CYRILLIC_CUE."""
53 return (
54 'REM GENRE "Rock"\n'
55 "REM DATE 1998\n"
56 f'PERFORMER "{performer}"\n'
57 f'TITLE "{title}"\n'
58 'FILE "CDImage.ape" WAVE\n'
59 " TRACK 01 AUDIO\n"
60 f' TITLE "{track}"\n'
61 " INDEX 01 00:00:00\n"
62 )
63
64
65# the other codepages rippers wrote; every one of these sits next to a neighbour
66# that decodes the same bytes into plausible but wrong text
67LEGACY_CUES = {
68 "cp1251": CYRILLIC_CUE,
69 "koi8-r": _cue_sheet("ÐкваÑиÑм", "Ð ÑÑÑкий алÑбом", "ÐикиÑа Ð ÑзанÑкий"),
70 "cp1250": _cue_sheet("Kabát", "Šťastný ÄlovÄk", "ZůstaÅ"),
71 "cp1252": _cue_sheet("Björk", "Homogenic", "Jóga"),
72 # the dotless i is what a detector has to get right to tell cp1254 from cp1252
73 "cp1254": _cue_sheet("BarıŠManço", "MaÄusa'da", "Gülpembe"), # noqa: RUF001
74 "cp1253": _cue_sheet("ÎÎ¯ÎºÎ·Ï ÎεοδÏÏάκηÏ", "Îξιον ÎÏÏί", "Îνα Ïο ÏελιδÏνι"),
75 "cp1255": _cue_sheet("×¢××× ×¨××××", "×פר×××§× ×©× ×¢××× ×¨××××", "××××"),
76 "cp1257": _cue_sheet("PrÄta VÄtra", "LupatkÄjis", "JÅ«ra"),
77 # a multi-byte charset, where a wrong guess costs whole characters rather than
78 # single letters
79 "gbk": _cue_sheet("å¨æ°ä¼¦", "å¶æ ç¾", "ä¸é£ç ´"),
80}
81
82
83class TestDetectCharset:
84 """detect_charset names the charset raw text has to be decoded with."""
85
86 async def test_ascii_and_utf8_are_taken_as_utf8(self) -> None:
87 """Anything that is already valid UTF-8 needs no detection."""
88 assert await detect_charset(b'TITLE "Greatest Hits"') == "utf-8"
89 assert await detect_charset(CYRILLIC_CUE.encode()) == "utf-8"
90
91 async def test_byte_order_mark_is_stripped(self) -> None:
92 """A UTF-8 BOM must not survive into the decoded text."""
93 raw = codecs.BOM_UTF8 + CYRILLIC_CUE.encode()
94 encoding = await detect_charset(raw)
95 assert raw.decode(encoding) == CYRILLIC_CUE
96
97 @pytest.mark.parametrize("charset", list(LEGACY_CUES))
98 async def test_legacy_charsets_survive_a_round_trip(self, charset: str) -> None:
99 """
100 Text in a legacy charset comes back readable instead of as replacement chars.
101
102 Mostly-ASCII files such as CUE sheets hold very little non-ASCII text, so the
103 charset has to be resolved from a thin sample rather than given up on and
104 decoded as UTF-8 (support #6093). The round trip is what is asserted, not the
105 charset name, because neighbouring codepages decode these bytes identically.
106 """
107 source = LEGACY_CUES[charset]
108 raw = source.encode(charset)
109 assert raw.decode(await detect_charset(raw)) == source
110
111 async def test_undetectable_data_uses_the_fallback(self) -> None:
112 """Bytes that hold no readable text at all fall back to the given charset."""
113 assert await detect_charset(b"\xff\x00\xff", fallback="cp1257") == "cp1257"
114
115 async def test_declared_charset_wins_over_detection(self) -> None:
116 """A charset the source declares itself beats guessing at the bytes."""
117 raw = CYRILLIC_CUE.encode("cp1251")
118 assert await detect_charset(raw, preferred="cp1251") == "cp1251"
119
120 async def test_byte_order_mark_beats_the_declared_charset(self) -> None:
121 """A source declaring plain utf-8 must not leave its own BOM in the text."""
122 raw = codecs.BOM_UTF8 + CYRILLIC_CUE.encode()
123 encoding = await detect_charset(raw, preferred="utf-8")
124 assert raw.decode(encoding) == CYRILLIC_CUE
125
126 async def test_unknown_declared_charset_is_ignored(self) -> None:
127 """A charset name Python has no codec for must not reach decode()."""
128 raw = CYRILLIC_CUE.encode("cp1251")
129 encoding = await detect_charset(raw, preferred="utf8mb4")
130 assert raw.decode(encoding) == CYRILLIC_CUE
131
132 @pytest.mark.parametrize("charset", ["base64", "zlib", "rot_13", "idna", "undefined"])
133 async def test_declared_charset_that_cannot_decode_text_is_ignored(self, charset: str) -> None:
134 """A charset name that resolves to a codec but cannot decode text is ignored."""
135 raw = CYRILLIC_CUE.encode("cp1251")
136 encoding = await detect_charset(raw, preferred=charset)
137 assert raw.decode(encoding) == CYRILLIC_CUE
138
139
140class TestGetSourceIpForTarget:
141 """get_source_ip_for_target reports the interface the routing table egresses from."""
142
143 @pytest.mark.asyncio
144 async def test_returns_routing_lookup_result(self) -> None:
145 """The address the kernel picks for the target is handed back as-is."""
146 with patch("music_assistant.helpers.util.socket.socket") as mock_socket:
147 sock = mock_socket.return_value.__enter__.return_value
148 sock.getsockname.return_value = ("10.10.20.106", 0)
149 result = await get_source_ip_for_target("10.10.20.31")
150 assert result == "10.10.20.106"
151
152 @pytest.mark.asyncio
153 async def test_returns_empty_string_when_unroutable(self) -> None:
154 """A target with no route resolves to nothing rather than a guess."""
155 with patch("music_assistant.helpers.util.socket.socket") as mock_socket:
156 sock = mock_socket.return_value.__enter__.return_value
157 sock.connect.side_effect = OSError("no route to host")
158 result = await get_source_ip_for_target("10.10.20.31")
159 assert result == ""
160
161
162class TestSelectFreePort:
163 """select_free_port hands out distinct ports even under concurrent calls."""
164
165 @pytest.fixture(autouse=True)
166 def _clear_reservations(self) -> Iterator[None]:
167 """Ensure no port reservation state leaks between tests."""
168 util._reserved_ports.clear()
169 yield
170 util._reserved_ports.clear()
171
172 @pytest.mark.asyncio
173 async def test_concurrent_calls_get_distinct_ports(self) -> None:
174 """Instances starting simultaneously must not be handed the same port."""
175 # All ports report free, mimicking instances that haven't bound yet.
176 with patch("music_assistant.helpers.util.is_port_in_use", return_value=False):
177 ports = await asyncio.gather(*(select_free_port(38800, 38900) for _ in range(5)))
178 assert len(set(ports)) == len(ports)
179
180 @pytest.mark.asyncio
181 async def test_expired_reservation_is_reusable(self) -> None:
182 """A reservation past its TTL is released so the port can be handed out again."""
183 with patch("music_assistant.helpers.util.is_port_in_use", return_value=False):
184 first = await select_free_port(38800, 38900)
185 # force the reservation to look expired so the next call can reuse it
186 util._reserved_ports[first] = 0.0
187 second = await select_free_port(38800, 38900)
188 assert first == second
189
190 @pytest.mark.asyncio
191 async def test_host_is_passed_to_port_probe(self) -> None:
192 """A bind address is forwarded to the availability probe."""
193 with patch("music_assistant.helpers.util.is_port_in_use", return_value=False) as probe:
194 port = await select_free_port(38800, 38900, host="127.0.0.1")
195 probe.assert_awaited_once_with(port, host="127.0.0.1")
196
197 @pytest.mark.asyncio
198 async def test_exhausted_range_error_mentions_inclusive_range(self) -> None:
199 """The exhausted-range error names the searched range with an inclusive end."""
200 with (
201 patch("music_assistant.helpers.util.is_port_in_use", return_value=True),
202 pytest.raises(OSError, match=r"38800-38809$"),
203 ):
204 await select_free_port(38800, 38810)
205
206
207class TestIsPortInUse:
208 """is_port_in_use can probe the exact address a server will bind."""
209
210 @pytest.mark.asyncio
211 async def test_bound_loopback_port_is_in_use(self) -> None:
212 """An active IPv4 loopback listener is detected on that same address."""
213 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
214 sock.bind(("127.0.0.1", 0))
215 sock.listen(1)
216 assert await is_port_in_use(sock.getsockname()[1], host="127.0.0.1")
217
218 @pytest.mark.asyncio
219 async def test_released_loopback_port_is_free(self) -> None:
220 """A port without a listener on the probed address is reported free."""
221 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
222 sock.bind(("127.0.0.1", 0))
223 port = sock.getsockname()[1]
224 assert not await is_port_in_use(port, host="127.0.0.1")
225
226 @pytest.mark.asyncio
227 @pytest.mark.parametrize(
228 ("host", "family"),
229 [("127.0.0.1", socket.AF_INET), ("::1", socket.AF_INET6)],
230 )
231 async def test_host_selects_matching_address_family(
232 self, host: str, family: socket.AddressFamily
233 ) -> None:
234 """A specific bind address is probed with its matching socket family."""
235 with patch("music_assistant.helpers.util.socket.socket") as socket_mock:
236 await is_port_in_use(38800, host=host)
237 socket_mock.assert_called_once_with(family, socket.SOCK_STREAM)
238 socket_mock.return_value.__enter__.return_value.bind.assert_called_once_with((host, 38800))
239
240
241class TestImportModuleInThread:
242 """import_module_in_thread imports off the event loop, one import at a time."""
243
244 @pytest.mark.asyncio
245 async def test_module_is_returned(self) -> None:
246 """A relative module name is resolved against the given package."""
247 assert await import_module_in_thread(".util", "music_assistant.helpers") is util
248
249 @pytest.mark.asyncio
250 async def test_concurrent_imports_are_serialized(self) -> None:
251 """Concurrent calls never have two imports in flight (which can deadlock)."""
252 in_flight = 0
253 max_in_flight = 0
254
255 def _slow_import(*_args: object) -> MagicMock:
256 nonlocal in_flight, max_in_flight
257 in_flight += 1
258 max_in_flight = max(max_in_flight, in_flight)
259 time.sleep(0.05)
260 in_flight -= 1
261 return MagicMock()
262
263 with patch("music_assistant.helpers.util.importlib.import_module", _slow_import):
264 await asyncio.gather(*(import_module_in_thread(f"module_{idx}") for idx in range(5)))
265
266 assert max_in_flight == 1
267
268 @pytest.mark.asyncio
269 async def test_module_lock_collision_is_retried_once(self) -> None:
270 """A deadlock reported by the import machinery is retried, not passed on."""
271 module = MagicMock()
272 attempts = 0
273
274 def _import(*_args: object) -> MagicMock:
275 nonlocal attempts
276 attempts += 1
277 if attempts == 1:
278 raise RuntimeError("deadlock detected by _ModuleLock('requests.structures')")
279 return module
280
281 with patch("music_assistant.helpers.util.importlib.import_module", _import):
282 assert await import_module_in_thread("some_module") is module
283 assert attempts == 2
284
285 @pytest.mark.asyncio
286 async def test_other_runtime_errors_are_not_retried(self) -> None:
287 """An unrelated RuntimeError from the module body is passed on as-is."""
288 with (
289 patch(
290 "music_assistant.helpers.util.importlib.import_module",
291 side_effect=RuntimeError("boom"),
292 ) as import_mock,
293 pytest.raises(RuntimeError, match="boom"),
294 ):
295 await import_module_in_thread("some_module")
296 assert import_mock.call_count == 1
297
298
299class TestLoadProviderModule:
300 """load_provider_module verifies pinned requirements before importing the provider."""
301
302 @pytest.fixture(autouse=True)
303 def _clear_checked_requirements(self) -> Iterator[None]:
304 """Ensure no requirement-check state leaks between tests."""
305 util._checked_requirements.clear()
306 yield
307 util._checked_requirements.clear()
308
309 @pytest.mark.asyncio
310 async def test_requirement_with_extras_not_reinstalled(self) -> None:
311 """A requirement with extras is version-checked on the bare package name."""
312 with (
313 patch(
314 "music_assistant.helpers.util.get_package_version", return_value="6.1.1"
315 ) as version_mock,
316 patch("music_assistant.helpers.util.install_package") as install_mock,
317 patch("music_assistant.helpers.util.importlib.import_module"),
318 ):
319 await load_provider_module("sendspin", ["aiosendspin[server]==6.1.1"])
320 version_mock.assert_awaited_once_with("aiosendspin")
321 install_mock.assert_not_awaited()
322
323 @pytest.mark.asyncio
324 async def test_outdated_requirement_installed_with_extras_preserved(self) -> None:
325 """An outdated requirement is (re)installed with the full requirement string."""
326 with (
327 patch("music_assistant.helpers.util.get_package_version", return_value="6.0.0"),
328 patch("music_assistant.helpers.util.install_package") as install_mock,
329 patch("music_assistant.helpers.util.importlib.import_module"),
330 ):
331 await load_provider_module("sendspin", ["aiosendspin[server]==6.1.1"])
332 install_mock.assert_awaited_once_with("aiosendspin[server]==6.1.1")
333
334 @pytest.mark.asyncio
335 async def test_requirement_checked_only_once(self) -> None:
336 """Repeated loads of the same provider don't re-run the version check."""
337 with (
338 patch(
339 "music_assistant.helpers.util.get_package_version", return_value="6.1.1"
340 ) as version_mock,
341 patch("music_assistant.helpers.util.install_package") as install_mock,
342 patch("music_assistant.helpers.util.importlib.import_module"),
343 ):
344 await load_provider_module("sendspin", ["aiosendspin[server]==6.1.1"])
345 await load_provider_module("sendspin", ["aiosendspin[server]==6.1.1"])
346 version_mock.assert_awaited_once()
347 install_mock.assert_not_awaited()
348
349 @pytest.mark.asyncio
350 async def test_failed_install_is_retried_on_next_load(self) -> None:
351 """A failed install is not marked as checked, so the next load retries it."""
352 with (
353 patch("music_assistant.helpers.util.get_package_version", return_value=None),
354 patch(
355 "music_assistant.helpers.util.install_package",
356 side_effect=RuntimeError("install failed"),
357 ) as install_mock,
358 patch("music_assistant.helpers.util.importlib.import_module"),
359 ):
360 with pytest.raises(RuntimeError):
361 await load_provider_module("sendspin", ["aiosendspin[server]==6.1.1"])
362 install_mock.side_effect = None
363 await load_provider_module("sendspin", ["aiosendspin[server]==6.1.1"])
364 assert install_mock.await_count == 2
365
366
367class TestGetIpAddresses:
368 """get_ip_addresses caches the (expensive) adapter enumeration for a short while."""
369
370 @pytest.fixture(autouse=True)
371 def _clean_cache(self) -> Iterator[None]:
372 """Run every test against an empty module-level cache."""
373 util._ip_addresses_cache.clear()
374 util._ip_addresses_pending.clear()
375 yield
376 util._ip_addresses_cache.clear()
377 util._ip_addresses_pending.clear()
378
379 @pytest.fixture
380 def enumerate_mock(self) -> Iterator[MagicMock]:
381 """Replace the blocking adapter enumeration with a counting fake."""
382 with patch(
383 "music_assistant.helpers.util._enumerate_ip_addresses",
384 return_value=("192.168.1.10",),
385 ) as mock:
386 yield mock
387
388 def test_falls_back_to_loopback_without_routable_addresses(self) -> None:
389 """With no routable addresses at all, loopback is returned instead of an empty tuple."""
390 with patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=[]):
391 assert util._enumerate_ip_addresses(True, False) == ("127.0.0.1",)
392 assert util._enumerate_ip_addresses(True, True) == ("127.0.0.1",)
393
394 @pytest.mark.asyncio
395 async def test_concurrent_callers_share_a_single_probe(self, enumerate_mock: MagicMock) -> None:
396 """Concurrent callers within the TTL all get the result of one enumeration."""
397 results = await asyncio.gather(*(util.get_ip_addresses() for _ in range(10)))
398 assert all(result == ("192.168.1.10",) for result in results)
399 assert enumerate_mock.call_count == 1
400
401 @pytest.mark.asyncio
402 async def test_sequential_calls_within_ttl_reuse_the_cache(
403 self, enumerate_mock: MagicMock
404 ) -> None:
405 """A repeated call shortly after the first is served from the cache."""
406 assert await util.get_ip_addresses() == ("192.168.1.10",)
407 assert await util.get_ip_addresses() == ("192.168.1.10",)
408 assert enumerate_mock.call_count == 1
409
410 @pytest.mark.asyncio
411 async def test_cache_is_kept_per_ipv6_flag(self, enumerate_mock: MagicMock) -> None:
412 """include_ipv6 True/False are distinct probes (each cached separately)."""
413 await util.get_ip_addresses(include_ipv6=False)
414 await util.get_ip_addresses(include_ipv6=True)
415 await util.get_ip_addresses(include_ipv6=True)
416 assert enumerate_mock.call_count == 2
417
418 @pytest.mark.asyncio
419 async def test_expired_cache_triggers_a_new_probe(self, enumerate_mock: MagicMock) -> None:
420 """Once the TTL passed, the next call enumerates the adapters again."""
421 await util.get_ip_addresses()
422 # age the cached entry beyond the TTL
423 cached_at, addresses = util._ip_addresses_cache[False, False]
424 util._ip_addresses_cache[False, False] = (
425 cached_at - util.IP_ADDRESSES_CACHE_TTL - 1,
426 addresses,
427 )
428 await util.get_ip_addresses()
429 assert enumerate_mock.call_count == 2
430
431 @pytest.mark.asyncio
432 async def test_cancelled_caller_does_not_break_concurrent_callers(self) -> None:
433 """Cancelling one caller must not cancel the shared probe for the others."""
434
435 def slow_enumerate(_include_ipv6: bool, _publish_candidates_only: bool) -> tuple[str, ...]:
436 time.sleep(0.1)
437 return ("192.168.1.10",)
438
439 with patch(
440 "music_assistant.helpers.util._enumerate_ip_addresses",
441 side_effect=slow_enumerate,
442 ) as enumerate_mock:
443 task_a = asyncio.create_task(util.get_ip_addresses())
444 task_b = asyncio.create_task(util.get_ip_addresses())
445 # let both callers await the (same) in-flight probe, then cancel one
446 await asyncio.sleep(0.02)
447 task_a.cancel()
448 with pytest.raises(asyncio.CancelledError):
449 await task_a
450 assert await task_b == ("192.168.1.10",)
451 assert enumerate_mock.call_count == 1
452
453 @pytest.mark.asyncio
454 async def test_cancelled_caller_of_a_failing_probe_logs_no_loop_error(self) -> None:
455 """A probe failing after a caller gave up is not reported to the loop handler."""
456 release = threading.Event()
457
458 def failing_enumerate(
459 _include_ipv6: bool, _publish_candidates_only: bool
460 ) -> tuple[str, ...]:
461 release.wait()
462 raise OSError("probe failed")
463
464 with (
465 collect_loop_errors() as reported,
466 patch(
467 "music_assistant.helpers.util._enumerate_ip_addresses",
468 side_effect=failing_enumerate,
469 ) as enumerate_mock,
470 ):
471 task_a = asyncio.create_task(util.get_ip_addresses())
472 task_b = asyncio.create_task(util.get_ip_addresses())
473 # let both callers await the (same) in-flight probe, then cancel one
474 await asyncio.sleep(0)
475 task_a.cancel()
476 with pytest.raises(asyncio.CancelledError):
477 await task_a
478 # release the probe only once the cancellation is fully processed, so the
479 # failure reliably lands after the giving-up caller is gone
480 release.set()
481 with pytest.raises(OSError, match="probe failed"):
482 await task_b
483
484 assert enumerate_mock.call_count == 1
485 assert reported == []
486
487 @pytest.mark.asyncio
488 async def test_sole_cancelled_caller_of_a_failing_probe_logs_no_loop_error(self) -> None:
489 """A probe failing with no caller left to receive it is not reported either."""
490 release = threading.Event()
491
492 def failing_enumerate(
493 _include_ipv6: bool, _publish_candidates_only: bool
494 ) -> tuple[str, ...]:
495 release.wait()
496 raise OSError("probe failed")
497
498 with (
499 collect_loop_errors() as reported,
500 patch(
501 "music_assistant.helpers.util._enumerate_ip_addresses",
502 side_effect=failing_enumerate,
503 ),
504 ):
505 caller = asyncio.create_task(util.get_ip_addresses())
506 await asyncio.sleep(0)
507 probe = util._ip_addresses_pending[False, False]
508 caller.cancel()
509 with pytest.raises(asyncio.CancelledError):
510 await caller
511 release.set()
512 await asyncio.wait((probe,))
513 # drop the last reference so asyncio would report an unretrieved exception
514 del probe
515 gc.collect()
516 await asyncio.sleep(0)
517
518 assert reported == []
519
520
521def _adapter(name: str, *ips: str) -> ifaddr.Adapter:
522 """Build an adapter of the given name holding the given IPv4 addresses."""
523 return ifaddr.Adapter(name, name, [ifaddr.IP(ip, 24, name) for ip in ips])
524
525
526@contextlib.contextmanager
527def _fake_adapters(*adapters: ifaddr.Adapter) -> Iterator[None]:
528 """Enumerate the given adapters, with the primary-route probe failing on this host."""
529 with (
530 patch("music_assistant.helpers.util.ifaddr.get_adapters", return_value=list(adapters)),
531 patch("music_assistant.helpers.util.socket.socket") as mock_socket,
532 ):
533 # without a primary route every address is ranked on its prefix alone, so the
534 # outcome does not depend on the network the test host happens to sit on
535 mock_socket.return_value.connect.side_effect = OSError
536 yield
537
538
539class TestGetPublishIpCandidates:
540 """get_publish_ip_candidates skips the addresses no local network device can reach."""
541
542 @pytest.fixture(autouse=True)
543 def _clean_cache(self) -> Iterator[None]:
544 """Run every test against an empty module-level cache."""
545 util._ip_addresses_cache.clear()
546 util._ip_addresses_pending.clear()
547 yield
548 util._ip_addresses_cache.clear()
549 util._ip_addresses_pending.clear()
550
551 @pytest.mark.asyncio
552 async def test_container_and_tunnel_addresses_are_left_out(self) -> None:
553 """A HA OS host publishes its LAN address, not the docker bridges alongside it."""
554 with _fake_adapters(
555 _adapter("end0", "192.168.1.10"),
556 _adapter("hassio", "172.30.32.1"),
557 _adapter("docker0", "172.30.232.1"),
558 _adapter("tailscale0", "100.64.0.1"),
559 ):
560 assert await util.get_publish_ip_candidates() == ("192.168.1.10",)
561
562 @pytest.mark.asyncio
563 async def test_real_lan_bridge_is_kept(self) -> None:
564 """A host whose LAN lives on a bridge (Proxmox, Unraid, OpenWrt) still publishes it."""
565 with _fake_adapters(
566 _adapter("vmbr0", "192.168.1.10"),
567 _adapter("br-lan", "10.0.0.5"),
568 _adapter("br-1a2b3c4d5e6f", "172.18.0.1"),
569 ):
570 assert await util.get_publish_ip_candidates() == ("192.168.1.10", "10.0.0.5")
571
572 @pytest.mark.asyncio
573 async def test_host_behind_a_tunnel_only_still_publishes(self) -> None:
574 """With nothing but a tunnel to offer, that tunnel beats publishing nothing."""
575 with _fake_adapters(_adapter("wg0", "10.6.0.2")):
576 assert await util.get_publish_ip_candidates() == ("10.6.0.2",)
577
578 @pytest.mark.asyncio
579 async def test_unfiltered_lookup_keeps_the_container_addresses(self) -> None:
580 """get_ip_addresses is unaffected: the HA ingress site binds one of those bridges."""
581 with _fake_adapters(
582 _adapter("end0", "192.168.1.10"),
583 _adapter("hassio", "172.30.32.1"),
584 ):
585 assert await util.get_ip_addresses() == ("192.168.1.10", "172.30.32.1")
586 assert await util.get_publish_ip_candidates() == ("192.168.1.10",)
587
588
589class TestJoinTask:
590 """join_task waits for a task without adopting it, so a waiter never cancels the work."""
591
592 @pytest.mark.asyncio
593 async def test_returns_the_task_result(self) -> None:
594 """A completed task hands its result to the waiter."""
595 release = asyncio.Event()
596 release.set()
597 task = asyncio.create_task(_gated_task(release, "done"))
598 assert await join_task(task) == "done"
599
600 @pytest.mark.asyncio
601 async def test_cancelled_waiter_leaves_the_task_running(self) -> None:
602 """Cancelling one waiter must not disturb the task or the waiters that remain."""
603 release = asyncio.Event()
604 task = asyncio.create_task(_gated_task(release, "done"))
605 waiter_a = asyncio.create_task(join_task(task))
606 waiter_b = asyncio.create_task(join_task(task))
607 await asyncio.sleep(0)
608 waiter_a.cancel()
609 with pytest.raises(asyncio.CancelledError):
610 await waiter_a
611
612 release.set()
613 assert await waiter_b == "done"
614 assert not task.cancelled()
615
616 @pytest.mark.asyncio
617 async def test_cancelled_waiter_of_a_failing_task_logs_no_loop_error(self) -> None:
618 """A task failing after a waiter gave up is not reported to the loop handler."""
619 release = asyncio.Event()
620 with collect_loop_errors() as reported:
621 task = asyncio.create_task(_gated_task(release, "done", fail=True))
622 waiter_a = asyncio.create_task(join_task(task))
623 waiter_b = asyncio.create_task(join_task(task))
624 await asyncio.sleep(0)
625 waiter_a.cancel()
626 with pytest.raises(asyncio.CancelledError):
627 await waiter_a
628 # release the task only once the cancellation is fully processed, so the failure
629 # reliably lands after the giving-up waiter is gone
630 release.set()
631 with pytest.raises(RuntimeError, match="task failed"):
632 await waiter_b
633
634 assert reported == []
635
636 @pytest.mark.asyncio
637 async def test_timeout_leaves_the_task_running(self) -> None:
638 """Giving up on the timeout raises TimeoutError but keeps the task alive."""
639 release = asyncio.Event()
640 task = asyncio.create_task(_gated_task(release, "done"))
641 with pytest.raises(TimeoutError):
642 await join_task(task, timeout=0.01)
643 assert not task.done()
644
645 release.set()
646 assert await task == "done"
647
648
649class TestSanitizeHttpHeaderValue:
650 """sanitize_http_header_value strips characters aiohttp forbids in response headers."""
651
652 def test_clean_value_unchanged(self) -> None:
653 """A regular track name passes through untouched."""
654 assert sanitize_http_header_value("AC/DC - Thunderstruck") == "AC/DC - Thunderstruck"
655
656 def test_newline_and_carriage_return_replaced(self) -> None:
657 """CR/LF (the classic header injection vector) are replaced with spaces."""
658 assert sanitize_http_header_value("Artist -\r\nEvil: header") == "Artist - Evil: header"
659
660 def test_all_c0_control_chars_and_del_replaced(self) -> None:
661 r"""
662 Every char aiohttp's _FORBIDDEN_HEADER_CHARS_RE rejects is replaced.
663
664 Regression test for https://github.com/music-assistant/support/issues/5791
665 where a control char (other than \n, \r, \t) in a FLAC tag crashed
666 serve_queue_item_stream with a 500.
667 """
668 for codepoint in [*range(0x20), 0x7F]:
669 value = f"Artist - Some{chr(codepoint)}Track"
670 sanitized = sanitize_http_header_value(value)
671 assert sanitized == "Artist - Some Track", f"codepoint {codepoint:#04x} not replaced"
672
673 def test_non_ascii_preserved(self) -> None:
674 """Non-ASCII text is allowed in headers and must be preserved."""
675 assert sanitize_http_header_value("Björk - Jóga") == "Björk - Jóga"
676
677 def test_leading_trailing_whitespace_stripped(self) -> None:
678 """Control chars at the edges don't leave dangling whitespace."""
679 assert sanitize_http_header_value("\x00Artist - Track\x1f") == "Artist - Track"
680
681
682class TestGuardSingleRequest:
683 """guard_single_request collapses identical concurrent calls into a single request."""
684
685 @pytest.mark.asyncio
686 async def test_cancelled_caller_does_not_affect_others(
687 self, mass_minimal: MusicAssistant
688 ) -> None:
689 """A caller giving up must not cancel the shared request for the other callers."""
690 caller = _GuardedCaller(mass_minimal)
691 calls = [asyncio.create_task(caller.fetch("123")) for _ in range(3)]
692 # give the callers a chance to line up behind the same in-flight request
693 await asyncio.sleep(0)
694 assert caller.calls == 1
695
696 calls[0].cancel()
697 caller.release.set()
698 results = await asyncio.gather(*calls, return_exceptions=True)
699
700 assert isinstance(results[0], asyncio.CancelledError)
701 assert results[1:] == ["result-123", "result-123"]
702 assert caller.calls == 1
703
704 @pytest.mark.asyncio
705 async def test_failure_reaches_the_caller_without_being_logged(
706 self, mass_minimal: MusicAssistant, caplog: pytest.LogCaptureFixture
707 ) -> None:
708 """A failure raised at the caller is not also warned about as an unhandled one."""
709 caller = _GuardedCaller(mass_minimal)
710 caller.error = ProviderUnavailableError("some_provider is not available")
711 caller.release.set()
712
713 with pytest.raises(ProviderUnavailableError):
714 await caller.fetch("123")
715
716 # the task's done callback runs an iteration after the task itself finished
717 await asyncio.sleep(0)
718 await asyncio.sleep(0)
719 assert not [
720 record
721 for record in caplog.records
722 if record.levelno >= logging.WARNING and "Exception in task" in record.getMessage()
723 ]
724
725 @pytest.mark.asyncio
726 async def test_instances_get_their_own_request(self, mass_minimal: MusicAssistant) -> None:
727 """Two objects of the same class each issue their own request."""
728 first = _GuardedCaller(mass_minimal)
729 second = _GuardedCaller(mass_minimal)
730 calls = [
731 asyncio.create_task(first.fetch("123")),
732 asyncio.create_task(second.fetch("123")),
733 ]
734 # both requests are in flight before either is released
735 await asyncio.sleep(0)
736 first.release.set()
737 second.release.set()
738
739 assert await asyncio.gather(*calls) == ["result-123", "result-123"]
740 assert first.calls == 1
741 assert second.calls == 1
742
743 # mass.create_task pins its tasks to mass.loop, so the test must run on the loop the
744 # class-scoped fixture was created on
745 @pytest.mark.asyncio(loop_scope="class")
746 async def test_controllers_sharing_a_method_get_their_own_request(
747 self, music_mass_class: MusicAssistant
748 ) -> None:
749 """The same item id requested on two media controllers resolves per media type."""
750 provider = _GatedMusicProvider()
751 with patch.object(music_mass_class, "get_provider", return_value=provider):
752 album_calls = [
753 asyncio.create_task(
754 music_mass_class.music.albums.get_provider_item("123", GUARDED_PROVIDER_ID)
755 )
756 for _ in range(2)
757 ]
758 track_call = asyncio.create_task(
759 music_mass_class.music.tracks.get_provider_item("123", GUARDED_PROVIDER_ID)
760 )
761 await asyncio.sleep(0)
762 provider.release.set()
763 first_album, second_album = await asyncio.gather(*album_calls)
764 track = await track_call
765
766 assert isinstance(first_album, Album)
767 assert isinstance(track, Track)
768 # the two album callers shared a single request, the track caller got its own
769 assert first_album is second_album
770 assert provider.album_calls == 1
771 assert provider.track_calls == 1
772
773 @pytest.mark.asyncio
774 async def test_equal_media_item_arguments_share_a_request(
775 self, mass_minimal: MusicAssistant
776 ) -> None:
777 """Equal media items key the same however their set fields happen to iterate."""
778 caller = _GuardedCaller(mass_minimal)
779 mapping_ids = ("a", "b", "c", "d")
780 calls = [
781 asyncio.create_task(
782 caller.fetch_item("123", fallback=_guarded_album("Album", mapping_ids))
783 ),
784 asyncio.create_task(
785 caller.fetch_item("123", fallback=_guarded_album("Album", mapping_ids[::-1]))
786 ),
787 ]
788 await asyncio.sleep(0)
789 caller.release.set()
790
791 assert await asyncio.gather(*calls) == [f"123-{GUARDED_PROVIDER_ID}-False-Album"] * 2
792 assert caller.calls == 1
793
794 @pytest.mark.asyncio
795 async def test_media_item_arguments_key_on_their_uri(
796 self, mass_minimal: MusicAssistant
797 ) -> None:
798 """Media items for the same provider item share a request, contents aside."""
799 caller = _GuardedCaller(mass_minimal)
800 calls = [
801 asyncio.create_task(
802 caller.fetch_item("123", fallback=_guarded_album("First", ("a", "b")))
803 ),
804 asyncio.create_task(
805 caller.fetch_item("123", fallback=_guarded_album("Second", ("b", "a", "c")))
806 ),
807 ]
808 await asyncio.sleep(0)
809 caller.release.set()
810
811 # both fallbacks describe item 123, so the second caller joins and is answered
812 # with the fallback of the caller that started the request
813 assert await asyncio.gather(*calls) == [f"123-{GUARDED_PROVIDER_ID}-False-First"] * 2
814 assert caller.calls == 1
815
816 @pytest.mark.asyncio
817 async def test_item_mapping_and_full_item_get_their_own_request(
818 self, mass_minimal: MusicAssistant
819 ) -> None:
820 """A mapping and a full item for one item resolve differently, so they never share."""
821 caller = _GuardedCaller(mass_minimal)
822 calls = [
823 asyncio.create_task(caller.fetch_item("123", fallback=_guarded_album("Album", ("a",)))),
824 asyncio.create_task(caller.fetch_item("123", fallback=_guarded_mapping("Mapping"))),
825 ]
826 await asyncio.sleep(0)
827 caller.release.set()
828
829 assert await asyncio.gather(*calls) == [
830 f"123-{GUARDED_PROVIDER_ID}-False-Album",
831 f"123-{GUARDED_PROVIDER_ID}-False-Mapping",
832 ]
833 assert caller.calls == 2
834
835 @pytest.mark.asyncio
836 async def test_calls_spelled_differently_share_a_request(
837 self, mass_minimal: MusicAssistant
838 ) -> None:
839 """The same call shares a request however its arguments are spelled."""
840 caller = _GuardedCaller(mass_minimal)
841 calls = [
842 asyncio.create_task(caller.fetch_item("123")),
843 asyncio.create_task(caller.fetch_item("123", GUARDED_PROVIDER_ID)),
844 asyncio.create_task(
845 caller.fetch_item("123", provider=GUARDED_PROVIDER_ID, force_refresh=False)
846 ),
847 ]
848 await asyncio.sleep(0)
849 caller.release.set()
850
851 assert await asyncio.gather(*calls) == [f"123-{GUARDED_PROVIDER_ID}-False-None"] * 3
852 assert caller.calls == 1
853
854 @pytest.mark.asyncio
855 async def test_arguments_containing_punctuation_do_not_collide(
856 self, mass_minimal: MusicAssistant
857 ) -> None:
858 """Ids carrying punctuation must not run into the argument that follows them."""
859 caller = _GuardedCaller(mass_minimal)
860 calls = [
861 asyncio.create_task(caller.fetch_item("a.b", "c")),
862 asyncio.create_task(caller.fetch_item("a", "b.c")),
863 ]
864 await asyncio.sleep(0)
865 caller.release.set()
866
867 assert await asyncio.gather(*calls) == ["a.b-c-False-None", "a-b.c-False-None"]
868 assert caller.calls == 2
869
870 @pytest.mark.asyncio
871 async def test_force_refresh_gets_its_own_request(self, mass_minimal: MusicAssistant) -> None:
872 """A force refresh must issue its own request instead of joining a normal one."""
873 caller = _GuardedCaller(mass_minimal)
874 calls = [
875 asyncio.create_task(caller.fetch_item("123")),
876 asyncio.create_task(caller.fetch_item("123", force_refresh=True)),
877 ]
878 await asyncio.sleep(0)
879 caller.release.set()
880
881 assert await asyncio.gather(*calls) == [
882 f"123-{GUARDED_PROVIDER_ID}-False-None",
883 f"123-{GUARDED_PROVIDER_ID}-True-None",
884 ]
885 assert caller.calls == 2
886
887
888class _GuardedCaller:
889 """Minimal stand-in for a controller exposing a guarded request."""
890
891 def __init__(self, mass: MusicAssistant) -> None:
892 """
893 Initialize the caller.
894
895 :param mass: The MusicAssistant instance tracking the guarded request.
896 """
897 self.mass = mass
898 self.calls = 0
899 self.release = asyncio.Event()
900 self.error: Exception | None = None
901
902 @guard_single_request
903 async def fetch(self, item_id: str) -> str:
904 """Return the result for the given item id, once released."""
905 self.calls += 1
906 await self.release.wait()
907 if self.error is not None:
908 raise self.error
909 return f"result-{item_id}"
910
911 @guard_single_request
912 async def fetch_item(
913 self,
914 item_id: str,
915 provider: str = GUARDED_PROVIDER_ID,
916 force_refresh: bool = False,
917 fallback: Album | ItemMapping | None = None,
918 ) -> str:
919 """Return the result for the given item id, once released."""
920 self.calls += 1
921 await self.release.wait()
922 # the fallback is echoed so a caller receiving another caller's argument is visible
923 return f"{item_id}-{provider}-{force_refresh}-{fallback.name if fallback else None}"
924
925
926class _GatedMusicProvider(MusicProvider):
927 """Minimal music provider returning items only once released."""
928
929 def __init__(self) -> None:
930 """Initialize the provider."""
931 self.release = asyncio.Event()
932 self.album_calls = 0
933 self.track_calls = 0
934 self.config = MagicMock()
935 self.config.instance_id = GUARDED_PROVIDER_ID
936 self.manifest = MagicMock()
937 self.manifest.domain = GUARDED_PROVIDER_ID
938 self.logger = MagicMock()
939
940 async def get_album(self, prov_album_id: str) -> Album:
941 """Return the album for the given provider album id."""
942 self.album_calls += 1
943 await self.release.wait()
944 return Album(
945 item_id=prov_album_id,
946 provider=GUARDED_PROVIDER_ID,
947 name="Album",
948 provider_mappings={_provider_mapping(prov_album_id)},
949 )
950
951 async def get_track(self, prov_track_id: str) -> Track:
952 """Return the track for the given provider track id."""
953 self.track_calls += 1
954 await self.release.wait()
955 return Track(
956 item_id=prov_track_id,
957 provider=GUARDED_PROVIDER_ID,
958 name="Track",
959 provider_mappings={_provider_mapping(prov_track_id)},
960 )
961
962
963async def _gated_task(release: asyncio.Event, result: str, fail: bool = False) -> str:
964 """
965 Return (or raise) once the given event is set.
966
967 :param release: Event that lets the task complete.
968 :param result: Value to return.
969 :param fail: Raise instead of returning the result.
970 """
971 await release.wait()
972 if fail:
973 raise RuntimeError("task failed")
974 return result
975
976
977def _guarded_album(name: str, mapping_ids: tuple[str, ...]) -> Album:
978 """
979 Build an album on the gated test provider to pass as a fallback argument.
980
981 :param name: The album name.
982 :param mapping_ids: The provider item ids to add as provider mappings.
983 """
984 return Album(
985 item_id="123",
986 provider=GUARDED_PROVIDER_ID,
987 name=name,
988 provider_mappings={_provider_mapping(item_id) for item_id in mapping_ids},
989 )
990
991
992def _guarded_mapping(name: str) -> ItemMapping:
993 """
994 Build an item mapping on the gated test provider to pass as a fallback argument.
995
996 :param name: The item name.
997 """
998 return ItemMapping(
999 media_type=MediaType.ALBUM,
1000 item_id="123",
1001 provider=GUARDED_PROVIDER_ID,
1002 name=name,
1003 )
1004
1005
1006def _provider_mapping(item_id: str) -> ProviderMapping:
1007 """
1008 Build a provider mapping for the gated test provider.
1009
1010 :param item_id: The provider item id to map.
1011 """
1012 return ProviderMapping(
1013 item_id=item_id,
1014 provider_domain=GUARDED_PROVIDER_ID,
1015 provider_instance=GUARDED_PROVIDER_ID,
1016 )
1017