/
/
/
1"""Tests for the plugin engine discovery/selection helper."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.config_entries import ProviderConfig
10from music_assistant_models.enums import ConfigEntryType, ProviderType
11from music_assistant_models.provider import ProviderManifest
12
13from music_assistant.constants import CONF_PROVIDERS
14from music_assistant.helpers.plugin_engines import (
15 create_ai_engine_config_entries,
16 create_tts_engine_config_entries,
17 get_ai_engines,
18 get_tts_engines,
19 resolve_ai_engine,
20 resolve_tts_engine,
21 select_ai_engine,
22 select_tts_engine,
23)
24from music_assistant.models.plugin import AIEngine, PluginProvider, TTSEngine
25
26if TYPE_CHECKING:
27 from music_assistant.mass import MusicAssistant
28
29CONSUMER_INSTANCE = "consumer--test"
30
31
32def _create_plugin(instance_id: str, name: str | None = None) -> MagicMock:
33 """Create a mock plugin provider exposing no engines."""
34 provider = MagicMock(spec=PluginProvider)
35 provider.instance_id = instance_id
36 provider.name = name or instance_id
37 provider.get_ai_engines = AsyncMock(return_value=[])
38 provider.get_tts_engines = AsyncMock(return_value=[])
39 return provider
40
41
42def _create_mass(*providers: MagicMock) -> MagicMock:
43 """Create a mock MusicAssistant returning the given providers for any feature."""
44 mass = MagicMock()
45 mass.get_providers_supporting_feature.return_value = list(providers)
46 return mass
47
48
49def _create_consumer(
50 mass: MusicAssistant, *providers: MagicMock, values: dict[str, Any] | None = None
51) -> PluginProvider:
52 """Create a stored consumer provider on a real config controller, serving the plugins."""
53 mass.config.set(
54 f"{CONF_PROVIDERS}/{CONSUMER_INSTANCE}",
55 {
56 "type": ProviderType.PLUGIN.value,
57 "domain": "consumer",
58 "instance_id": CONSUMER_INSTANCE,
59 "values": dict(values or {}),
60 },
61 )
62 mass.get_providers_supporting_feature = MagicMock( # type: ignore[method-assign]
63 return_value=list(providers)
64 )
65 manifest = ProviderManifest(
66 type=ProviderType.PLUGIN,
67 domain="consumer",
68 name="Consumer",
69 description="",
70 codeowners=[],
71 )
72 config = ProviderConfig(
73 values={},
74 type=ProviderType.PLUGIN,
75 domain="consumer",
76 instance_id=CONSUMER_INSTANCE,
77 )
78 return PluginProvider(mass, manifest, config)
79
80
81def test_engine_uid_composes_provider_and_engine_id() -> None:
82 """The uid is the owning provider's instance id joined with the engine id."""
83 provider = _create_plugin("hass--abc")
84 assert AIEngine(id="ai_task.google", name="Google", provider=provider).uid == (
85 "hass--abc/ai_task.google"
86 )
87
88
89async def test_engines_ordered_by_provider_then_name() -> None:
90 """Provider order is preserved and engines are sorted by name within a provider."""
91 first = _create_plugin("plugin_a")
92 second = _create_plugin("plugin_b")
93 first.get_ai_engines.return_value = [
94 AIEngine(id="zulu", name="Zulu", provider=first),
95 AIEngine(id="alpha", name="Alpha", provider=first),
96 ]
97 second.get_ai_engines.return_value = [
98 AIEngine(id="mike", name="Mike", provider=second),
99 ]
100 mass = _create_mass(first, second)
101 assert [engine.uid for engine in await get_ai_engines(mass)] == [
102 "plugin_a/alpha",
103 "plugin_a/zulu",
104 "plugin_b/mike",
105 ]
106
107
108async def test_failing_provider_is_skipped() -> None:
109 """A provider raising while listing its engines is skipped, not propagated."""
110 broken = _create_plugin("broken")
111 healthy = _create_plugin("healthy")
112 broken.get_ai_engines.side_effect = RuntimeError("boom")
113 healthy.get_ai_engines.return_value = [AIEngine(id="one", name="One", provider=healthy)]
114 mass = _create_mass(broken, healthy)
115 assert [engine.uid for engine in await get_ai_engines(mass)] == ["healthy/one"]
116
117
118async def test_tts_engines_are_collected() -> None:
119 """TTS discovery uses the plugin's TTS engine listing."""
120 plugin = _create_plugin("plugin_a")
121 plugin.get_tts_engines.return_value = [TTSEngine(id="voice", name="Voice", provider=plugin)]
122 mass = _create_mass(plugin)
123 assert [engine.uid for engine in await get_tts_engines(mass)] == ["plugin_a/voice"]
124
125
126@pytest.mark.parametrize("selected", [None, ""])
127async def test_resolve_without_a_selection_returns_nothing(selected: str | None) -> None:
128 """An unset selection resolves to nothing instead of an arbitrary engine."""
129 plugin = _create_plugin("plugin_a")
130 plugin.get_ai_engines.return_value = [
131 AIEngine(id="alpha", name="Alpha", provider=plugin),
132 AIEngine(id="bravo", name="Bravo", provider=plugin),
133 ]
134 mass = _create_mass(plugin)
135 assert await resolve_ai_engine(mass, selected) is None
136
137
138async def test_resolve_concrete_uid_returns_the_selected_engine() -> None:
139 """A concrete selection yields exactly the matching engine."""
140 plugin = _create_plugin("plugin_a")
141 plugin.get_tts_engines.return_value = [
142 TTSEngine(id="alpha", name="Alpha", provider=plugin),
143 TTSEngine(id="bravo", name="Bravo", provider=plugin),
144 ]
145 mass = _create_mass(plugin)
146 resolved = await resolve_tts_engine(mass, "plugin_a/bravo")
147 assert resolved is not None
148 assert resolved.uid == "plugin_a/bravo"
149
150
151async def test_resolve_vanished_uid_returns_none() -> None:
152 """A selection that no longer exists resolves to nothing instead of another engine."""
153 plugin = _create_plugin("plugin_a")
154 plugin.get_ai_engines.return_value = [AIEngine(id="alpha", name="Alpha", provider=plugin)]
155 mass = _create_mass(plugin)
156 assert await resolve_ai_engine(mass, "plugin_a/gone") is None
157
158
159async def test_select_stores_the_first_engine_in_values(mass_minimal: MusicAssistant) -> None:
160 """A consumer without a stored selection adopts the first engine as a concrete uid."""
161 plugin = _create_plugin("plugin_a")
162 plugin.get_ai_engines.return_value = [
163 AIEngine(id="alpha", name="Alpha", provider=plugin),
164 AIEngine(id="bravo", name="Bravo", provider=plugin),
165 ]
166 consumer = _create_consumer(mass_minimal, plugin)
167
168 engine = await select_ai_engine(consumer, "ai_engine")
169 assert engine is not None
170 assert engine.uid == "plugin_a/alpha"
171
172 assert (
173 mass_minimal.config.get_raw_provider_config_value(CONSUMER_INSTANCE, "ai_engine")
174 == "plugin_a/alpha"
175 )
176
177
178async def test_select_stores_the_first_engine_encrypted_in_setup_data(
179 mass_minimal: MusicAssistant,
180) -> None:
181 """The setup_data variant persists the uid encrypted at rest, like the setup flows do."""
182 plugin = _create_plugin("plugin_a")
183 plugin.get_tts_engines.return_value = [TTSEngine(id="voice", name="Voice", provider=plugin)]
184 consumer = _create_consumer(mass_minimal, plugin)
185
186 engine = await select_tts_engine(consumer, "tts_engine", in_setup_data=True)
187
188 assert engine is not None
189 assert engine.uid == "plugin_a/voice"
190 stored = mass_minimal.config.get(f"{CONF_PROVIDERS}/{CONSUMER_INSTANCE}/setup_data/tts_engine")
191 assert stored != "plugin_a/voice"
192 assert mass_minimal.config.decrypt_string(stored) == "plugin_a/voice"
193 assert (
194 mass_minimal.config.get_provider_setup_value(CONSUMER_INSTANCE, "tts_engine")
195 == "plugin_a/voice"
196 )
197
198
199async def test_select_keeps_an_existing_selection(mass_minimal: MusicAssistant) -> None:
200 """An already stored selection is returned untouched, never replaced by the first engine."""
201 plugin = _create_plugin("plugin_a")
202 plugin.get_ai_engines.return_value = [
203 AIEngine(id="alpha", name="Alpha", provider=plugin),
204 AIEngine(id="bravo", name="Bravo", provider=plugin),
205 ]
206 consumer = _create_consumer(mass_minimal, plugin, values={"ai_engine": "plugin_a/bravo"})
207
208 engine = await select_ai_engine(consumer, "ai_engine")
209 assert engine is not None
210 assert engine.uid == "plugin_a/bravo"
211
212 assert (
213 mass_minimal.config.get_raw_provider_config_value(CONSUMER_INSTANCE, "ai_engine")
214 == "plugin_a/bravo"
215 )
216
217
218async def test_select_without_engines_stores_nothing(mass_minimal: MusicAssistant) -> None:
219 """Without any engine nothing is stored, so a later load can still seed a real one."""
220 consumer = _create_consumer(mass_minimal)
221
222 assert await select_ai_engine(consumer, "ai_engine") is None
223
224 assert mass_minimal.config.get_raw_provider_config_value(CONSUMER_INSTANCE, "ai_engine") is None
225
226
227async def test_selection_that_vanished_does_not_fall_back(
228 mass_minimal: MusicAssistant,
229) -> None:
230 """A stored engine that disappeared is reported missing, not swapped for another."""
231 plugin = _create_plugin("plugin_a")
232 plugin.get_ai_engines.return_value = [AIEngine(id="alpha", name="Alpha", provider=plugin)]
233 consumer = _create_consumer(mass_minimal, plugin, values={"ai_engine": "plugin_a/gone"})
234
235 assert await select_ai_engine(consumer, "ai_engine") is None
236
237 assert (
238 mass_minimal.config.get_raw_provider_config_value(CONSUMER_INSTANCE, "ai_engine")
239 == "plugin_a/gone"
240 )
241
242
243async def test_config_entries_list_the_concrete_engines() -> None:
244 """The picker offers the available engines only, with no preselected default."""
245 plugin = _create_plugin("plugin_a")
246 plugin.get_ai_engines.return_value = [AIEngine(id="alpha", name="Alpha", provider=plugin)]
247 mass = _create_mass(plugin)
248 entries = await create_ai_engine_config_entries(mass, "ai_engine", depends_on="use_ai")
249 assert len(entries) == 1
250 entry = entries[0]
251 assert entry.key == "ai_engine"
252 assert entry.default_value is None
253 assert entry.depends_on == "use_ai"
254 assert entry.category == "features"
255 assert entry.read_only is False
256 assert [(option.value, option.title) for option in entry.options] == [
257 ("plugin_a/alpha", "plugin_a | Alpha"),
258 ]
259
260
261async def test_config_entries_name_the_plugin_each_engine_came_from() -> None:
262 """Engines from different plugins stay distinguishable even when named the same."""
263 first = _create_plugin("plugin_a", name="Home Assistant")
264 second = _create_plugin("plugin_b", name="Chatterbox")
265 first.get_ai_engines.return_value = [AIEngine(id="shared", name="Shared", provider=first)]
266 second.get_ai_engines.return_value = [AIEngine(id="shared", name="Shared", provider=second)]
267 mass = _create_mass(first, second)
268
269 entries = await create_ai_engine_config_entries(mass, "ai_engine")
270
271 assert [(option.value, option.title) for option in entries[0].options] == [
272 ("plugin_a/shared", "Home Assistant | Shared"),
273 ("plugin_b/shared", "Chatterbox | Shared"),
274 ]
275
276
277async def test_config_entries_without_engines_are_read_only_with_alert() -> None:
278 """Without any engine the picker is read-only and an alert entry is appended."""
279 mass = _create_mass()
280 entries = await create_tts_engine_config_entries(mass, "tts_engine")
281 assert len(entries) == 2
282 picker, alert = entries
283 assert picker.read_only is True
284 assert picker.options == []
285 assert alert.key == "tts_engine_unavailable"
286 assert alert.type == ConfigEntryType.ALERT
287
288
289async def test_create_config_entries_alert_follows_picker_visibility() -> None:
290 """Test the unavailable alert is gated by the same entry as the picker."""
291 mass = _create_mass()
292 picker, alert = await create_ai_engine_config_entries(
293 mass, "ai_engine", depends_on="ai_descriptions"
294 )
295 assert picker.depends_on == "ai_descriptions"
296 assert alert.depends_on == "ai_descriptions"
297