/
/
/
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