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