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