music-assistant-server

1"""Tests for the capacity-aware source selection behind ``get_audio_buffer``."""
2
3from __future__ import annotations
4
5import asyncio
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.enums import ContentType, MediaType, StreamType
10from music_assistant_models.errors import (
11    AudioError,
12    MediaNotFoundError,
13    ProviderUnavailableError,
14)
15from music_assistant_models.media_items import AudioFormat, ProviderMapping, SoundEffect
16from music_assistant_models.queue_item import QueueItem
17from music_assistant_models.streamdetails import StreamDetails
18
19from music_assistant.controllers.streams.audio import StreamsAudio
20from music_assistant.controllers.streams.audio_buffer import AudioBuffer
21from music_assistant.models.music_provider import MusicProvider, ProviderStreamLimitError
22
23BUSY_INSTANCE = "service--busy"
24FALLBACK_INSTANCE = "service--fallback"
25ITEM_ID = "item-1"
26
27
28def _mapping(instance: str, quality: ContentType = ContentType.MP3) -> ProviderMapping:
29    """Build a streamable provider mapping."""
30    return ProviderMapping(
31        item_id=ITEM_ID,
32        provider_domain=instance.split("--", maxsplit=1)[0],
33        provider_instance=instance,
34        audio_format=AudioFormat(content_type=quality),
35    )
36
37
38def _streamdetails(instance: str) -> StreamDetails:
39    """Build HTTP stream details for one provider instance."""
40    return StreamDetails(
41        provider=instance,
42        item_id=ITEM_ID,
43        audio_format=AudioFormat(content_type=ContentType.MP3),
44        media_type=MediaType.SOUND_EFFECT,
45        stream_type=StreamType.HTTP,
46        path="http://test.invalid/item.mp3",
47        duration=30,
48    )
49
50
51def _queue_item(*mappings: ProviderMapping) -> QueueItem:
52    """Build a queue item with the given provider mappings."""
53    media_item = SoundEffect(
54        item_id=ITEM_ID,
55        provider=mappings[0].provider_instance,
56        name="Effect",
57        provider_mappings=set(mappings),
58    )
59    return QueueItem(
60        queue_id="queue-1",
61        queue_item_id="queue-item-1",
62        name="Effect",
63        duration=30,
64        media_item=media_item,
65    )
66
67
68def _limit_error(instance: str) -> ProviderStreamLimitError:
69    """Build a typed source-capacity error for a provider instance."""
70    provider = MagicMock(spec=MusicProvider)
71    provider.max_concurrent_streams = 1
72    provider.name = "Limited"
73    provider.instance_id = instance
74    return ProviderStreamLimitError(provider, 0)
75
76
77def _music_provider(instance: str, has_slot: bool) -> MagicMock:
78    """Build a loaded streaming provider instance that resolves its own stream details."""
79    provider = MagicMock(spec=MusicProvider)
80    provider.instance_id = instance
81    provider.domain = instance.split("--", maxsplit=1)[0]
82    provider.available = True
83    provider.is_streaming_provider = True
84    provider.has_available_stream_slot = has_slot
85    provider.get_stream_details = AsyncMock(return_value=_streamdetails(instance))
86    return provider
87
88
89def _mass(providers: dict[str, MagicMock] | None = None) -> MagicMock:
90    """Build a mass double that resolves the given provider instances."""
91    mass = MagicMock()
92    if providers is None:
93        mass.get_provider.return_value = MagicMock()
94    else:
95        mass.providers = list(providers.values())
96        mass.get_provider.side_effect = lambda instance, **_kwargs: providers.get(instance)
97    mass.player_queues.queue_data_or_none.return_value = None
98    mass.streams.get_config_value.return_value = -17
99    return mass
100
101
102async def test_reselects_another_mapping_after_a_capacity_failure(
103    monkeypatch: pytest.MonkeyPatch,
104) -> None:
105    """A source that has no free slot is replaced with another compatible mapping."""
106    queue_item = _queue_item(
107        _mapping(BUSY_INSTANCE, ContentType.FLAC),
108        _mapping(FALLBACK_INSTANCE),
109    )
110    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
111    audio = StreamsAudio(_mass())
112    fallback_details = _streamdetails(FALLBACK_INSTANCE)
113    audio.get_stream_details = AsyncMock(return_value=fallback_details)  # type: ignore[method-assign]
114    expected_buffer = MagicMock(spec=AudioBuffer)
115    get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
116    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
117
118    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
119
120    assert result is expected_buffer
121    assert queue_item.streamdetails is fallback_details
122    assert audio.get_stream_details.await_args is not None
123    assert audio.get_stream_details.await_args.kwargs["excluded_provider_instances"] == {
124        BUSY_INSTANCE
125    }
126
127
128async def test_falls_back_to_a_compatible_instance_of_the_same_mapping(
129    monkeypatch: pytest.MonkeyPatch,
130) -> None:
131    """One mapping is retried on another loaded instance of its streaming catalog."""
132    queue_item = _queue_item(_mapping(BUSY_INSTANCE, ContentType.FLAC))
133    primary = _music_provider(BUSY_INSTANCE, has_slot=False)
134    fallback = _music_provider(FALLBACK_INSTANCE, has_slot=True)
135    audio = StreamsAudio(_mass({BUSY_INSTANCE: primary, FALLBACK_INSTANCE: fallback}))
136    expected_buffer = MagicMock(spec=AudioBuffer)
137    get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
138    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
139
140    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
141
142    assert result is expected_buffer
143    assert queue_item.streamdetails is not None
144    assert queue_item.streamdetails.provider == FALLBACK_INSTANCE
145    # every candidate is probed (0s) while a reselection can still follow; a slot
146    # snapshot is never trusted, so a free fallback still acquires instantly
147    assert get_buffer.await_args_list[0].kwargs["source_wait_timeout"] == 0
148    assert get_buffer.await_args_list[1].kwargs["source_wait_timeout"] == 0
149    fallback.get_stream_details.assert_awaited_once_with(ITEM_ID, MediaType.SOUND_EFFECT)
150
151
152async def test_all_candidates_busy_ends_in_one_blocking_pass_on_the_best_one(
153    monkeypatch: pytest.MonkeyPatch,
154) -> None:
155    """With every candidate saturated, the budget is spent waiting on the preferred mapping."""
156    queue_item = _queue_item(
157        _mapping(BUSY_INSTANCE, ContentType.FLAC),
158        _mapping(FALLBACK_INSTANCE),
159    )
160    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
161    providers = {
162        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
163        FALLBACK_INSTANCE: _music_provider(FALLBACK_INSTANCE, has_slot=False),
164    }
165    audio = StreamsAudio(_mass(providers))
166    expected_buffer = MagicMock(spec=AudioBuffer)
167    get_buffer = AsyncMock(
168        side_effect=[
169            _limit_error(BUSY_INSTANCE),
170            _limit_error(FALLBACK_INSTANCE),
171            expected_buffer,
172        ]
173    )
174    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
175
176    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
177
178    assert result is expected_buffer
179    assert get_buffer.await_count == 3
180    # every saturated candidate is only probed, so the budget survives for the final pass
181    waits = [call.kwargs["source_wait_timeout"] for call in get_buffer.await_args_list]
182    assert waits[0] == 0
183    assert waits[1] == 0
184    assert waits[2] > 0
185    # the single blocking wait is spent on the highest quality mapping, not the last tried
186    probed = [call.kwargs["streamdetails"].provider for call in get_buffer.await_args_list]
187    assert probed == [BUSY_INSTANCE, FALLBACK_INSTANCE, BUSY_INSTANCE]
188
189
190async def test_exhausted_budget_raises_typed_error_and_keeps_the_item_playable(
191    monkeypatch: pytest.MonkeyPatch,
192) -> None:
193    """A spent capacity budget surfaces the typed error without revoking availability."""
194    queue_item = _queue_item(_mapping(BUSY_INSTANCE, ContentType.FLAC))
195    original_details = _streamdetails(BUSY_INSTANCE)
196    queue_item.streamdetails = original_details
197    audio = StreamsAudio(_mass())
198    get_buffer = AsyncMock(side_effect=_limit_error(BUSY_INSTANCE))
199    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
200
201    with pytest.raises(ProviderStreamLimitError):
202        await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=0)
203
204    assert queue_item.available
205    assert queue_item.streamdetails is original_details
206    assert get_buffer.await_count == 1
207
208
209async def test_capacity_reselection_is_shared_by_concurrent_waiters(
210    monkeypatch: pytest.MonkeyPatch,
211) -> None:
212    """Speculative and playback waiters on one item share the single replacement source."""
213    queue_item = _queue_item(
214        _mapping(BUSY_INSTANCE, ContentType.FLAC),
215        _mapping(FALLBACK_INSTANCE),
216    )
217    busy_details = _streamdetails(BUSY_INSTANCE)
218    queue_item.streamdetails = busy_details
219    audio = StreamsAudio(_mass())
220    fallback_details = _streamdetails(FALLBACK_INSTANCE)
221    audio.get_stream_details = AsyncMock(return_value=fallback_details)  # type: ignore[method-assign]
222    fallback_buffer = MagicMock(spec=AudioBuffer)
223
224    async def _get_buffer(**kwargs: object) -> MagicMock:
225        # yield control so both waiters would race without the per-item lock
226        await asyncio.sleep(0)
227        if kwargs["streamdetails"] is busy_details:
228            raise _limit_error(BUSY_INSTANCE)
229        return fallback_buffer
230
231    monkeypatch.setattr(AudioBuffer, "get_buffer", _get_buffer)
232
233    results = await asyncio.gather(
234        audio.get_audio_buffer(queue_item, reason="prepare_next", capacity_wait_timeout=1),
235        audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1),
236    )
237
238    assert results[0] is fallback_buffer
239    assert results[1] is fallback_buffer
240    assert queue_item.streamdetails is fallback_details
241    assert audio.get_stream_details.await_count == 1
242
243
244async def test_a_failed_reselection_spends_the_budget_on_the_blocked_provider(
245    monkeypatch: pytest.MonkeyPatch,
246) -> None:
247    """An unplayable alternative falls back to waiting out the budget on the busy source."""
248    queue_item = _queue_item(
249        _mapping(BUSY_INSTANCE, ContentType.FLAC),
250        _mapping(FALLBACK_INSTANCE),
251    )
252    busy_details = _streamdetails(BUSY_INSTANCE)
253    queue_item.streamdetails = busy_details
254    providers = {
255        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
256        FALLBACK_INSTANCE: _music_provider(FALLBACK_INSTANCE, has_slot=True),
257    }
258    audio = StreamsAudio(_mass(providers))
259    # the alternative mapping exists but can not be resolved (e.g. region locked)
260    audio.get_stream_details = AsyncMock(side_effect=MediaNotFoundError("not here"))  # type: ignore[method-assign]
261    expected_buffer = MagicMock(spec=AudioBuffer)
262    get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
263    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
264
265    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
266
267    # the capacity budget is spent on the blocked provider instead of being abandoned
268    assert result is expected_buffer
269    assert get_buffer.await_count == 2
270    assert get_buffer.await_args_list[0].kwargs["source_wait_timeout"] == 0
271    assert get_buffer.await_args_list[1].kwargs["source_wait_timeout"] > 0
272    assert get_buffer.await_args_list[1].kwargs["streamdetails"] is busy_details
273    assert queue_item.streamdetails is busy_details
274
275
276async def test_a_broken_alternate_falls_back_to_the_capacity_blocked_source(
277    monkeypatch: pytest.MonkeyPatch,
278) -> None:
279    """A failing alternate source must not turn a transient capacity miss into a hard failure."""
280    queue_item = _queue_item(
281        _mapping(BUSY_INSTANCE, ContentType.FLAC),
282        _mapping(FALLBACK_INSTANCE),
283    )
284    busy_details = _streamdetails(BUSY_INSTANCE)
285    queue_item.streamdetails = busy_details
286    providers = {
287        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
288        FALLBACK_INSTANCE: _music_provider(FALLBACK_INSTANCE, has_slot=True),
289    }
290    audio = StreamsAudio(_mass(providers))
291    audio.get_stream_details = AsyncMock(return_value=_streamdetails(FALLBACK_INSTANCE))  # type: ignore[method-assign]
292    expected_buffer = MagicMock(spec=AudioBuffer)
293    get_buffer = AsyncMock(
294        side_effect=[
295            _limit_error(BUSY_INSTANCE),
296            AudioError("alternate source is broken"),
297            expected_buffer,
298        ]
299    )
300    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
301
302    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
303
304    assert result is expected_buffer
305    assert get_buffer.await_count == 3
306    # the budget is returned to the blocked source instead of surfacing the alternate's error
307    assert get_buffer.await_args_list[2].kwargs["streamdetails"] is busy_details
308    assert get_buffer.await_args_list[2].kwargs["source_wait_timeout"] > 0
309    assert queue_item.streamdetails is busy_details
310
311
312async def test_the_final_pass_surfaces_the_blocked_sources_own_error(
313    monkeypatch: pytest.MonkeyPatch,
314) -> None:
315    """Once the budget is spent on the preferred source, its real failure is the answer."""
316    queue_item = _queue_item(
317        _mapping(BUSY_INSTANCE, ContentType.FLAC),
318        _mapping(FALLBACK_INSTANCE),
319    )
320    busy_details = _streamdetails(BUSY_INSTANCE)
321    queue_item.streamdetails = busy_details
322    providers = {
323        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
324        FALLBACK_INSTANCE: _music_provider(FALLBACK_INSTANCE, has_slot=True),
325    }
326    audio = StreamsAudio(_mass(providers))
327    audio.get_stream_details = AsyncMock(return_value=_streamdetails(FALLBACK_INSTANCE))  # type: ignore[method-assign]
328    monkeypatch.setattr(
329        AudioBuffer,
330        "get_buffer",
331        AsyncMock(
332            side_effect=[
333                _limit_error(BUSY_INSTANCE),
334                AudioError("alternate source is broken"),
335                AudioError("preferred source is broken"),
336            ]
337        ),
338    )
339
340    with pytest.raises(AudioError, match="preferred source is broken"):
341        await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
342
343    assert queue_item.streamdetails is busy_details
344
345
346@pytest.mark.parametrize(
347    "reselection_error",
348    [ProviderUnavailableError("gone"), asyncio.CancelledError()],
349    ids=["provider_unavailable", "cancelled"],
350)
351async def test_streamdetails_survive_an_unexpected_reselection_failure(
352    monkeypatch: pytest.MonkeyPatch,
353    reselection_error: BaseException,
354) -> None:
355    """No exit path may leave the queue item without stream details."""
356    queue_item = _queue_item(
357        _mapping(BUSY_INSTANCE, ContentType.FLAC),
358        _mapping(FALLBACK_INSTANCE),
359    )
360    busy_details = _streamdetails(BUSY_INSTANCE)
361    queue_item.streamdetails = busy_details
362    providers = {
363        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
364        FALLBACK_INSTANCE: _music_provider(FALLBACK_INSTANCE, has_slot=True),
365    }
366    audio = StreamsAudio(_mass(providers))
367    audio.get_stream_details = AsyncMock(side_effect=reselection_error)  # type: ignore[method-assign]
368    monkeypatch.setattr(
369        AudioBuffer, "get_buffer", AsyncMock(side_effect=_limit_error(BUSY_INSTANCE))
370    )
371
372    with pytest.raises(type(reselection_error)):
373        await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
374
375    # a None here crashes the flow stream's end-of-track bookkeeping
376    assert queue_item.streamdetails is busy_details
377
378
379async def test_flow_mode_skips_the_item_on_capacity_exhaustion() -> None:
380    """Flow mode drops an item it can not open a source for, leaving it playable."""
381    queue_item = _queue_item(_mapping(BUSY_INSTANCE, ContentType.FLAC))
382    streamdetails = _streamdetails(BUSY_INSTANCE)
383    streamdetails.loudness = -10.0  # skip the audio-analysis hydration call
384    queue_item.streamdetails = streamdetails
385    audio = StreamsAudio(_mass())
386    audio.get_audio_buffer = AsyncMock(  # type: ignore[method-assign]
387        side_effect=_limit_error(BUSY_INSTANCE)
388    )
389    pcm_format = AudioFormat(
390        content_type=ContentType.PCM_S16LE,
391        sample_rate=8000,
392        bit_depth=16,
393        channels=2,
394    )
395
396    chunks = [
397        chunk
398        async for chunk in audio.get_queue_item_stream(queue_item, pcm_format, raise_on_error=False)
399    ]
400
401    assert chunks == []
402    assert queue_item.available
403    assert streamdetails.stream_error is True
404