music-assistant-server

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