music-assistant-server

1"""Unit tests for AI Radio provider helper logic."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from pathlib import Path
8from types import SimpleNamespace
9from typing import Any, cast
10from unittest.mock import AsyncMock, MagicMock
11
12import pytest
13from music_assistant_models.auth import Scope
14from music_assistant_models.enums import EventType, PlaybackState, ProviderFeature
15from music_assistant_models.errors import (
16    InvalidDataError,
17    PlayerUnavailableError,
18    SetupFailedError,
19)
20
21from music_assistant.models.plugin import AIEngine, PluginProvider, TTSEngine
22from music_assistant.providers.ai_radio import provider as ai_radio_provider
23from music_assistant.providers.ai_radio.constants import (
24    CONF_AI_ENGINE,
25    CONF_TTS_ENGINE,
26    ENGINE_RETRY_DELAY,
27    MAX_FINISHED_SESSIONS,
28)
29from music_assistant.providers.ai_radio.models import DJQueueState, SessionState
30from music_assistant.providers.ai_radio.provider import AIRadioProvider
31
32
33def _close(coro: Any) -> None:
34    """Close an un-awaited coroutine so the test does not warn."""
35    coro.close()
36
37
38def _recording_stop(recorder: list[str]) -> Any:
39    """Return an async queue-stop stub that records the queue ids it was called with."""
40
41    async def _stop(queue_id: str) -> None:
42        recorder.append(queue_id)
43
44    return _stop
45
46
47def _make_provider() -> AIRadioProvider:
48    """Create a minimal AIRadioProvider object without full init."""
49    provider = AIRadioProvider.__new__(AIRadioProvider)
50    provider._sessions = {}
51    provider._session_lock = asyncio.Lock()
52    return provider
53
54
55def _make_dynamic_provider(player_obj: object | None, default_player_id: str) -> AIRadioProvider:
56    """Create a minimal provider object suitable for dynamic-mode validation tests."""
57    provider = _make_provider()
58    provider._stations = {
59        "station_a": {
60            "id": "station_a",
61            "name": "Station A",
62            "source_playlist_id": "1",
63            "source_playlist_provider": "library",
64            "default_player_id": default_player_id,
65        }
66    }
67    provider.mass = cast(
68        "Any",
69        SimpleNamespace(
70            players=SimpleNamespace(get_player=lambda _player_id: player_obj),
71        ),
72    )
73    return provider
74
75
76@pytest.fixture
77def provider(tmp_path: Path) -> AIRadioProvider:
78    """Build a minimal AIRadioProvider instance for host/station CRUD tests."""
79    instance = AIRadioProvider.__new__(AIRadioProvider)
80    instance.logger = logging.getLogger("test.ai_radio.provider")
81    instance._station_lock = asyncio.Lock()
82    instance._stations = {}
83    instance._hosts = {}
84    instance._dj_queues = {}
85    instance._hosts_file = tmp_path / "hosts.json"
86    instance._sections = {item["id"]: item for item in instance._default_sections_template()}
87    return instance
88
89
90def test_resolve_session_for_stop_by_session_id() -> None:
91    """Resolve explicit session id directly."""
92    provider = _make_provider()
93    session = SessionState(session_id="s1", station_id="st")
94    provider._sessions[session.session_id] = session
95
96    resolved = provider._resolve_session_for_stop(session_id="s1", station_id=None)
97
98    assert resolved is session
99
100
101def test_resolve_session_for_stop_uses_latest_running_for_station() -> None:
102    """Resolve latest running session for selected station."""
103    provider = _make_provider()
104    older = SessionState(
105        session_id="s_old",
106        station_id="station_a",
107        created_at="2026-01-01T10:00:00+00:00",
108    )
109    newer = SessionState(
110        session_id="s_new",
111        station_id="station_a",
112        created_at="2026-01-01T11:00:00+00:00",
113    )
114    other = SessionState(
115        session_id="s_other",
116        station_id="station_b",
117        created_at="2026-01-01T12:00:00+00:00",
118    )
119    provider._sessions = {s.session_id: s for s in (older, newer, other)}
120
121    resolved = provider._resolve_session_for_stop(session_id=None, station_id="station_a")
122
123    assert resolved.session_id == "s_new"
124
125
126def test_resolve_session_for_stop_raises_when_nothing_running() -> None:
127    """Raise when no running sessions exist."""
128    provider = _make_provider()
129
130    with pytest.raises(KeyError, match="No active AI Radio run found"):
131        provider._resolve_session_for_stop(session_id=None, station_id=None)
132
133
134def test_session_state_as_dict_reports_the_resolved_queue() -> None:
135    """Serialize the queue id once a run has resolved its target queue."""
136    session = SessionState(session_id="s1", station_id="st", queue_id="living_room")
137
138    assert session.as_dict()["queue_id"] == "living_room"
139
140
141def test_session_state_as_dict_reports_no_queue_before_resolution() -> None:
142    """Report no queue id before a run resolves its target queue."""
143    session = SessionState(session_id="s1", station_id="st")
144
145    assert session.as_dict()["queue_id"] is None
146
147
148@pytest.mark.asyncio
149async def test_start_run_dynamic_requires_player_id() -> None:
150    """Reject dynamic run start when no player is configured."""
151    provider = _make_dynamic_provider(player_obj=None, default_player_id="")
152
153    with pytest.raises(InvalidDataError, match="requires a target player"):
154        await provider.start_run(station_id="station_a")
155
156
157@pytest.mark.asyncio
158async def test_start_run_rejects_unknown_station() -> None:
159    """Reject starting a run for a station that does not exist."""
160    provider = _make_provider()
161    provider._stations = {}
162
163    with pytest.raises(KeyError, match="Unknown station id: missing_station"):
164        await provider.start_run(station_id="missing_station")
165
166
167@pytest.mark.asyncio
168async def test_start_run_dynamic_rejects_unavailable_player() -> None:
169    """Reject dynamic run start when configured player is unavailable."""
170    unavailable_player = SimpleNamespace(player_id="living_room", available=False, enabled=True)
171    provider = _make_dynamic_provider(
172        player_obj=unavailable_player,
173        default_player_id="living_room",
174    )
175
176    with pytest.raises(InvalidDataError, match="Target player is unavailable"):
177        await provider.start_run(station_id="station_a")
178
179
180@pytest.mark.asyncio
181async def test_start_run_dynamic_rejects_negative_source_playtime_cap_override() -> None:
182    """Reject dynamic run start when source playtime cap override is negative."""
183    provider = _make_dynamic_provider(player_obj=None, default_player_id="")
184
185    with pytest.raises(InvalidDataError, match="dynamic_source_playtime_cap_override must be >= 0"):
186        await provider.start_run(
187            station_id="station_a",
188            dynamic_source_playtime_cap_override=-1,
189        )
190
191
192@pytest.mark.asyncio
193async def test_start_run_dynamic_rejects_disabled_player() -> None:
194    """Reject dynamic run start when target player is disabled."""
195    disabled_player = SimpleNamespace(player_id="living_room", available=True, enabled=False)
196    provider = _make_dynamic_provider(
197        player_obj=disabled_player,
198        default_player_id="living_room",
199    )
200
201    with pytest.raises(InvalidDataError, match="Target player is disabled"):
202        await provider.start_run(station_id="station_a")
203
204
205@pytest.mark.asyncio
206async def test_stop_run_rejects_already_completed_session() -> None:
207    """Reject stopping a session that is already completed."""
208    provider = _make_provider()
209    provider.logger = cast(
210        "Any",
211        SimpleNamespace(debug=lambda *_a, **_kw: None, info=lambda *_a, **_kw: None),
212    )
213    provider._sessions["s_done"] = SessionState(
214        session_id="s_done",
215        station_id="st",
216        status="completed",
217        ended_at="2026-01-01T10:00:00+00:00",
218    )
219
220    with pytest.raises(InvalidDataError):
221        await provider.stop_run(session_id="s_done")
222
223
224@pytest.mark.asyncio
225async def test_stop_run_accepts_running_session_by_id() -> None:
226    """Stop a running session resolved by explicit session id."""
227    provider = _make_provider()
228    provider.logger = cast(
229        "Any",
230        SimpleNamespace(debug=lambda *_a, **_kw: None, info=lambda *_a, **_kw: None),
231    )
232    provider._sessions["s_run"] = SessionState(
233        session_id="s_run",
234        station_id="st",
235        status="running",
236    )
237
238    result = await provider.stop_run(session_id="s_run")
239
240    assert result["status"] == "stopped"
241
242
243@pytest.mark.asyncio
244async def test_start_run_prunes_oldest_finished_sessions() -> None:
245    """Drop the oldest finished sessions beyond the retention limit on run start."""
246    provider = _make_provider()
247    provider.logger = cast(
248        "Any",
249        SimpleNamespace(debug=lambda *_a, **_kw: None, info=lambda *_a, **_kw: None),
250    )
251    player = SimpleNamespace(player_id="living_room", available=True, enabled=True)
252    provider._stations = {
253        "station_a": {
254            "id": "station_a",
255            "name": "Station A",
256            "source_playlist_id": "1",
257            "source_playlist_provider": "library",
258            "default_player_id": "living_room",
259            "host_id": "host_a",
260        }
261    }
262    provider._hosts = {"host_a": {"id": "host_a", "name": "Host A"}}
263    provider._sections = {}
264    provider.mass = cast(
265        "Any",
266        SimpleNamespace(
267            players=SimpleNamespace(get_player=lambda _player_id: player),
268            create_task=lambda coro, **_kw: coro.close(),
269        ),
270    )
271    for index in range(MAX_FINISHED_SESSIONS + 5):
272        session_id = f"s_{index}"
273        provider._sessions[session_id] = SessionState(
274            session_id=session_id,
275            station_id="station_a",
276            status="completed",
277            created_at=f"2026-01-01T10:{index:02d}:00+00:00",
278        )
279
280    await provider.start_run(station_id="station_a")
281
282    finished = [s for s in provider._sessions.values() if s.status != "running"]
283    assert len(finished) == MAX_FINISHED_SESSIONS
284    # the five oldest sessions are gone, the newest finished ones remain
285    for index in range(5):
286        assert f"s_{index}" not in provider._sessions
287    assert f"s_{MAX_FINISHED_SESSIONS + 4}" in provider._sessions
288
289
290@pytest.mark.asyncio
291async def test_validate_station_does_not_mutate_shared_sections() -> None:
292    """Keep shared hosts and sections untouched when a station payload is only validated."""
293    provider = _make_provider()
294    provider._stations = {}
295    provider._sections = {}
296    provider._hosts = {"host_a": {"id": "host_a", "name": "Host A"}}
297    provider._station_lock = asyncio.Lock()
298    station = {
299        "id": "station_a",
300        "name": "Station A",
301        "source_playlist_id": "playlist-1",
302        "source_playlist_provider": "library",
303        "host_id": "host_a",
304    }
305
306    normalized = await provider.validate_station(station)
307
308    assert normalized["host_id"] == "host_a"
309    assert provider._sections == {}
310    assert provider._hosts == {"host_a": {"id": "host_a", "name": "Host A"}}
311
312
313@pytest.mark.asyncio
314async def test_station_template_points_at_an_existing_host(provider: Any) -> None:
315    """The template's host has to be one the install really has, or saving it is rejected."""
316    provider._hosts = {
317        "music_nerd": {"id": "music_nerd", "name": "Music nerd"},
318        "chill_dj": {"id": "chill_dj", "name": "Chill DJ"},
319    }
320
321    template = await provider.station_template()
322
323    # lowest by name, the order hosts are listed in
324    assert template["host_id"] == "chill_dj"
325    # the station validator rejects any other host, playlist aside
326    filled_in = {**template, "source_playlist_id": "playlist-1"}
327    assert (await provider.validate_station(filled_in))["host_id"] == "chill_dj"
328
329
330@pytest.mark.asyncio
331async def test_station_template_falls_back_to_the_host_template_id(provider: Any) -> None:
332    """With no hosts yet, the template pairs with the host the host template creates."""
333    template = await provider.station_template()
334
335    assert template["host_id"] == "default_host"
336
337
338@pytest.mark.asyncio
339async def test_host_crud_roundtrip(provider: Any) -> None:
340    """Create, list, fetch and delete a host through the public CRUD API."""
341    template = await provider.host_template()
342    saved = await provider.save_host(template)
343    assert saved["id"] == "default_host"
344    assert [h["id"] for h in await provider.list_hosts()] == ["default_host"]
345    fetched = await provider.get_host("default_host")
346    assert fetched["name"] == saved["name"]
347    await provider.delete_host("default_host")
348    assert await provider.list_hosts() == []
349
350
351@pytest.mark.asyncio
352async def test_list_host_presets_returns_every_preset_with_sections(provider: Any) -> None:
353    """The presets command returns each bundled preset paired with its sections."""
354    expected = provider._default_preset_hosts()
355
356    presets = await provider.list_host_presets()
357
358    assert len(presets) == len(expected)
359    assert {entry["host"]["id"] for entry in presets} == {host["id"] for host, _ in expected}
360    for entry in presets:
361        assert set(entry) == {"host", "sections"}
362        assert entry["sections"]
363
364
365@pytest.mark.asyncio
366async def test_list_host_presets_does_not_mutate_stored_state(provider: Any) -> None:
367    """Presets are templates: fetching them must not touch stored hosts or sections."""
368    presets = await provider.list_host_presets()
369
370    presets[0]["host"]["name"] = "mutated"
371    presets[0]["sections"][0]["name"] = "mutated"
372
373    assert provider._hosts == {}
374    fresh = await provider.list_host_presets()
375    assert fresh[0]["host"]["name"] != "mutated"
376    assert fresh[0]["sections"][0]["name"] != "mutated"
377
378
379@pytest.mark.asyncio
380async def test_save_section_leaves_the_stations_file_alone(provider: Any, tmp_path: Path) -> None:
381    """Sections no longer live inside stations, so saving one must not rewrite them."""
382    provider._sections_file = tmp_path / "sections.json"
383    provider._stations_file = tmp_path / "stations.json"
384    provider._stations_file.write_text("untouched")
385
386    await provider.save_section(
387        {"id": "New_Section", "name": "New Section", "type": "ai_text", "prompt": "Say something"}
388    )
389
390    assert "New_Section" in provider._sections
391    assert provider._stations_file.read_text() == "untouched"
392
393
394async def test_delete_host_refuses_when_station_references_it(provider: Any) -> None:
395    """Refuse to delete a host that a station still references."""
396    saved = await provider.save_host(await provider.host_template())
397    provider._stations["station_a"] = {
398        "id": "station_a",
399        "name": "Station A",
400        "source_playlist_id": "p1",
401        "source_playlist_provider": "library",
402        "default_player_id": "",
403        "max_duration_minutes": 0.0,
404        "shuffle_source_tracks": True,
405        "host_id": saved["id"],
406    }
407    with pytest.raises(InvalidDataError):
408        await provider.delete_host(saved["id"])
409
410
411@pytest.mark.asyncio
412async def test_delete_host_refuses_when_it_is_an_active_queue_dj(provider: Any) -> None:
413    """Refuse to delete a host that is the active DJ on a queue."""
414    saved = await provider.save_host(await provider.host_template())
415    provider._dj_queues["queue-1"] = DJQueueState(
416        queue_id="queue-1",
417        host_id=saved["id"],
418        dj_session_id="dj0123456789",
419    )
420
421    with pytest.raises(InvalidDataError, match="is the active DJ on queues: queue-1"):
422        await provider.delete_host(saved["id"])
423
424
425@pytest.mark.asyncio
426async def test_concurrent_start_run_calls_respect_the_run_limit() -> None:
427    """
428    Concurrent start_run calls must not both get past the concurrency guards.
429
430    The guards and the session insert are one critical section; without it an await
431    introduced between them would let both callers observe zero running sessions.
432    """
433    player = SimpleNamespace(player_id="living_room", available=True, enabled=True)
434    provider = _make_provider()
435    provider.logger = logging.getLogger("tests.ai_radio.provider")
436    provider._stations = {
437        "station_a": {
438            "id": "station_a",
439            "name": "Station A",
440            "default_player_id": "living_room",
441            "host_id": "host_a",
442        },
443        "station_b": {
444            "id": "station_b",
445            "name": "Station B",
446            "default_player_id": "living_room",
447            "host_id": "host_a",
448        },
449    }
450    provider._hosts = {"host_a": {"id": "host_a", "name": "Host A"}}
451    provider._sections = {}
452    provider.mass = cast(
453        "Any",
454        SimpleNamespace(
455            players=SimpleNamespace(get_player=lambda _player_id: player),
456            create_task=lambda coro, **_kw: _close(coro),
457        ),
458    )
459
460    results = await asyncio.gather(
461        provider.start_run(station_id="station_a"),
462        provider.start_run(station_id="station_b"),
463        return_exceptions=True,
464    )
465
466    started = [item for item in results if isinstance(item, dict)]
467    rejected = [item for item in results if isinstance(item, InvalidDataError)]
468    assert len(started) == 1
469    assert len(rejected) == 1
470    assert "Max concurrent runs reached" in str(rejected[0])
471
472
473@pytest.mark.asyncio
474async def test_stop_run_stops_the_queue_it_owns() -> None:
475    """Stop playback on the target queue when the show is stopped from the UI."""
476    provider = _make_provider()
477    provider.logger = cast(
478        "Any",
479        SimpleNamespace(debug=lambda *_a, **_kw: None, info=lambda *_a, **_kw: None),
480    )
481    stopped: list[str] = []
482    provider.mass = cast(
483        "Any",
484        SimpleNamespace(
485            player_queues=SimpleNamespace(
486                get=lambda _queue_id: SimpleNamespace(state=PlaybackState.PLAYING, current_index=3),
487                stop=_recording_stop(stopped),
488            )
489        ),
490    )
491    provider._sessions["s_run"] = SessionState(
492        session_id="s_run",
493        station_id="st",
494        status="running",
495        queue_id="living_room",
496    )
497
498    result = await provider.stop_run(session_id="s_run")
499
500    assert result["status"] == "stopped"
501    assert stopped == ["living_room"]
502
503
504@pytest.mark.asyncio
505async def test_stop_run_survives_an_unavailable_player() -> None:
506    """Still mark the session stopped when the target player has gone away."""
507    provider = _make_provider()
508    provider.logger = cast(
509        "Any",
510        SimpleNamespace(debug=lambda *_a, **_kw: None, info=lambda *_a, **_kw: None),
511    )
512
513    async def _raise(queue_id: str) -> None:
514        raise PlayerUnavailableError(f"Player {queue_id} is not available")
515
516    provider.mass = cast(
517        "Any",
518        SimpleNamespace(
519            player_queues=SimpleNamespace(
520                get=lambda _queue_id: SimpleNamespace(state=PlaybackState.PLAYING, current_index=3),
521                stop=_raise,
522            )
523        ),
524    )
525    provider._sessions["s_run"] = SessionState(
526        session_id="s_run",
527        station_id="st",
528        status="running",
529        queue_id="living_room",
530    )
531
532    result = await provider.stop_run(session_id="s_run")
533
534    assert result["status"] == "stopped"
535
536
537def _make_engine_provider(
538    plugins: list[Any],
539    setup_values: dict[str, Any] | None = None,
540) -> tuple[AIRadioProvider, list[Any], dict[str, Any]]:
541    """
542    Create a provider whose mass serves the given plugins.
543
544    :return: The provider, the list its event subscribers land in, and the (plain text)
545        stand-in for its stored setup_data.
546    """
547    provider = _make_provider()
548    provider.config = cast("Any", SimpleNamespace(instance_id="ai_radio", setup_data={}))
549    subscribers: list[Any] = []
550    stored = dict(setup_values or {})
551
552    def _subscribe(callback: Any, *_args: Any, **_kwargs: Any) -> Any:
553        subscribers.append(callback)
554        return lambda: subscribers.remove(callback)
555
556    def _providers(feature: ProviderFeature, **_kwargs: Any) -> list[Any]:
557        attribute = "get_ai_engines" if feature == ProviderFeature.AI_QUERY else "get_tts_engines"
558        return [plugin for plugin in plugins if getattr(plugin, attribute).return_value]
559
560    mass = MagicMock()
561    mass.closing = False
562    mass.subscribe.side_effect = _subscribe
563    # mirror mass.create_task, which runs the coroutine up to its first suspension
564    # before handing back the task
565    mass.create_task.side_effect = lambda coro, **_kwargs: asyncio.Task(
566        coro, loop=asyncio.get_running_loop(), eager_start=True
567    )
568    mass.get_providers_supporting_feature.side_effect = _providers
569    mass.config.get_provider_setup_value.side_effect = lambda _instance_id, key, default=None: (
570        stored.get(key, default)
571    )
572    provider.mass = cast("Any", mass)
573    provider.logger = logging.getLogger("test.ai_radio")
574    provider._dj_queues = {}
575    provider._unloading = False
576    provider._engine_recheck_task = None
577    provider._unregister_handles = []
578    provider._update_setup_data = cast(  # type: ignore[method-assign]
579        "Any", lambda key, value, **_kwargs: stored.__setitem__(key, value)
580    )
581    return provider, subscribers, stored
582
583
584def _record_unload_with_error(provider: AIRadioProvider) -> list[Any]:
585    """Capture the errors the provider unloads itself with instead of really unloading."""
586    errors: list[Any] = []
587    provider.unload_with_error = cast(  # type: ignore[method-assign]
588        "Any", errors.append
589    )
590    return errors
591
592
593def _make_engine_plugin(instance_id: str, ai_ids: list[str], tts_ids: list[str]) -> MagicMock:
594    """Create a mock plugin provider exposing the given AI and TTS engines."""
595    plugin = MagicMock(spec=PluginProvider)
596    plugin.instance_id = instance_id
597    plugin.get_ai_engines = AsyncMock(
598        return_value=[AIEngine(id=engine, name=engine, provider=plugin) for engine in ai_ids]
599    )
600    plugin.get_tts_engines = AsyncMock(
601        return_value=[TTSEngine(id=engine, name=engine, provider=plugin) for engine in tts_ids]
602    )
603    return plugin
604
605
606async def test_wait_for_engines_seeds_a_concrete_selection() -> None:
607    """An instance without a stored selection adopts concrete engines and waits for nothing."""
608    provider, subscribers, stored = _make_engine_provider(
609        [_make_engine_plugin("p1", ["ai"], ["tts"])]
610    )
611
612    await provider._wait_for_engines()
613
614    assert stored == {CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
615    assert subscribers == []
616
617
618async def test_wait_for_engines_keeps_an_existing_selection() -> None:
619    """A stored selection survives the load instead of being reseeded to the first engine."""
620    provider, _, stored = _make_engine_provider(
621        [_make_engine_plugin("p1", ["ai", "other"], ["tts"])],
622        setup_values={CONF_AI_ENGINE: "p1/other"},
623    )
624
625    await provider._wait_for_engines()
626
627    assert stored[CONF_AI_ENGINE] == "p1/other"
628
629
630async def test_wait_for_engines_fails_the_load_when_no_engine_appears(
631    monkeypatch: pytest.MonkeyPatch,
632) -> None:
633    """A provider without engines refuses to load so the UI offers a reconfigure."""
634    provider, subscribers, stored = _make_engine_provider([])
635    monkeypatch.setattr(ai_radio_provider, "ENGINE_DISCOVERY_TIMEOUT", 0.05)
636
637    with pytest.raises(SetupFailedError) as error:
638        async with asyncio.timeout(1):
639            await provider._wait_for_engines()
640
641    assert error.value.translation_key == "ai_radio_no_ai_engine"
642    assert stored == {}
643    assert subscribers == []
644
645
646async def test_wait_for_engines_fails_when_only_the_tts_engine_is_missing(
647    monkeypatch: pytest.MonkeyPatch,
648) -> None:
649    """The reported error names the engine kind that is actually missing."""
650    provider, _, _ = _make_engine_provider([_make_engine_plugin("p1", ["ai"], [])])
651    monkeypatch.setattr(ai_radio_provider, "ENGINE_DISCOVERY_TIMEOUT", 0.05)
652
653    with pytest.raises(SetupFailedError) as error:
654        async with asyncio.timeout(1):
655            await provider._wait_for_engines()
656
657    assert error.value.translation_key == "ai_radio_no_tts_engine"
658
659
660async def test_wait_for_engines_resumes_when_a_supplier_loads_later() -> None:
661    """A plugin that finishes loading after AI Radio still satisfies the bounded wait."""
662    plugins: list[Any] = []
663    provider, subscribers, stored = _make_engine_provider(plugins)
664
665    task = asyncio.create_task(provider._wait_for_engines())
666    while not subscribers:
667        await asyncio.sleep(0)
668    plugins.append(_make_engine_plugin("p1", ["ai"], ["tts"]))
669    for callback in list(subscribers):
670        callback(MagicMock())
671
672    await task
673
674    assert stored == {CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
675    assert subscribers == []
676
677
678async def test_wait_for_engines_rejects_a_configured_engine_that_disappeared(
679    monkeypatch: pytest.MonkeyPatch,
680) -> None:
681    """A concrete selection that no longer exists is never substituted by another engine."""
682    provider, _, stored = _make_engine_provider(
683        [_make_engine_plugin("p1", ["ai"], ["tts"])],
684        setup_values={CONF_AI_ENGINE: "p1/gone"},
685    )
686    monkeypatch.setattr(ai_radio_provider, "ENGINE_DISCOVERY_TIMEOUT", 0.05)
687
688    with pytest.raises(SetupFailedError) as error:
689        async with asyncio.timeout(1):
690            await provider._wait_for_engines()
691
692    assert error.value.translation_key == "ai_radio_no_ai_engine"
693    assert stored[CONF_AI_ENGINE] == "p1/gone"
694
695
696async def test_providers_updated_leaves_a_healthy_selection_alone() -> None:
697    """A providers change that does not affect the engines is a no-op."""
698    provider, _, _ = _make_engine_provider(
699        [_make_engine_plugin("p1", ["ai"], ["tts"])],
700        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
701    )
702    errors = _record_unload_with_error(provider)
703
704    await provider._on_providers_updated(MagicMock())
705
706    assert provider._engine_recheck_task is None
707    assert errors == []
708
709
710async def test_providers_updated_unloads_when_an_engine_stays_gone(
711    monkeypatch: pytest.MonkeyPatch,
712) -> None:
713    """An engine removed after the load surfaces as a provider error instead of at playtime."""
714    plugins: list[Any] = [_make_engine_plugin("p1", ["ai"], ["tts"])]
715    provider, _, _ = _make_engine_provider(
716        plugins,
717        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
718    )
719    errors = _record_unload_with_error(provider)
720    monkeypatch.setattr(ai_radio_provider, "ENGINE_RECHECK_GRACE", 0.05)
721    plugins.clear()
722
723    await provider._on_providers_updated(MagicMock())
724    assert provider._engine_recheck_task is not None
725    async with asyncio.timeout(1):
726        await provider._engine_recheck_task
727
728    assert [error.translation_key for error in errors] == ["ai_radio_no_ai_engine"]
729
730
731async def test_providers_updated_survives_a_supplier_reload() -> None:
732    """A plugin reload briefly takes its engines with it, which must not unload AI Radio."""
733    plugins: list[Any] = []
734    provider, subscribers, _ = _make_engine_provider(
735        plugins,
736        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
737    )
738    errors = _record_unload_with_error(provider)
739
740    await provider._on_providers_updated(MagicMock())
741    assert provider._engine_recheck_task is not None
742    assert subscribers
743    plugins.append(_make_engine_plugin("p1", ["ai"], ["tts"]))
744    for callback in list(subscribers):
745        callback(MagicMock())
746    await provider._engine_recheck_task
747
748    assert errors == []
749    assert subscribers == []
750
751
752async def test_providers_updated_ignored_while_closing() -> None:
753    """The watch does not act on a providers change while the server is closing."""
754    provider, _, _ = _make_engine_provider(
755        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
756    )
757    errors = _record_unload_with_error(provider)
758    cast("Any", provider.mass).closing = True
759
760    await provider._on_providers_updated(MagicMock())
761
762    assert provider._engine_recheck_task is None
763    assert errors == []
764
765
766async def test_providers_updated_ignored_while_unloading() -> None:
767    """A providers change landing during our own unload is not acted on."""
768    provider, _, _ = _make_engine_provider(
769        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
770    )
771    errors = _record_unload_with_error(provider)
772    provider._unloading = True
773
774    await provider._on_providers_updated(MagicMock())
775
776    assert provider._engine_recheck_task is None
777    assert errors == []
778
779
780async def test_providers_updated_keeps_a_single_recheck_in_flight() -> None:
781    """Providers changes arriving during the grace period do not stack up rechecks."""
782    plugins: list[Any] = []
783    provider, subscribers, _ = _make_engine_provider(
784        plugins,
785        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
786    )
787    _record_unload_with_error(provider)
788
789    await provider._on_providers_updated(MagicMock())
790    first_task = provider._engine_recheck_task
791    assert first_task is not None
792    assert subscribers
793    await provider._on_providers_updated(MagicMock())
794
795    assert provider._engine_recheck_task is first_task
796    assert cast("Any", provider.mass).create_task.call_count == 1
797    plugins.append(_make_engine_plugin("p1", ["ai"], ["tts"]))
798    for callback in list(subscribers):
799        callback(MagicMock())
800    await first_task
801
802
803async def test_loaded_in_mass_watches_the_loaded_providers() -> None:
804    """Loading the provider wires up the engine watch, and unloading tears it down."""
805    provider, subscribers, _ = _make_engine_provider(
806        [_make_engine_plugin("p1", ["ai"], ["tts"])],
807        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
808    )
809
810    await provider.loaded_in_mass()
811
812    subscribe_calls = cast("Any", provider.mass).subscribe.call_args_list
813    assert subscribe_calls[0].args == (
814        provider._on_providers_updated,
815        EventType.PROVIDERS_UPDATED,
816    )
817    assert subscribe_calls[1].args == (
818        provider._on_dj_queue_event,
819        (EventType.QUEUE_ADDED, EventType.QUEUE_ITEMS_UPDATED, EventType.PLAYER_REMOVED),
820    )
821    assert subscribers == [provider._on_providers_updated, provider._on_dj_queue_event]
822
823    await provider.unload()
824
825    assert subscribers == []
826
827
828async def test_queue_dj_commands_are_registered_as_queue_control() -> None:
829    """Reading and arming the queue DJ menu are both queue-scoped, not provider config."""
830    provider, _, _ = _make_engine_provider(
831        [_make_engine_plugin("p1", ["ai"], ["tts"])],
832        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
833    )
834
835    await provider.loaded_in_mass()
836
837    scopes = {
838        call.args[0]: call.kwargs["required_scope"]
839        for call in cast("Any", provider.mass).register_api_command.call_args_list
840    }
841    assert scopes["ai_radio/queue_dj/set"] == Scope.QUEUES_CONTROL
842    assert scopes["ai_radio/queue_dj/status"] == Scope.QUEUES_CONTROL
843    assert scopes["ai_radio/status"] == Scope.CONFIG_PROVIDERS_READ
844
845
846async def test_unload_cancels_an_in_flight_engine_recheck() -> None:
847    """Unloading stops a running grace period instead of letting it report an error."""
848    provider, subscribers, _ = _make_engine_provider(
849        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
850    )
851    errors = _record_unload_with_error(provider)
852
853    await provider._on_providers_updated(MagicMock())
854    recheck_task = provider._engine_recheck_task
855    assert recheck_task is not None
856    assert subscribers
857    await provider.unload()
858
859    with pytest.raises(asyncio.CancelledError):
860        await recheck_task
861    assert provider._unloading is True
862    assert errors == []
863
864
865async def test_unload_cancels_an_in_flight_queue_dj_replan() -> None:
866    """Unloading stops replan work that is still running for an armed queue."""
867    provider, _, _ = _make_engine_provider([])
868
869    async def _never_returns() -> None:
870        await asyncio.sleep(3600)
871
872    replan_task = asyncio.ensure_future(_never_returns())
873    provider._dj_queues["queue-1"] = DJQueueState(
874        queue_id="queue-1",
875        host_id="rick",
876        dj_session_id="dj0123456789",
877        task=replan_task,
878    )
879
880    await provider.unload()
881
882    with pytest.raises(asyncio.CancelledError):
883        await asyncio.wait_for(replan_task, timeout=5)
884
885
886async def test_engine_recheck_stays_silent_when_the_provider_unloads_during_the_wait(
887    monkeypatch: pytest.MonkeyPatch,
888) -> None:
889    """An unload surfacing as the wait's timeout is not reported as an engine error."""
890    provider, _, _ = _make_engine_provider(
891        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
892    )
893    errors = _record_unload_with_error(provider)
894    monkeypatch.setattr(ai_radio_provider, "ENGINE_RECHECK_GRACE", 0.05)
895
896    await provider._on_providers_updated(MagicMock())
897    recheck_task = provider._engine_recheck_task
898    assert recheck_task is not None
899    provider._unloading = True
900    await recheck_task
901
902    assert errors == []
903    assert cast("Any", provider.mass).call_later.call_count == 0
904
905
906async def test_engine_recheck_stays_silent_when_the_server_closes_during_the_wait(
907    monkeypatch: pytest.MonkeyPatch,
908) -> None:
909    """A shutdown surfacing as the wait's timeout is not reported as an engine error."""
910    provider, _, _ = _make_engine_provider(
911        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
912    )
913    errors = _record_unload_with_error(provider)
914    monkeypatch.setattr(ai_radio_provider, "ENGINE_RECHECK_GRACE", 0.05)
915
916    await provider._on_providers_updated(MagicMock())
917    recheck_task = provider._engine_recheck_task
918    assert recheck_task is not None
919    cast("Any", provider.mass).closing = True
920    await recheck_task
921
922    assert errors == []
923    assert cast("Any", provider.mass).call_later.call_count == 0
924
925
926async def test_engine_watchdog_arms_a_reload_after_unloading(
927    monkeypatch: pytest.MonkeyPatch,
928) -> None:
929    """The unload arms the reload that picks the provider back up once engines return."""
930    provider, _, _ = _make_engine_provider(
931        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
932    )
933    errors = _record_unload_with_error(provider)
934    monkeypatch.setattr(ai_radio_provider, "ENGINE_RECHECK_GRACE", 0.05)
935
936    await provider._on_providers_updated(MagicMock())
937    recheck_task = provider._engine_recheck_task
938    assert recheck_task is not None
939    # the watch waits out the grace, not the (much shorter) discovery timeout of the load
940    async with asyncio.timeout(1):
941        await recheck_task
942
943    assert [error.translation_key for error in errors] == ["ai_radio_no_ai_engine"]
944    retry = cast("Any", provider.mass).call_later.call_args
945    assert retry.args == (ENGINE_RETRY_DELAY, provider.mass.load_provider, "ai_radio")
946    assert retry.kwargs == {"allow_retry": True, "task_id": "load_provider_ai_radio"}
947