music-assistant-server

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