music-assistant-server

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