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