/
/
1"""
2Tests that resolving streamdetails asks each provider mapping at most once.
3
4``get_stream_details`` walks the provider mappings twice: the first pass is limited to the
5providers the user's provider filter steers to, the second widens to the rest. Without a
6provider filter every mapping counts as preferred, so both passes cover the same set and a
7mapping that failed would be asked again -- doubling the cost of every failure, which for a
8just-in-time renderer like AI Radio means a second full text-to-speech render.
9
10The mappings below are given distinct qualities wherever order matters, so the pass the
11loop reaches them in is fixed rather than left to the iteration order of a set.
12"""
13
14from __future__ import annotations
15
16from unittest.mock import AsyncMock, MagicMock
17
18import pytest
19from music_assistant_models.enums import ContentType, MediaType, StreamType
20from music_assistant_models.errors import MediaNotFoundError
21from music_assistant_models.media_items import AudioFormat, ProviderMapping, SoundEffect
22from music_assistant_models.queue_item import QueueItem
23from music_assistant_models.streamdetails import StreamDetails
24
25from music_assistant.controllers.streams.audio import StreamsAudio
26
27INSTANCE = "ai_radio--abc"
28OTHER_INSTANCE = "tidal--xyz"
29ITEM_ID = "session123_0"
30
31
32def _mapping(
33 instance: str, item_id: str = ITEM_ID, content_type: ContentType = ContentType.MP3
34) -> ProviderMapping:
35 """
36 Build a provider mapping.
37
38 :param instance: The provider instance the mapping points at.
39 :param item_id: The item id on that provider.
40 :param content_type: Drives the mapping's quality score, which decides the order the
41 mappings are tried in. Pass a lossless type to have a mapping tried first.
42 """
43 return ProviderMapping(
44 item_id=item_id,
45 provider_domain=instance.split("--", maxsplit=1)[0],
46 provider_instance=instance,
47 audio_format=AudioFormat(content_type=content_type),
48 )
49
50
51def _queue_item(*mappings: ProviderMapping) -> QueueItem:
52 """Build a queue item whose media item carries the given provider mappings."""
53 media_item = SoundEffect(
54 item_id=ITEM_ID,
55 provider=mappings[0].provider_instance,
56 name="Intro",
57 provider_mappings=set(mappings),
58 )
59 return QueueItem(
60 queue_id="q1",
61 queue_item_id="qi1",
62 name="Intro",
63 duration=None,
64 media_item=media_item,
65 )
66
67
68def _streamdetails(item_id: str, media_type: MediaType, provider: str) -> StreamDetails:
69 """Build the streamdetails a healthy provider would hand back."""
70 return StreamDetails(
71 provider=provider,
72 item_id=item_id,
73 audio_format=AudioFormat(content_type=ContentType.MP3),
74 media_type=media_type,
75 stream_type=StreamType.HTTP,
76 path="http://localhost/clip.mp3",
77 duration=45,
78 )
79
80
81def _audio(
82 providers: dict[str, MagicMock], provider_filter: list[str] | None = None
83) -> StreamsAudio:
84 """
85 Build a StreamsAudio whose mass resolves the given provider instances.
86
87 :param providers: The provider instances the mass should hand back, by instance id.
88 :param provider_filter: The playback user's provider steering, omit for no playback user
89 (which makes every mapping on the item count as preferred).
90 """
91 mass = MagicMock()
92 mass.get_provider.side_effect = lambda instance: providers.get(instance)
93 mass.player_queues.queue_data_or_none.return_value = (
94 MagicMock(userid="user1") if provider_filter else None
95 )
96 mass.webserver.auth.get_user = AsyncMock(
97 return_value=MagicMock(provider_filter=provider_filter) if provider_filter else None
98 )
99 mass.streams.get_config_value.return_value = -17
100 return StreamsAudio(mass)
101
102
103async def test_a_failing_provider_is_asked_only_once() -> None:
104 """A mapping that fails is not asked again by the widening pass."""
105 calls: list[str] = []
106
107 async def _fail(item_id: str, _media_type: MediaType) -> StreamDetails:
108 calls.append(item_id)
109 raise MediaNotFoundError(f"clip {item_id} failed TTS")
110
111 provider = MagicMock()
112 provider.get_stream_details = _fail
113 audio = _audio({INSTANCE: provider})
114
115 with pytest.raises(MediaNotFoundError):
116 await audio.get_stream_details(queue_item=_queue_item(_mapping(INSTANCE)))
117
118 assert calls == [ITEM_ID]
119
120
121async def test_the_widening_pass_still_reaches_a_provider_the_filter_held_back() -> None:
122 """A mapping the steering skipped in the first pass is tried by the second."""
123 calls: list[str] = []
124
125 async def _fail(item_id: str, _media_type: MediaType) -> StreamDetails:
126 calls.append(INSTANCE)
127 raise MediaNotFoundError(f"clip {item_id} failed TTS")
128
129 async def _succeed(item_id: str, media_type: MediaType) -> StreamDetails:
130 calls.append(OTHER_INSTANCE)
131 return _streamdetails(item_id, media_type, OTHER_INSTANCE)
132
133 failing = MagicMock()
134 failing.get_stream_details = _fail
135 working = MagicMock()
136 working.get_stream_details = _succeed
137 # steer to the failing instance so the working one is only reachable via the second pass
138 audio = _audio({INSTANCE: failing, OTHER_INSTANCE: working}, provider_filter=[INSTANCE])
139
140 streamdetails = await audio.get_stream_details(
141 # the steered mapping also sorts first, so a repeat of it would land before the
142 # widened one rather than depending on how the mapping set happens to iterate
143 queue_item=_queue_item(
144 _mapping(INSTANCE, content_type=ContentType.FLAC), _mapping(OTHER_INSTANCE)
145 )
146 )
147
148 assert streamdetails.provider == OTHER_INSTANCE
149 assert calls == [INSTANCE, OTHER_INSTANCE]
150
151
152async def test_two_mappings_on_one_provider_are_both_attempted() -> None:
153 """One provider carrying two items for the media item still gets asked for each."""
154 calls: list[str] = []
155
156 async def _by_item_id(item_id: str, media_type: MediaType) -> StreamDetails:
157 calls.append(item_id)
158 if item_id == "bad":
159 raise MediaNotFoundError(f"clip {item_id} failed TTS")
160 return _streamdetails(item_id, media_type, INSTANCE)
161
162 provider = MagicMock()
163 provider.get_stream_details = _by_item_id
164 audio = _audio({INSTANCE: provider})
165
166 streamdetails = await audio.get_stream_details(
167 # the failing mapping sorts first, so the good one is only reached by carrying on
168 # through the mappings rather than by a repeat attempt
169 queue_item=_queue_item(
170 _mapping(INSTANCE, item_id="bad", content_type=ContentType.FLAC),
171 _mapping(INSTANCE, item_id="good"),
172 )
173 )
174
175 assert streamdetails.item_id == "good"
176 assert calls == ["bad", "good"]
177