/
/
/
1"""Tests for the cross-provider track match when every known source is capacity-saturated."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, MagicMock
8
9import pytest
10from music_assistant_models.enums import ContentType, MediaType, ProviderFeature, StreamType
11from music_assistant_models.errors import MediaNotFoundError
12from music_assistant_models.media_items import AudioFormat, ProviderMapping, Track
13from music_assistant_models.queue_item import QueueItem
14from music_assistant_models.streamdetails import StreamDetails
15
16from music_assistant.controllers.streams.audio import StreamsAudio
17from music_assistant.controllers.streams.audio_buffer import AudioBuffer
18from music_assistant.models.music_provider import MusicProvider, ProviderStreamLimitError
19
20BUSY_INSTANCE = "spotify--busy"
21MATCH_INSTANCE = "tidal--match"
22FAILING_INSTANCE = "qobuz--failing"
23ITEM_ID = "item-1"
24MATCHED_ITEM_ID = "item-on-tidal"
25
26
27def _mapping(
28 instance: str, item_id: str = ITEM_ID, quality: ContentType = ContentType.MP3
29) -> ProviderMapping:
30 """Build a streamable provider mapping."""
31 return ProviderMapping(
32 item_id=item_id,
33 provider_domain=instance.split("--", maxsplit=1)[0],
34 provider_instance=instance,
35 audio_format=AudioFormat(content_type=quality),
36 )
37
38
39def _streamdetails(instance: str) -> StreamDetails:
40 """Build HTTP stream details for one provider instance."""
41 return StreamDetails(
42 provider=instance,
43 item_id=ITEM_ID,
44 audio_format=AudioFormat(content_type=ContentType.MP3),
45 media_type=MediaType.TRACK,
46 stream_type=StreamType.HTTP,
47 path="http://test.invalid/item.mp3",
48 duration=30,
49 )
50
51
52def _queue_item(
53 *mappings: ProviderMapping, provider: str | None = None, item_id: str = ITEM_ID
54) -> QueueItem:
55 """Build a queue item whose track carries the given provider mappings."""
56 media_item = Track(
57 item_id=item_id,
58 provider=provider or mappings[0].provider_instance,
59 name="Song",
60 provider_mappings=set(mappings),
61 )
62 return QueueItem(
63 queue_id="queue-1",
64 queue_item_id="queue-item-1",
65 name="Song",
66 duration=30,
67 media_item=media_item,
68 )
69
70
71def _limit_error(instance: str) -> ProviderStreamLimitError:
72 """Build a typed source-capacity error for a provider instance."""
73 provider = MagicMock(spec=MusicProvider)
74 provider.max_concurrent_streams = 1
75 provider.name = "Limited"
76 provider.instance_id = instance
77 return ProviderStreamLimitError(provider, 0)
78
79
80def _music_provider(instance: str, has_slot: bool = True) -> MagicMock:
81 """Build a loaded, match-eligible streaming provider instance."""
82 provider = MagicMock(spec=MusicProvider)
83 provider.instance_id = instance
84 provider.domain = instance.split("--", maxsplit=1)[0]
85 provider.available = True
86 provider.is_streaming_provider = True
87 provider.has_available_stream_slot = has_slot
88 provider.supported_features = {ProviderFeature.SEARCH}
89 provider.supported_media_types = {MediaType.TRACK}
90 provider.get_stream_details = AsyncMock(return_value=_streamdetails(instance))
91 return provider
92
93
94def _mass(providers: dict[str, MagicMock]) -> MagicMock:
95 """Build a mass double that resolves the given provider instances."""
96 mass = MagicMock()
97 mass.providers = list(providers.values())
98 mass.get_provider.side_effect = lambda instance, **_kwargs: providers.get(instance)
99 mass.player_queues.queue_data_or_none.return_value = None
100 mass.streams.get_config_value.return_value = -17
101 mass.music.providers = list(providers.values())
102 mass.music.tracks.match_provider = AsyncMock(return_value=[])
103 mass.music.tracks.add_provider_mappings = AsyncMock()
104
105 def _schedule(coro: Any, *_args: Any, **_kwargs: Any) -> asyncio.Task[Any]:
106 task = asyncio.get_running_loop().create_task(coro)
107 task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
108 return task
109
110 mass.create_task.side_effect = _schedule
111 return mass
112
113
114async def test_saturated_single_mapping_is_rescued_by_a_cross_provider_match(
115 monkeypatch: pytest.MonkeyPatch,
116) -> None:
117 """A track with one saturated source plays through a match found on another provider."""
118 queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
119 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
120 providers = {
121 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
122 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
123 }
124 mass = _mass(providers)
125 matched_mapping = _mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)
126 mass.music.tracks.match_provider = AsyncMock(return_value=[matched_mapping])
127 audio = StreamsAudio(mass)
128 expected_buffer = MagicMock(spec=AudioBuffer)
129 get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
130 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
131
132 result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
133
134 assert result is expected_buffer
135 assert queue_item.streamdetails is not None
136 assert queue_item.streamdetails.provider == MATCH_INSTANCE
137 # the discovered mapping is kept on the media item, so the reselection loop can use it
138 assert isinstance(queue_item.media_item, Track)
139 assert matched_mapping in queue_item.media_item.provider_mappings
140 # the saturated source is only probed, never granted a blocking wait
141 waits = [call.kwargs["source_wait_timeout"] for call in get_buffer.await_args_list]
142 assert waits == [0, 0]
143 mass.music.tracks.match_provider.assert_awaited_once()
144 assert mass.music.tracks.match_provider.await_args.args[1] is providers[MATCH_INSTANCE]
145 # strictness is what prevents ever playing the wrong track
146 assert mass.music.tracks.match_provider.await_args.kwargs["strict"] is True
147 # a non-library track keeps the mapping in memory only
148 mass.music.tracks.add_provider_mappings.assert_not_awaited()
149
150
151async def test_a_provider_without_track_support_is_never_searched(
152 monkeypatch: pytest.MonkeyPatch,
153) -> None:
154 """Matching only consults providers that declare support for tracks."""
155 queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
156 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
157 radio_only = _music_provider("radioprov--1")
158 radio_only.supported_media_types = {MediaType.RADIO}
159 providers = {
160 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
161 "radioprov--1": radio_only,
162 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
163 }
164 mass = _mass(providers)
165 mass.music.tracks.match_provider = AsyncMock(
166 return_value=[_mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)]
167 )
168 audio = StreamsAudio(mass)
169 expected_buffer = MagicMock(spec=AudioBuffer)
170 get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
171 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
172
173 result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
174
175 assert result is expected_buffer
176 # the radio-only provider is skipped; the track-capable provider gets the search
177 mass.music.tracks.match_provider.assert_awaited_once()
178 assert mass.music.tracks.match_provider.await_args.args[1] is providers[MATCH_INSTANCE]
179
180
181async def test_a_matchless_search_falls_back_to_the_blocking_wait(
182 monkeypatch: pytest.MonkeyPatch,
183) -> None:
184 """Without a match, the budget is still spent waiting on the known source."""
185 queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
186 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
187 providers = {
188 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
189 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
190 }
191 # match_provider finds nothing (the _mass default)
192 mass = _mass(providers)
193 audio = StreamsAudio(mass)
194 expected_buffer = MagicMock(spec=AudioBuffer)
195 get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
196 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
197
198 result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
199
200 assert result is expected_buffer
201 mass.music.tracks.match_provider.assert_awaited_once()
202 waits = [call.kwargs["source_wait_timeout"] for call in get_buffer.await_args_list]
203 assert waits[0] == 0
204 assert waits[1] > 0
205 assert get_buffer.await_args_list[1].kwargs["streamdetails"].provider == BUSY_INSTANCE
206
207
208async def test_a_matchless_search_still_raises_when_the_slot_never_frees(
209 monkeypatch: pytest.MonkeyPatch,
210) -> None:
211 """Without a match and without a freed slot, the typed capacity error surfaces."""
212 queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
213 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
214 providers = {
215 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
216 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
217 }
218 mass = _mass(providers)
219 audio = StreamsAudio(mass)
220 get_buffer = AsyncMock(side_effect=_limit_error(BUSY_INSTANCE))
221 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
222
223 with pytest.raises(ProviderStreamLimitError):
224 await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=0.2)
225
226 mass.music.tracks.match_provider.assert_awaited_once()
227 assert get_buffer.await_count == 2
228
229
230async def test_disallowed_provider_match_is_never_attempted(
231 monkeypatch: pytest.MonkeyPatch,
232) -> None:
233 """A caller that opts out of matching keeps the plain blocking behavior."""
234 queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
235 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
236 providers = {
237 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
238 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
239 }
240 mass = _mass(providers)
241 mass.music.tracks.match_provider = AsyncMock(
242 return_value=[_mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)]
243 )
244 audio = StreamsAudio(mass)
245 get_buffer = AsyncMock(side_effect=_limit_error(BUSY_INSTANCE))
246 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
247
248 with pytest.raises(ProviderStreamLimitError):
249 await audio.get_audio_buffer(
250 queue_item,
251 reason="prepare_next",
252 capacity_wait_timeout=0.2,
253 allow_provider_match=False,
254 )
255
256 mass.music.tracks.match_provider.assert_not_awaited()
257 # without a pending match, the sole candidate gets the blocking wait right away
258 assert get_buffer.await_args_list[0].kwargs["source_wait_timeout"] > 0
259
260
261async def test_a_library_track_persists_the_discovered_mapping(
262 monkeypatch: pytest.MonkeyPatch,
263) -> None:
264 """A match for a library track is written back to the library."""
265 queue_item = _queue_item(
266 _mapping(BUSY_INSTANCE, quality=ContentType.FLAC), provider="library", item_id="42"
267 )
268 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
269 providers = {
270 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
271 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
272 }
273 mass = _mass(providers)
274 matched_mapping = _mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)
275 mass.music.tracks.match_provider = AsyncMock(return_value=[matched_mapping])
276 audio = StreamsAudio(mass)
277 expected_buffer = MagicMock(spec=AudioBuffer)
278 get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
279 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
280
281 result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
282
283 assert result is expected_buffer
284 # the write-back runs as a background task, detached from this playback request
285 await asyncio.sleep(0)
286 mass.music.tracks.add_provider_mappings.assert_awaited_once_with("42", [matched_mapping])
287
288
289async def test_one_failing_provider_does_not_end_the_search(
290 monkeypatch: pytest.MonkeyPatch,
291) -> None:
292 """A provider that errors while searching is skipped in favor of the next one."""
293 queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
294 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
295 providers = {
296 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
297 FAILING_INSTANCE: _music_provider(FAILING_INSTANCE),
298 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
299 }
300 mass = _mass(providers)
301 matched_mapping = _mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)
302 # a raw (non MusicAssistantError) provider bug must be contained as well
303 mass.music.tracks.match_provider = AsyncMock(
304 side_effect=[KeyError("unexpected api response"), [matched_mapping]]
305 )
306 audio = StreamsAudio(mass)
307 expected_buffer = MagicMock(spec=AudioBuffer)
308 get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
309 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
310
311 result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
312
313 assert result is expected_buffer
314 assert mass.music.tracks.match_provider.await_count == 2
315 assert queue_item.streamdetails is not None
316 assert queue_item.streamdetails.provider == MATCH_INSTANCE
317
318
319async def test_a_failed_mapping_write_back_does_not_fail_the_rescue(
320 monkeypatch: pytest.MonkeyPatch,
321) -> None:
322 """A library write-back error still leaves the discovered mapping serving playback."""
323 queue_item = _queue_item(
324 _mapping(BUSY_INSTANCE, quality=ContentType.FLAC), provider="library", item_id="42"
325 )
326 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
327 providers = {
328 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
329 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
330 }
331 mass = _mass(providers)
332 mass.music.tracks.match_provider = AsyncMock(
333 return_value=[_mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)]
334 )
335 mass.music.tracks.add_provider_mappings = AsyncMock(
336 side_effect=MediaNotFoundError("Track not found: 42")
337 )
338 audio = StreamsAudio(mass)
339 expected_buffer = MagicMock(spec=AudioBuffer)
340 get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
341 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
342
343 result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
344
345 assert result is expected_buffer
346 assert queue_item.streamdetails is not None
347 assert queue_item.streamdetails.provider == MATCH_INSTANCE
348 await asyncio.sleep(0)
349 mass.music.tracks.add_provider_mappings.assert_awaited_once()
350
351
352async def test_discovery_runs_at_most_once_per_buffer_request(
353 monkeypatch: pytest.MonkeyPatch,
354) -> None:
355 """A discovered provider that is itself saturated does not trigger a second search."""
356 queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
357 queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
358 providers = {
359 BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
360 MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
361 }
362 mass = _mass(providers)
363 mass.music.tracks.match_provider = AsyncMock(
364 return_value=[_mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)]
365 )
366 audio = StreamsAudio(mass)
367 expected_buffer = MagicMock(spec=AudioBuffer)
368 get_buffer = AsyncMock(
369 side_effect=[
370 _limit_error(BUSY_INSTANCE),
371 _limit_error(MATCH_INSTANCE),
372 expected_buffer,
373 ]
374 )
375 monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
376
377 result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
378
379 assert result is expected_buffer
380 mass.music.tracks.match_provider.assert_awaited_once()
381 assert get_buffer.await_count == 3
382 # after the one search, saturation everywhere ends in the final blocking pass as before
383 waits = [call.kwargs["source_wait_timeout"] for call in get_buffer.await_args_list]
384 assert waits[0] == 0
385 assert waits[1] == 0
386 assert waits[2] > 0
387 assert get_buffer.await_args_list[2].kwargs["streamdetails"].provider == BUSY_INSTANCE
388