music-assistant-server

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.get_stream_details = AsyncMock(return_value=_streamdetails(instance))
90    return provider
91
92
93def _mass(providers: dict[str, MagicMock]) -> MagicMock:
94    """Build a mass double that resolves the given provider instances."""
95    mass = MagicMock()
96    mass.providers = list(providers.values())
97    mass.get_provider.side_effect = lambda instance, **_kwargs: providers.get(instance)
98    mass.player_queues.queue_data_or_none.return_value = None
99    mass.streams.get_config_value.return_value = -17
100    mass.music.providers = list(providers.values())
101    mass.music.library_supported.return_value = True
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_matchless_search_falls_back_to_the_blocking_wait(
152    monkeypatch: pytest.MonkeyPatch,
153) -> None:
154    """Without a match, the budget is still spent waiting on the known source."""
155    queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
156    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
157    providers = {
158        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
159        MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
160    }
161    # match_provider finds nothing (the _mass default)
162    mass = _mass(providers)
163    audio = StreamsAudio(mass)
164    expected_buffer = MagicMock(spec=AudioBuffer)
165    get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
166    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
167
168    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
169
170    assert result is expected_buffer
171    mass.music.tracks.match_provider.assert_awaited_once()
172    waits = [call.kwargs["source_wait_timeout"] for call in get_buffer.await_args_list]
173    assert waits[0] == 0
174    assert waits[1] > 0
175    assert get_buffer.await_args_list[1].kwargs["streamdetails"].provider == BUSY_INSTANCE
176
177
178async def test_a_matchless_search_still_raises_when_the_slot_never_frees(
179    monkeypatch: pytest.MonkeyPatch,
180) -> None:
181    """Without a match and without a freed slot, the typed capacity error surfaces."""
182    queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
183    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
184    providers = {
185        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
186        MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
187    }
188    mass = _mass(providers)
189    audio = StreamsAudio(mass)
190    get_buffer = AsyncMock(side_effect=_limit_error(BUSY_INSTANCE))
191    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
192
193    with pytest.raises(ProviderStreamLimitError):
194        await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=0.2)
195
196    mass.music.tracks.match_provider.assert_awaited_once()
197    assert get_buffer.await_count == 2
198
199
200async def test_disallowed_provider_match_is_never_attempted(
201    monkeypatch: pytest.MonkeyPatch,
202) -> None:
203    """A caller that opts out of matching keeps the plain blocking behavior."""
204    queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
205    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
206    providers = {
207        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
208        MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
209    }
210    mass = _mass(providers)
211    mass.music.tracks.match_provider = AsyncMock(
212        return_value=[_mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)]
213    )
214    audio = StreamsAudio(mass)
215    get_buffer = AsyncMock(side_effect=_limit_error(BUSY_INSTANCE))
216    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
217
218    with pytest.raises(ProviderStreamLimitError):
219        await audio.get_audio_buffer(
220            queue_item,
221            reason="prepare_next",
222            capacity_wait_timeout=0.2,
223            allow_provider_match=False,
224        )
225
226    mass.music.tracks.match_provider.assert_not_awaited()
227    # without a pending match, the sole candidate gets the blocking wait right away
228    assert get_buffer.await_args_list[0].kwargs["source_wait_timeout"] > 0
229
230
231async def test_a_library_track_persists_the_discovered_mapping(
232    monkeypatch: pytest.MonkeyPatch,
233) -> None:
234    """A match for a library track is written back to the library."""
235    queue_item = _queue_item(
236        _mapping(BUSY_INSTANCE, quality=ContentType.FLAC), provider="library", item_id="42"
237    )
238    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
239    providers = {
240        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
241        MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
242    }
243    mass = _mass(providers)
244    matched_mapping = _mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)
245    mass.music.tracks.match_provider = AsyncMock(return_value=[matched_mapping])
246    audio = StreamsAudio(mass)
247    expected_buffer = MagicMock(spec=AudioBuffer)
248    get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
249    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
250
251    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
252
253    assert result is expected_buffer
254    # the write-back runs as a background task, detached from this playback request
255    await asyncio.sleep(0)
256    mass.music.tracks.add_provider_mappings.assert_awaited_once_with("42", [matched_mapping])
257
258
259async def test_one_failing_provider_does_not_end_the_search(
260    monkeypatch: pytest.MonkeyPatch,
261) -> None:
262    """A provider that errors while searching is skipped in favor of the next one."""
263    queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
264    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
265    providers = {
266        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
267        FAILING_INSTANCE: _music_provider(FAILING_INSTANCE),
268        MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
269    }
270    mass = _mass(providers)
271    matched_mapping = _mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)
272    # a raw (non MusicAssistantError) provider bug must be contained as well
273    mass.music.tracks.match_provider = AsyncMock(
274        side_effect=[KeyError("unexpected api response"), [matched_mapping]]
275    )
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    assert mass.music.tracks.match_provider.await_count == 2
285    assert queue_item.streamdetails is not None
286    assert queue_item.streamdetails.provider == MATCH_INSTANCE
287
288
289async def test_a_failed_mapping_write_back_does_not_fail_the_rescue(
290    monkeypatch: pytest.MonkeyPatch,
291) -> None:
292    """A library write-back error still leaves the discovered mapping serving playback."""
293    queue_item = _queue_item(
294        _mapping(BUSY_INSTANCE, quality=ContentType.FLAC), provider="library", item_id="42"
295    )
296    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
297    providers = {
298        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
299        MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
300    }
301    mass = _mass(providers)
302    mass.music.tracks.match_provider = AsyncMock(
303        return_value=[_mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)]
304    )
305    mass.music.tracks.add_provider_mappings = AsyncMock(
306        side_effect=MediaNotFoundError("Track not found: 42")
307    )
308    audio = StreamsAudio(mass)
309    expected_buffer = MagicMock(spec=AudioBuffer)
310    get_buffer = AsyncMock(side_effect=[_limit_error(BUSY_INSTANCE), expected_buffer])
311    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
312
313    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
314
315    assert result is expected_buffer
316    assert queue_item.streamdetails is not None
317    assert queue_item.streamdetails.provider == MATCH_INSTANCE
318    await asyncio.sleep(0)
319    mass.music.tracks.add_provider_mappings.assert_awaited_once()
320
321
322async def test_discovery_runs_at_most_once_per_buffer_request(
323    monkeypatch: pytest.MonkeyPatch,
324) -> None:
325    """A discovered provider that is itself saturated does not trigger a second search."""
326    queue_item = _queue_item(_mapping(BUSY_INSTANCE, quality=ContentType.FLAC))
327    queue_item.streamdetails = _streamdetails(BUSY_INSTANCE)
328    providers = {
329        BUSY_INSTANCE: _music_provider(BUSY_INSTANCE, has_slot=False),
330        MATCH_INSTANCE: _music_provider(MATCH_INSTANCE),
331    }
332    mass = _mass(providers)
333    mass.music.tracks.match_provider = AsyncMock(
334        return_value=[_mapping(MATCH_INSTANCE, item_id=MATCHED_ITEM_ID)]
335    )
336    audio = StreamsAudio(mass)
337    expected_buffer = MagicMock(spec=AudioBuffer)
338    get_buffer = AsyncMock(
339        side_effect=[
340            _limit_error(BUSY_INSTANCE),
341            _limit_error(MATCH_INSTANCE),
342            expected_buffer,
343        ]
344    )
345    monkeypatch.setattr(AudioBuffer, "get_buffer", get_buffer)
346
347    result = await audio.get_audio_buffer(queue_item, reason="streaming", capacity_wait_timeout=1)
348
349    assert result is expected_buffer
350    mass.music.tracks.match_provider.assert_awaited_once()
351    assert get_buffer.await_count == 3
352    # after the one search, saturation everywhere ends in the final blocking pass as before
353    waits = [call.kwargs["source_wait_timeout"] for call in get_buffer.await_args_list]
354    assert waits[0] == 0
355    assert waits[1] == 0
356    assert waits[2] > 0
357    assert get_buffer.await_args_list[2].kwargs["streamdetails"].provider == BUSY_INSTANCE
358