/
/
/
1"""Tests for the MusicProvider base model."""
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 MediaType, ProviderFeature, 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](
42 provider_cls: type[ProviderT], supported_features: set[ProviderFeature] | None = None
43) -> ProviderT:
44 """Construct a minimal MusicProvider instance."""
45 mass = MagicMock()
46 manifest = MagicMock()
47 manifest.type = ProviderType.MUSIC
48 manifest.domain = "test_provider"
49 manifest.name = "Test Provider"
50 config = MagicMock()
51 config.name = "Test Provider"
52 config.instance_id = "test_provider--1"
53 config.get_value.return_value = "GLOBAL"
54 return provider_cls(mass, manifest, config, supported_features)
55
56
57def test_supported_media_types_default_follows_library_features() -> None:
58 """Without an override, media type support is derived from the library features."""
59 provider = _make_provider(
60 _StreamingProvider,
61 supported_features={
62 ProviderFeature.SEARCH,
63 ProviderFeature.LIBRARY_TRACKS,
64 ProviderFeature.LIBRARY_PLAYLISTS,
65 },
66 )
67
68 assert provider.supported_media_types == {MediaType.TRACK, MediaType.PLAYLIST}
69
70
71def test_streaming_provider_has_conservative_default() -> None:
72 """Unknown streaming providers default to five concurrent sources."""
73 provider = _make_provider(_StreamingProvider)
74
75 assert provider.max_concurrent_streams == DEFAULT_MAX_CONCURRENT_STREAMS == 5
76
77
78def test_non_streaming_provider_is_unlimited() -> None:
79 """Non-streaming providers do not allocate a source limiter."""
80 provider = _make_provider(_NonStreamingProvider)
81
82 assert provider.max_concurrent_streams is None
83 assert provider.has_available_stream_slot
84
85
86async def test_stream_slot_is_scoped_per_provider_instance() -> None:
87 """One busy provider instance does not consume another instance's slot."""
88 first = _make_provider(_SingleStreamProvider)
89 second = _make_provider(_SingleStreamProvider)
90 cast("MagicMock", second.config).instance_id = "test_provider--2"
91
92 async with first.acquire_stream_slot(0.1):
93 assert not first.has_available_stream_slot
94 assert second.has_available_stream_slot
95 with pytest.raises(ProviderStreamLimitError) as exc_info:
96 async with first.acquire_stream_slot(0):
97 pytest.fail("A second source slot was unexpectedly acquired")
98 # the error must carry the localizable reason (provider name + limit) for API clients
99 assert exc_info.value.translation_key == "provider_stream_limit"
100 assert exc_info.value.translation_args == ["Test Provider", 1]
101
102 assert first.has_available_stream_slot
103
104
105async def test_stream_slot_is_granted_once_released() -> None:
106 """A waiter within its timeout is served as soon as the active source releases."""
107 provider = _make_provider(_SingleStreamProvider)
108 acquired = asyncio.Event()
109
110 async def _wait_for_slot() -> None:
111 async with provider.acquire_stream_slot(5):
112 acquired.set()
113
114 async with provider.acquire_stream_slot(5):
115 waiter = asyncio.create_task(_wait_for_slot())
116 await asyncio.sleep(0)
117 assert not acquired.is_set()
118
119 await waiter
120
121 assert acquired.is_set()
122 assert provider.has_available_stream_slot
123
124
125async def test_cancelled_stream_slot_wait_does_not_consume_capacity() -> None:
126 """Cancelling a waiter leaves the active source as the only leased slot."""
127 provider = _make_provider(_SingleStreamProvider)
128
129 async with provider.acquire_stream_slot(0.1):
130 waiter = asyncio.create_task(provider.acquire_stream_slot(None).__aenter__())
131 await asyncio.sleep(0)
132 waiter.cancel()
133 with pytest.raises(asyncio.CancelledError):
134 await waiter
135 assert not provider.has_available_stream_slot
136
137 assert provider.has_available_stream_slot
138