music-assistant-server

1"""
2Tests that resolving streamdetails asks each provider mapping at most once.
3
4``get_stream_details`` builds its candidates once: mappings in quality order, the instances
5that can serve each mapping within it, and the providers the user's filter steers to ahead of
6the rest. Every (instance, item id) pair appears at most once, so a mapping that failed is not
7asked again -- which for a just-in-time renderer like AI Radio would mean a second full
8text-to-speech render.
9
10The mappings below are given distinct qualities wherever order matters, so the order the
11candidates are reached in is fixed rather than left to the iteration order of a set.
12"""
13
14from __future__ import annotations
15
16from typing import cast
17from unittest.mock import AsyncMock, MagicMock
18
19import pytest
20from music_assistant_models.enums import ContentType, MediaType, StreamType
21from music_assistant_models.errors import MediaNotFoundError
22from music_assistant_models.media_items import AudioFormat, ProviderMapping, SoundEffect
23from music_assistant_models.queue_item import QueueItem
24from music_assistant_models.streamdetails import StreamDetails
25
26from music_assistant.controllers.streams.audio import StreamsAudio
27from music_assistant.models.music_provider import MusicProvider
28
29INSTANCE = "ai_radio--abc"
30OTHER_INSTANCE = "tidal--xyz"
31ITEM_ID = "session123_0"
32
33
34def _mapping(
35    instance: str, item_id: str = ITEM_ID, content_type: ContentType = ContentType.MP3
36) -> ProviderMapping:
37    """
38    Build a provider mapping.
39
40    :param instance: The provider instance the mapping points at.
41    :param item_id: The item id on that provider.
42    :param content_type: Drives the mapping's quality score, which decides the order the
43        mappings are tried in. Pass a lossless type to have a mapping tried first.
44    """
45    return ProviderMapping(
46        item_id=item_id,
47        provider_domain=instance.split("--", maxsplit=1)[0],
48        provider_instance=instance,
49        audio_format=AudioFormat(content_type=content_type),
50    )
51
52
53def _queue_item(*mappings: ProviderMapping) -> QueueItem:
54    """Build a queue item whose media item carries the given provider mappings."""
55    media_item = SoundEffect(
56        item_id=ITEM_ID,
57        provider=mappings[0].provider_instance,
58        name="Intro",
59        provider_mappings=set(mappings),
60    )
61    return QueueItem(
62        queue_id="q1",
63        queue_item_id="qi1",
64        name="Intro",
65        duration=None,
66        media_item=media_item,
67    )
68
69
70def _streamdetails(item_id: str, media_type: MediaType, provider: str) -> StreamDetails:
71    """Build the streamdetails a healthy provider would hand back."""
72    return StreamDetails(
73        provider=provider,
74        item_id=item_id,
75        audio_format=AudioFormat(content_type=ContentType.MP3),
76        media_type=media_type,
77        stream_type=StreamType.HTTP,
78        path="http://localhost/clip.mp3",
79        duration=45,
80    )
81
82
83def _audio(
84    providers: dict[str, MagicMock], provider_filter: list[str] | None = None
85) -> StreamsAudio:
86    """
87    Build a StreamsAudio whose mass resolves the given provider instances.
88
89    :param providers: The provider instances the mass should hand back, by instance id.
90    :param provider_filter: The playback user's provider steering, omit for no playback user
91        (which makes every mapping on the item count as preferred).
92    """
93    mass = MagicMock()
94    for instance_id, provider in providers.items():
95        if isinstance(provider, MusicProvider):
96            provider.instance_id = instance_id
97            provider.domain = instance_id.split("--", maxsplit=1)[0]
98            provider.available = True
99            provider.is_streaming_provider = True
100    mass.get_provider.side_effect = lambda instance, **_kwargs: providers.get(instance)
101    # no other loaded instances to widen a mapping to
102    mass.providers = []
103    mass.player_queues.queue_data_or_none.return_value = (
104        MagicMock(userid="user1") if provider_filter else None
105    )
106    mass.webserver.auth.get_user = AsyncMock(
107        return_value=MagicMock(provider_filter=provider_filter) if provider_filter else None
108    )
109    mass.streams.get_config_value.return_value = -17
110    return StreamsAudio(mass)
111
112
113async def test_a_failing_provider_is_asked_only_once() -> None:
114    """A mapping that fails is not asked again by the widening pass."""
115    calls: list[str] = []
116
117    async def _fail(item_id: str, _media_type: MediaType) -> StreamDetails:
118        calls.append(item_id)
119        raise MediaNotFoundError(f"clip {item_id} failed TTS")
120
121    provider = MagicMock()
122    provider.get_stream_details = _fail
123    audio = _audio({INSTANCE: provider})
124
125    with pytest.raises(MediaNotFoundError):
126        await audio.get_stream_details(queue_item=_queue_item(_mapping(INSTANCE)))
127
128    assert calls == [ITEM_ID]
129
130
131async def test_the_widening_pass_still_reaches_a_provider_the_filter_held_back() -> None:
132    """A mapping the steering skipped in the first pass is tried by the second."""
133    calls: list[str] = []
134
135    async def _fail(item_id: str, _media_type: MediaType) -> StreamDetails:
136        calls.append(INSTANCE)
137        raise MediaNotFoundError(f"clip {item_id} failed TTS")
138
139    async def _succeed(item_id: str, media_type: MediaType) -> StreamDetails:
140        calls.append(OTHER_INSTANCE)
141        return _streamdetails(item_id, media_type, OTHER_INSTANCE)
142
143    failing = MagicMock()
144    failing.get_stream_details = _fail
145    working = MagicMock()
146    working.get_stream_details = _succeed
147    # steer to the failing instance so the working one is only reachable via the second pass
148    audio = _audio({INSTANCE: failing, OTHER_INSTANCE: working}, provider_filter=[INSTANCE])
149
150    streamdetails = await audio.get_stream_details(
151        # the steered mapping also sorts first, so a repeat of it would land before the
152        # widened one rather than depending on how the mapping set happens to iterate
153        queue_item=_queue_item(
154            _mapping(INSTANCE, content_type=ContentType.FLAC), _mapping(OTHER_INSTANCE)
155        )
156    )
157
158    assert streamdetails.provider == OTHER_INSTANCE
159    assert calls == [INSTANCE, OTHER_INSTANCE]
160
161
162async def test_two_mappings_on_one_provider_are_both_attempted() -> None:
163    """One provider carrying two items for the media item still gets asked for each."""
164    calls: list[str] = []
165
166    async def _by_item_id(item_id: str, media_type: MediaType) -> StreamDetails:
167        calls.append(item_id)
168        if item_id == "bad":
169            raise MediaNotFoundError(f"clip {item_id} failed TTS")
170        return _streamdetails(item_id, media_type, INSTANCE)
171
172    provider = MagicMock()
173    provider.get_stream_details = _by_item_id
174    audio = _audio({INSTANCE: provider})
175
176    streamdetails = await audio.get_stream_details(
177        # the failing mapping sorts first, so the good one is only reached by carrying on
178        # through the mappings rather than by a repeat attempt
179        queue_item=_queue_item(
180            _mapping(INSTANCE, item_id="bad", content_type=ContentType.FLAC),
181            _mapping(INSTANCE, item_id="good"),
182        )
183    )
184
185    assert streamdetails.item_id == "good"
186    assert calls == ["bad", "good"]
187
188
189def _music_provider(instance: str, has_slot: bool = True) -> MagicMock:
190    """
191    Build a streaming music provider test double.
192
193    :param instance: The instance id the provider is registered under.
194    :param has_slot: Whether the provider has a free source-stream slot.
195    """
196    provider = MagicMock(spec=MusicProvider)
197    provider.has_available_stream_slot = has_slot
198    provider.get_stream_details = AsyncMock(
199        return_value=_streamdetails(ITEM_ID, MediaType.SOUND_EFFECT, instance)
200    )
201    return provider
202
203
204async def test_mapping_quality_order_is_preserved_when_first_provider_is_busy() -> None:
205    """Capacity does not reorder a higher-quality mapping behind a lower-quality one."""
206    busy_instance = "tidal--busy"
207    available_instance = "tidal--available"
208    busy = _music_provider(busy_instance, has_slot=False)
209    available = _music_provider(available_instance)
210    audio = _audio({busy_instance: busy, available_instance: available})
211
212    streamdetails = await audio.get_stream_details(
213        queue_item=_queue_item(
214            _mapping(busy_instance, content_type=ContentType.FLAC),
215            _mapping(available_instance),
216        )
217    )
218
219    assert streamdetails.provider == busy_instance
220    busy.get_stream_details.assert_awaited_once()
221    available.get_stream_details.assert_not_awaited()
222
223
224async def test_busy_instance_preserves_playback_user_steering_order() -> None:
225    """A busy steered instance stays first; capacity is handled while acquiring the source."""
226    busy_instance = "tidal--preferred"
227    available_instance = "tidal--fallback"
228    busy = _music_provider(busy_instance, has_slot=False)
229    available = _music_provider(available_instance)
230    audio = _audio(
231        {busy_instance: busy, available_instance: available},
232        provider_filter=[busy_instance],
233    )
234
235    streamdetails = await audio.get_stream_details(
236        queue_item=_queue_item(
237            _mapping(busy_instance, content_type=ContentType.FLAC),
238            _mapping(available_instance),
239        )
240    )
241
242    assert streamdetails.provider == busy_instance
243    busy.get_stream_details.assert_awaited_once()
244    available.get_stream_details.assert_not_awaited()
245
246
247async def test_excluded_instance_is_skipped_including_its_cached_details() -> None:
248    """An excluded instance is passed over, and its unexpired details are not reused."""
249    excluded_instance = "tidal--busy"
250    available_instance = "tidal--available"
251    excluded = _music_provider(excluded_instance)
252    available = _music_provider(available_instance)
253    queue_item = _queue_item(
254        _mapping(excluded_instance, content_type=ContentType.FLAC),
255        _mapping(available_instance),
256    )
257    queue_item.streamdetails = _streamdetails(ITEM_ID, MediaType.SOUND_EFFECT, excluded_instance)
258    audio = _audio({excluded_instance: excluded, available_instance: available})
259
260    streamdetails = await audio.get_stream_details(
261        queue_item=queue_item,
262        excluded_provider_instances={excluded_instance},
263    )
264
265    assert streamdetails.provider == available_instance
266    excluded.get_stream_details.assert_not_awaited()
267
268
269async def test_mapping_falls_back_to_compatible_streaming_provider_instance() -> None:
270    """The same mapping item ID is retried on another loaded instance of its streaming domain."""
271    primary_instance = "tidal--primary"
272    fallback_instance = "tidal--fallback"
273    primary = _music_provider(primary_instance)
274    fallback = _music_provider(fallback_instance)
275    audio = _audio({primary_instance: primary, fallback_instance: fallback})
276    cast("MagicMock", audio.mass).providers = [primary, fallback]
277
278    streamdetails = await audio.get_stream_details(
279        queue_item=_queue_item(_mapping(primary_instance)),
280        excluded_provider_instances={primary_instance},
281    )
282
283    assert streamdetails.provider == fallback_instance
284    primary.get_stream_details.assert_not_awaited()
285    fallback.get_stream_details.assert_awaited_once_with(ITEM_ID, MediaType.SOUND_EFFECT)
286
287
288async def test_playback_user_steers_compatible_instance_within_mapping() -> None:
289    """Playback-user steering picks its compatible instance without changing mapping order."""
290    primary_instance = "tidal--primary"
291    preferred_instance = "tidal--preferred"
292    primary = _music_provider(primary_instance)
293    preferred = _music_provider(preferred_instance)
294    audio = _audio(
295        {primary_instance: primary, preferred_instance: preferred},
296        provider_filter=[preferred_instance],
297    )
298    cast("MagicMock", audio.mass).providers = [primary, preferred]
299
300    streamdetails = await audio.get_stream_details(
301        queue_item=_queue_item(_mapping(primary_instance))
302    )
303
304    assert streamdetails.provider == preferred_instance
305    preferred.get_stream_details.assert_awaited_once_with(ITEM_ID, MediaType.SOUND_EFFECT)
306    primary.get_stream_details.assert_not_awaited()
307
308
309async def test_playback_user_steering_precedes_cross_domain_quality() -> None:
310    """A lower-quality steered mapping is tried before widening to a higher-quality one."""
311    high_quality_instance = "tidal--high"
312    preferred_instance = "spotify--preferred"
313    high_quality = _music_provider(high_quality_instance)
314    preferred = _music_provider(preferred_instance)
315    audio = _audio(
316        {high_quality_instance: high_quality, preferred_instance: preferred},
317        provider_filter=[preferred_instance],
318    )
319
320    streamdetails = await audio.get_stream_details(
321        queue_item=_queue_item(
322            _mapping(high_quality_instance, content_type=ContentType.FLAC),
323            _mapping(preferred_instance),
324        )
325    )
326
327    assert streamdetails.provider == preferred_instance
328    preferred.get_stream_details.assert_awaited_once()
329    high_quality.get_stream_details.assert_not_awaited()
330