music-assistant-server

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