music-assistant-server

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