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_save_section_leaves_the_stations_file_alone(provider: Any, tmp_path: Path) -> None:
353    """Sections no longer live inside stations, so saving one must not rewrite them."""
354    provider._sections_file = tmp_path / "sections.json"
355    provider._stations_file = tmp_path / "stations.json"
356    provider._stations_file.write_text("untouched")
357
358    await provider.save_section(
359        {"id": "New_Section", "name": "New Section", "type": "ai_text", "prompt": "Say something"}
360    )
361
362    assert "New_Section" in provider._sections
363    assert provider._stations_file.read_text() == "untouched"
364
365
366async def test_delete_host_refuses_when_station_references_it(provider: Any) -> None:
367    """Refuse to delete a host that a station still references."""
368    saved = await provider.save_host(await provider.host_template())
369    provider._stations["station_a"] = {
370        "id": "station_a",
371        "name": "Station A",
372        "source_playlist_id": "p1",
373        "source_playlist_provider": "library",
374        "default_player_id": "",
375        "max_duration_minutes": 0.0,
376        "shuffle_source_tracks": True,
377        "host_id": saved["id"],
378    }
379    with pytest.raises(InvalidDataError):
380        await provider.delete_host(saved["id"])
381
382
383@pytest.mark.asyncio
384async def test_delete_host_refuses_when_it_is_an_active_queue_dj(provider: Any) -> None:
385    """Refuse to delete a host that is the active DJ on a queue."""
386    saved = await provider.save_host(await provider.host_template())
387    provider._dj_queues["queue-1"] = DJQueueState(
388        queue_id="queue-1",
389        host_id=saved["id"],
390        dj_session_id="dj0123456789",
391    )
392
393    with pytest.raises(InvalidDataError, match="is the active DJ on queues: queue-1"):
394        await provider.delete_host(saved["id"])
395
396
397@pytest.mark.asyncio
398async def test_concurrent_start_run_calls_respect_the_run_limit() -> None:
399    """
400    Concurrent start_run calls must not both get past the concurrency guards.
401
402    The guards and the session insert are one critical section; without it an await
403    introduced between them would let both callers observe zero running sessions.
404    """
405    player = SimpleNamespace(player_id="living_room", available=True, enabled=True)
406    provider = _make_provider()
407    provider.logger = logging.getLogger("tests.ai_radio.provider")
408    provider._stations = {
409        "station_a": {
410            "id": "station_a",
411            "name": "Station A",
412            "default_player_id": "living_room",
413            "host_id": "host_a",
414        },
415        "station_b": {
416            "id": "station_b",
417            "name": "Station B",
418            "default_player_id": "living_room",
419            "host_id": "host_a",
420        },
421    }
422    provider._hosts = {"host_a": {"id": "host_a", "name": "Host A"}}
423    provider._sections = {}
424    provider.mass = cast(
425        "Any",
426        SimpleNamespace(
427            players=SimpleNamespace(get_player=lambda _player_id: player),
428            create_task=lambda coro, **_kw: _close(coro),
429        ),
430    )
431
432    results = await asyncio.gather(
433        provider.start_run(station_id="station_a"),
434        provider.start_run(station_id="station_b"),
435        return_exceptions=True,
436    )
437
438    started = [item for item in results if isinstance(item, dict)]
439    rejected = [item for item in results if isinstance(item, InvalidDataError)]
440    assert len(started) == 1
441    assert len(rejected) == 1
442    assert "Max concurrent runs reached" in str(rejected[0])
443
444
445@pytest.mark.asyncio
446async def test_stop_run_stops_the_queue_it_owns() -> None:
447    """Stop playback on the target queue when the show is stopped from the UI."""
448    provider = _make_provider()
449    provider.logger = cast(
450        "Any",
451        SimpleNamespace(debug=lambda *_a, **_kw: None, info=lambda *_a, **_kw: None),
452    )
453    stopped: list[str] = []
454    provider.mass = cast(
455        "Any",
456        SimpleNamespace(
457            player_queues=SimpleNamespace(
458                get=lambda _queue_id: SimpleNamespace(state=PlaybackState.PLAYING, current_index=3),
459                stop=_recording_stop(stopped),
460            )
461        ),
462    )
463    provider._sessions["s_run"] = SessionState(
464        session_id="s_run",
465        station_id="st",
466        status="running",
467        queue_id="living_room",
468    )
469
470    result = await provider.stop_run(session_id="s_run")
471
472    assert result["status"] == "stopped"
473    assert stopped == ["living_room"]
474
475
476@pytest.mark.asyncio
477async def test_stop_run_survives_an_unavailable_player() -> None:
478    """Still mark the session stopped when the target player has gone away."""
479    provider = _make_provider()
480    provider.logger = cast(
481        "Any",
482        SimpleNamespace(debug=lambda *_a, **_kw: None, info=lambda *_a, **_kw: None),
483    )
484
485    async def _raise(queue_id: str) -> None:
486        raise PlayerUnavailableError(f"Player {queue_id} is not available")
487
488    provider.mass = cast(
489        "Any",
490        SimpleNamespace(
491            player_queues=SimpleNamespace(
492                get=lambda _queue_id: SimpleNamespace(state=PlaybackState.PLAYING, current_index=3),
493                stop=_raise,
494            )
495        ),
496    )
497    provider._sessions["s_run"] = SessionState(
498        session_id="s_run",
499        station_id="st",
500        status="running",
501        queue_id="living_room",
502    )
503
504    result = await provider.stop_run(session_id="s_run")
505
506    assert result["status"] == "stopped"
507
508
509def _make_engine_provider(
510    plugins: list[Any],
511    setup_values: dict[str, Any] | None = None,
512) -> tuple[AIRadioProvider, list[Any], dict[str, Any]]:
513    """
514    Create a provider whose mass serves the given plugins.
515
516    :return: The provider, the list its event subscribers land in, and the (plain text)
517        stand-in for its stored setup_data.
518    """
519    provider = _make_provider()
520    provider.config = cast("Any", SimpleNamespace(instance_id="ai_radio", setup_data={}))
521    subscribers: list[Any] = []
522    stored = dict(setup_values or {})
523
524    def _subscribe(callback: Any, *_args: Any, **_kwargs: Any) -> Any:
525        subscribers.append(callback)
526        return lambda: subscribers.remove(callback)
527
528    def _providers(feature: ProviderFeature, **_kwargs: Any) -> list[Any]:
529        attribute = "get_ai_engines" if feature == ProviderFeature.AI_QUERY else "get_tts_engines"
530        return [plugin for plugin in plugins if getattr(plugin, attribute).return_value]
531
532    mass = MagicMock()
533    mass.closing = False
534    mass.subscribe.side_effect = _subscribe
535    # mirror mass.create_task, which runs the coroutine up to its first suspension
536    # before handing back the task
537    mass.create_task.side_effect = lambda coro, **_kwargs: asyncio.Task(
538        coro, loop=asyncio.get_running_loop(), eager_start=True
539    )
540    mass.get_providers_supporting_feature.side_effect = _providers
541    mass.config.get_provider_setup_value.side_effect = lambda _instance_id, key, default=None: (
542        stored.get(key, default)
543    )
544    provider.mass = cast("Any", mass)
545    provider.logger = logging.getLogger("test.ai_radio")
546    provider._dj_queues = {}
547    provider._unloading = False
548    provider._engine_recheck_task = None
549    provider._unregister_handles = []
550    provider._update_setup_data = cast(  # type: ignore[method-assign]
551        "Any", lambda key, value, **_kwargs: stored.__setitem__(key, value)
552    )
553    return provider, subscribers, stored
554
555
556def _record_unload_with_error(provider: AIRadioProvider) -> list[Any]:
557    """Capture the errors the provider unloads itself with instead of really unloading."""
558    errors: list[Any] = []
559    provider.unload_with_error = cast(  # type: ignore[method-assign]
560        "Any", errors.append
561    )
562    return errors
563
564
565def _make_engine_plugin(instance_id: str, ai_ids: list[str], tts_ids: list[str]) -> MagicMock:
566    """Create a mock plugin provider exposing the given AI and TTS engines."""
567    plugin = MagicMock(spec=PluginProvider)
568    plugin.instance_id = instance_id
569    plugin.get_ai_engines = AsyncMock(
570        return_value=[AIEngine(id=engine, name=engine, provider=plugin) for engine in ai_ids]
571    )
572    plugin.get_tts_engines = AsyncMock(
573        return_value=[TTSEngine(id=engine, name=engine, provider=plugin) for engine in tts_ids]
574    )
575    return plugin
576
577
578async def test_wait_for_engines_seeds_a_concrete_selection() -> None:
579    """An instance without a stored selection adopts concrete engines and waits for nothing."""
580    provider, subscribers, stored = _make_engine_provider(
581        [_make_engine_plugin("p1", ["ai"], ["tts"])]
582    )
583
584    await provider._wait_for_engines()
585
586    assert stored == {CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
587    assert subscribers == []
588
589
590async def test_wait_for_engines_keeps_an_existing_selection() -> None:
591    """A stored selection survives the load instead of being reseeded to the first engine."""
592    provider, _, stored = _make_engine_provider(
593        [_make_engine_plugin("p1", ["ai", "other"], ["tts"])],
594        setup_values={CONF_AI_ENGINE: "p1/other"},
595    )
596
597    await provider._wait_for_engines()
598
599    assert stored[CONF_AI_ENGINE] == "p1/other"
600
601
602async def test_wait_for_engines_fails_the_load_when_no_engine_appears(
603    monkeypatch: pytest.MonkeyPatch,
604) -> None:
605    """A provider without engines refuses to load so the UI offers a reconfigure."""
606    provider, subscribers, stored = _make_engine_provider([])
607    monkeypatch.setattr(ai_radio_provider, "ENGINE_DISCOVERY_TIMEOUT", 0.05)
608
609    with pytest.raises(SetupFailedError) as error:
610        async with asyncio.timeout(1):
611            await provider._wait_for_engines()
612
613    assert error.value.translation_key == "ai_radio_no_ai_engine"
614    assert stored == {}
615    assert subscribers == []
616
617
618async def test_wait_for_engines_fails_when_only_the_tts_engine_is_missing(
619    monkeypatch: pytest.MonkeyPatch,
620) -> None:
621    """The reported error names the engine kind that is actually missing."""
622    provider, _, _ = _make_engine_provider([_make_engine_plugin("p1", ["ai"], [])])
623    monkeypatch.setattr(ai_radio_provider, "ENGINE_DISCOVERY_TIMEOUT", 0.05)
624
625    with pytest.raises(SetupFailedError) as error:
626        async with asyncio.timeout(1):
627            await provider._wait_for_engines()
628
629    assert error.value.translation_key == "ai_radio_no_tts_engine"
630
631
632async def test_wait_for_engines_resumes_when_a_supplier_loads_later() -> None:
633    """A plugin that finishes loading after AI Radio still satisfies the bounded wait."""
634    plugins: list[Any] = []
635    provider, subscribers, stored = _make_engine_provider(plugins)
636
637    task = asyncio.create_task(provider._wait_for_engines())
638    while not subscribers:
639        await asyncio.sleep(0)
640    plugins.append(_make_engine_plugin("p1", ["ai"], ["tts"]))
641    for callback in list(subscribers):
642        callback(MagicMock())
643
644    await task
645
646    assert stored == {CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
647    assert subscribers == []
648
649
650async def test_wait_for_engines_rejects_a_configured_engine_that_disappeared(
651    monkeypatch: pytest.MonkeyPatch,
652) -> None:
653    """A concrete selection that no longer exists is never substituted by another engine."""
654    provider, _, stored = _make_engine_provider(
655        [_make_engine_plugin("p1", ["ai"], ["tts"])],
656        setup_values={CONF_AI_ENGINE: "p1/gone"},
657    )
658    monkeypatch.setattr(ai_radio_provider, "ENGINE_DISCOVERY_TIMEOUT", 0.05)
659
660    with pytest.raises(SetupFailedError) as error:
661        async with asyncio.timeout(1):
662            await provider._wait_for_engines()
663
664    assert error.value.translation_key == "ai_radio_no_ai_engine"
665    assert stored[CONF_AI_ENGINE] == "p1/gone"
666
667
668async def test_providers_updated_leaves_a_healthy_selection_alone() -> None:
669    """A providers change that does not affect the engines is a no-op."""
670    provider, _, _ = _make_engine_provider(
671        [_make_engine_plugin("p1", ["ai"], ["tts"])],
672        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
673    )
674    errors = _record_unload_with_error(provider)
675
676    await provider._on_providers_updated(MagicMock())
677
678    assert provider._engine_recheck_task is None
679    assert errors == []
680
681
682async def test_providers_updated_unloads_when_an_engine_stays_gone(
683    monkeypatch: pytest.MonkeyPatch,
684) -> None:
685    """An engine removed after the load surfaces as a provider error instead of at playtime."""
686    plugins: list[Any] = [_make_engine_plugin("p1", ["ai"], ["tts"])]
687    provider, _, _ = _make_engine_provider(
688        plugins,
689        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
690    )
691    errors = _record_unload_with_error(provider)
692    monkeypatch.setattr(ai_radio_provider, "ENGINE_RECHECK_GRACE", 0.05)
693    plugins.clear()
694
695    await provider._on_providers_updated(MagicMock())
696    assert provider._engine_recheck_task is not None
697    async with asyncio.timeout(1):
698        await provider._engine_recheck_task
699
700    assert [error.translation_key for error in errors] == ["ai_radio_no_ai_engine"]
701
702
703async def test_providers_updated_survives_a_supplier_reload() -> None:
704    """A plugin reload briefly takes its engines with it, which must not unload AI Radio."""
705    plugins: list[Any] = []
706    provider, subscribers, _ = _make_engine_provider(
707        plugins,
708        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
709    )
710    errors = _record_unload_with_error(provider)
711
712    await provider._on_providers_updated(MagicMock())
713    assert provider._engine_recheck_task is not None
714    assert subscribers
715    plugins.append(_make_engine_plugin("p1", ["ai"], ["tts"]))
716    for callback in list(subscribers):
717        callback(MagicMock())
718    await provider._engine_recheck_task
719
720    assert errors == []
721    assert subscribers == []
722
723
724async def test_providers_updated_ignored_while_closing() -> None:
725    """The watch does not act on a providers change while the server is closing."""
726    provider, _, _ = _make_engine_provider(
727        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
728    )
729    errors = _record_unload_with_error(provider)
730    cast("Any", provider.mass).closing = True
731
732    await provider._on_providers_updated(MagicMock())
733
734    assert provider._engine_recheck_task is None
735    assert errors == []
736
737
738async def test_providers_updated_ignored_while_unloading() -> None:
739    """A providers change landing during our own unload is not acted on."""
740    provider, _, _ = _make_engine_provider(
741        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
742    )
743    errors = _record_unload_with_error(provider)
744    provider._unloading = True
745
746    await provider._on_providers_updated(MagicMock())
747
748    assert provider._engine_recheck_task is None
749    assert errors == []
750
751
752async def test_providers_updated_keeps_a_single_recheck_in_flight() -> None:
753    """Providers changes arriving during the grace period do not stack up rechecks."""
754    plugins: list[Any] = []
755    provider, subscribers, _ = _make_engine_provider(
756        plugins,
757        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
758    )
759    _record_unload_with_error(provider)
760
761    await provider._on_providers_updated(MagicMock())
762    first_task = provider._engine_recheck_task
763    assert first_task is not None
764    assert subscribers
765    await provider._on_providers_updated(MagicMock())
766
767    assert provider._engine_recheck_task is first_task
768    assert cast("Any", provider.mass).create_task.call_count == 1
769    plugins.append(_make_engine_plugin("p1", ["ai"], ["tts"]))
770    for callback in list(subscribers):
771        callback(MagicMock())
772    await first_task
773
774
775async def test_loaded_in_mass_watches_the_loaded_providers() -> None:
776    """Loading the provider wires up the engine watch, and unloading tears it down."""
777    provider, subscribers, _ = _make_engine_provider(
778        [_make_engine_plugin("p1", ["ai"], ["tts"])],
779        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
780    )
781
782    await provider.loaded_in_mass()
783
784    subscribe_calls = cast("Any", provider.mass).subscribe.call_args_list
785    assert subscribe_calls[0].args == (
786        provider._on_providers_updated,
787        EventType.PROVIDERS_UPDATED,
788    )
789    assert subscribe_calls[1].args == (
790        provider._on_dj_queue_event,
791        (EventType.QUEUE_ADDED, EventType.QUEUE_ITEMS_UPDATED, EventType.PLAYER_REMOVED),
792    )
793    assert subscribers == [provider._on_providers_updated, provider._on_dj_queue_event]
794
795    await provider.unload()
796
797    assert subscribers == []
798
799
800async def test_queue_dj_commands_are_registered_as_queue_control() -> None:
801    """Reading and arming the queue DJ menu are both queue-scoped, not provider config."""
802    provider, _, _ = _make_engine_provider(
803        [_make_engine_plugin("p1", ["ai"], ["tts"])],
804        setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"},
805    )
806
807    await provider.loaded_in_mass()
808
809    scopes = {
810        call.args[0]: call.kwargs["required_scope"]
811        for call in cast("Any", provider.mass).register_api_command.call_args_list
812    }
813    assert scopes["ai_radio/queue_dj/set"] == Scope.QUEUES_CONTROL
814    assert scopes["ai_radio/queue_dj/status"] == Scope.QUEUES_CONTROL
815    assert scopes["ai_radio/status"] == Scope.CONFIG_PROVIDERS_READ
816
817
818async def test_unload_cancels_an_in_flight_engine_recheck() -> None:
819    """Unloading stops a running grace period instead of letting it report an error."""
820    provider, subscribers, _ = _make_engine_provider(
821        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
822    )
823    errors = _record_unload_with_error(provider)
824
825    await provider._on_providers_updated(MagicMock())
826    recheck_task = provider._engine_recheck_task
827    assert recheck_task is not None
828    assert subscribers
829    await provider.unload()
830
831    with pytest.raises(asyncio.CancelledError):
832        await recheck_task
833    assert provider._unloading is True
834    assert errors == []
835
836
837async def test_unload_cancels_an_in_flight_queue_dj_replan() -> None:
838    """Unloading stops replan work that is still running for an armed queue."""
839    provider, _, _ = _make_engine_provider([])
840
841    async def _never_returns() -> None:
842        await asyncio.sleep(3600)
843
844    replan_task = asyncio.ensure_future(_never_returns())
845    provider._dj_queues["queue-1"] = DJQueueState(
846        queue_id="queue-1",
847        host_id="rick",
848        dj_session_id="dj0123456789",
849        task=replan_task,
850    )
851
852    await provider.unload()
853
854    with pytest.raises(asyncio.CancelledError):
855        await asyncio.wait_for(replan_task, timeout=5)
856
857
858async def test_engine_recheck_stays_silent_when_the_provider_unloads_during_the_wait(
859    monkeypatch: pytest.MonkeyPatch,
860) -> None:
861    """An unload surfacing as the wait's timeout is not reported as an engine error."""
862    provider, _, _ = _make_engine_provider(
863        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
864    )
865    errors = _record_unload_with_error(provider)
866    monkeypatch.setattr(ai_radio_provider, "ENGINE_RECHECK_GRACE", 0.05)
867
868    await provider._on_providers_updated(MagicMock())
869    recheck_task = provider._engine_recheck_task
870    assert recheck_task is not None
871    provider._unloading = True
872    await recheck_task
873
874    assert errors == []
875    assert cast("Any", provider.mass).call_later.call_count == 0
876
877
878async def test_engine_recheck_stays_silent_when_the_server_closes_during_the_wait(
879    monkeypatch: pytest.MonkeyPatch,
880) -> None:
881    """A shutdown surfacing as the wait's timeout is not reported as an engine error."""
882    provider, _, _ = _make_engine_provider(
883        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
884    )
885    errors = _record_unload_with_error(provider)
886    monkeypatch.setattr(ai_radio_provider, "ENGINE_RECHECK_GRACE", 0.05)
887
888    await provider._on_providers_updated(MagicMock())
889    recheck_task = provider._engine_recheck_task
890    assert recheck_task is not None
891    cast("Any", provider.mass).closing = True
892    await recheck_task
893
894    assert errors == []
895    assert cast("Any", provider.mass).call_later.call_count == 0
896
897
898async def test_engine_watchdog_arms_a_reload_after_unloading(
899    monkeypatch: pytest.MonkeyPatch,
900) -> None:
901    """The unload arms the reload that picks the provider back up once engines return."""
902    provider, _, _ = _make_engine_provider(
903        [], setup_values={CONF_AI_ENGINE: "p1/ai", CONF_TTS_ENGINE: "p1/tts"}
904    )
905    errors = _record_unload_with_error(provider)
906    monkeypatch.setattr(ai_radio_provider, "ENGINE_RECHECK_GRACE", 0.05)
907
908    await provider._on_providers_updated(MagicMock())
909    recheck_task = provider._engine_recheck_task
910    assert recheck_task is not None
911    # the watch waits out the grace, not the (much shorter) discovery timeout of the load
912    async with asyncio.timeout(1):
913        await recheck_task
914
915    assert [error.translation_key for error in errors] == ["ai_radio_no_ai_engine"]
916    retry = cast("Any", provider.mass).call_later.call_args
917    assert retry.args == (ENGINE_RETRY_DELAY, provider.mass.load_provider, "ai_radio")
918    assert retry.kwargs == {"allow_retry": True, "task_id": "load_provider_ai_radio"}
919