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    @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