/
/
1"""Tests for MusicProvider source-stream capacity."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import cast
7from unittest.mock import MagicMock
8
9import pytest
10from music_assistant_models.enums import ProviderType
11
12from music_assistant.models.music_provider import (
13 DEFAULT_MAX_CONCURRENT_STREAMS,
14 MusicProvider,
15 ProviderStreamLimitError,
16)
17
18
19class _StreamingProvider(MusicProvider):
20 """Streaming provider using the default source limit."""
21
22
23class _NonStreamingProvider(MusicProvider):
24 """Non-streaming provider with no source limit."""
25
26 @property
27 def is_streaming_provider(self) -> bool:
28 """Return False for this local provider."""
29 return False
30
31
32class _SingleStreamProvider(MusicProvider):
33 """Streaming provider with one source slot."""
34
35 @property
36 def max_concurrent_streams(self) -> int:
37 """Return one source slot."""
38 return 1
39
40
41def _make_provider[ProviderT: MusicProvider](provider_cls: type[ProviderT]) -> ProviderT:
42 """Construct a minimal MusicProvider instance."""
43 mass = MagicMock()
44 manifest = MagicMock()
45 manifest.type = ProviderType.MUSIC
46 manifest.domain = "test_provider"
47 manifest.name = "Test Provider"
48 config = MagicMock()
49 config.name = "Test Provider"
50 config.instance_id = "test_provider--1"
51 config.get_value.return_value = "GLOBAL"
52 return provider_cls(mass, manifest, config)
53
54
55def test_streaming_provider_has_conservative_default() -> None:
56 """Unknown streaming providers default to five concurrent sources."""
57 provider = _make_provider(_StreamingProvider)
58
59 assert provider.max_concurrent_streams == DEFAULT_MAX_CONCURRENT_STREAMS == 5
60
61
62def test_non_streaming_provider_is_unlimited() -> None:
63 """Non-streaming providers do not allocate a source limiter."""
64 provider = _make_provider(_NonStreamingProvider)
65
66 assert provider.max_concurrent_streams is None
67 assert provider.has_available_stream_slot
68
69
70async def test_stream_slot_is_scoped_per_provider_instance() -> None:
71 """One busy provider instance does not consume another instance's slot."""
72 first = _make_provider(_SingleStreamProvider)
73 second = _make_provider(_SingleStreamProvider)
74 cast("MagicMock", second.config).instance_id = "test_provider--2"
75
76 async with first.acquire_stream_slot(0.1):
77 assert not first.has_available_stream_slot
78 assert second.has_available_stream_slot
79 with pytest.raises(ProviderStreamLimitError) as exc_info:
80 async with first.acquire_stream_slot(0):
81 pytest.fail("A second source slot was unexpectedly acquired")
82 # the error must carry the localizable reason (provider name + limit) for API clients
83 assert exc_info.value.translation_key == "provider_stream_limit"
84 assert exc_info.value.translation_args == ["Test Provider", 1]
85
86 assert first.has_available_stream_slot
87
88
89async def test_stream_slot_is_granted_once_released() -> None:
90 """A waiter within its timeout is served as soon as the active source releases."""
91 provider = _make_provider(_SingleStreamProvider)
92 acquired = asyncio.Event()
93
94 async def _wait_for_slot() -> None:
95 async with provider.acquire_stream_slot(5):
96 acquired.set()
97
98 async with provider.acquire_stream_slot(5):
99 waiter = asyncio.create_task(_wait_for_slot())
100 await asyncio.sleep(0)
101 assert not acquired.is_set()
102
103 await waiter
104
105 assert acquired.is_set()
106 assert provider.has_available_stream_slot
107
108
109async def test_cancelled_stream_slot_wait_does_not_consume_capacity() -> None:
110 """Cancelling a waiter leaves the active source as the only leased slot."""
111 provider = _make_provider(_SingleStreamProvider)
112
113 async with provider.acquire_stream_slot(0.1):
114 waiter = asyncio.create_task(provider.acquire_stream_slot(None).__aenter__())
115 await asyncio.sleep(0)
116 waiter.cancel()
117 with pytest.raises(asyncio.CancelledError):
118 await waiter
119 assert not provider.has_available_stream_slot
120
121 assert provider.has_available_stream_slot
122