music-assistant-server

1"""Unit tests for AI Radio runtime session flow and logging."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7import logging
8import random
9from collections.abc import Awaitable, Callable
10from contextlib import suppress
11from copy import deepcopy
12from pathlib import Path
13from types import SimpleNamespace
14from typing import Any, cast
15from unittest.mock import AsyncMock, MagicMock
16
17import pytest
18from music_assistant_models.enums import (
19    EventType,
20    MediaType,
21    PlaybackState,
22    ProviderFeature,
23    ProviderType,
24)
25from music_assistant_models.errors import MusicAssistantError
26from music_assistant_models.event import MassEvent
27from music_assistant_models.media_items import ProviderMapping, Track
28
29from music_assistant.helpers.datetime import now as host_now
30from music_assistant.models.plugin import AIEngine, PluginProvider, TTSEngine
31from music_assistant.providers.ai_radio import runtime as runtime_module
32from music_assistant.providers.ai_radio.constants import (
33    ATTR_HOST_ID,
34    ATTR_MAX_CHARS,
35    ATTR_PROMPT,
36    ATTR_SESSION_ID,
37    ATTR_STATION_ID,
38    ATTR_WEB_SEARCH_MODE,
39    CONF_AI_ENGINE,
40    CONF_TTS_ENGINE,
41    CONF_WEATHER_PROVIDER,
42    TTS_PRONUNCIATION_INSTRUCTIONS,
43)
44from music_assistant.providers.ai_radio.models import PlannedSection, SessionState, Slot
45from music_assistant.providers.ai_radio.queue_dj import AIRadioQueueDJMixin
46from music_assistant.providers.ai_radio.runtime import AIRadioRuntimeMixin
47from music_assistant.providers.ai_radio.storage import AIRadioStorageMixin
48
49
50class StubConfig:
51    """Minimal ProviderConfig stand-in exposing get_value."""
52
53    def __init__(self, values: dict[str, Any] | None = None) -> None:
54        """Initialize with an optional map of config key -> value."""
55        self._values = values or {}
56
57    def get_value(self, key: str, default: Any = None) -> Any:
58        """Return the stubbed value for key, or default when absent."""
59        return self._values.get(key, default)
60
61
62class DummyRuntime(AIRadioRuntimeMixin):
63    """Minimal runtime harness for testing mixin behavior."""
64
65    def __init__(self, setup_values: dict[str, Any] | None = None) -> None:
66        """Initialize minimal state for runtime tests."""
67        self.logger = logging.getLogger("tests.ai_radio.runtime")
68        self._sessions: dict[str, SessionState] = {}
69        self._sections: dict[str, dict[str, Any]] = {}
70        self.config = cast("Any", StubConfig())
71        self.instance_id = "ai_radio_test"
72        self.domain = "ai_radio"
73        self._setup_values = setup_values or {}
74
75    def get_setup_value(self, key: str, default: Any = None) -> Any:
76        """Return the stubbed setup flow value for key, or default when absent."""
77        return self._setup_values.get(key, default)
78
79    def _schedule_replan(self, queue_id: str) -> None:
80        """No-op stand-in for the queue DJ mixin's replan scheduling."""
81
82    async def set_queue_dj(self, queue_id: str, host_id: str | None) -> dict[str, str]:
83        """No-op stand-in for the queue DJ mixin's set_queue_dj."""
84        return {}
85
86    def _materialize_sections(
87        self, section_ids: list[str], sections_map: dict[str, dict[str, Any]] | None = None
88    ) -> tuple[list[dict[str, Any]], list[str]]:
89        """Resolve section ids against self._sections, mirroring the storage mixin."""
90        source = self._sections if sections_map is None else sections_map
91        sections: list[dict[str, Any]] = []
92        missing: list[str] = []
93        for section_id in section_ids:
94            section = source.get(section_id)
95            if section is None:
96                missing.append(section_id)
97                continue
98            sections.append(deepcopy(section))
99        return sections, missing
100
101
102class FailingRuntime(DummyRuntime):
103    """Runtime harness that forces show execution failure."""
104
105    async def _run_show(
106        self,
107        session: SessionState,
108        station: dict[str, Any],
109    ) -> dict[str, Any]:
110        """Raise to test failed-session behavior."""
111        raise RuntimeError("boom")
112
113
114def _set_runtime_mass(runtime: AIRadioRuntimeMixin, mass: Any) -> None:
115    """Attach lightweight test mass object while bypassing strict runtime typing."""
116    cast("Any", runtime).mass = mass
117
118
119def _create_ai_plugin(instance_id: str, *engine_ids: str) -> MagicMock:
120    """Create a mock plugin provider exposing the given AI engines."""
121    provider = MagicMock(spec=PluginProvider)
122    provider.instance_id = instance_id
123    provider.get_ai_engines = AsyncMock(
124        return_value=[
125            AIEngine(id=engine_id, name=engine_id, provider=provider) for engine_id in engine_ids
126        ]
127    )
128    return provider
129
130
131def _create_tts_plugin(instance_id: str, *engine_ids: str) -> MagicMock:
132    """Create a mock plugin provider exposing the given TTS engines."""
133    provider = MagicMock(spec=PluginProvider)
134    provider.instance_id = instance_id
135    provider.get_tts_engines = AsyncMock(
136        return_value=[
137            TTSEngine(id=engine_id, name=engine_id, provider=provider) for engine_id in engine_ids
138        ]
139    )
140    return provider
141
142
143def _create_engine_mass(feature: ProviderFeature, *providers: Any, **attrs: Any) -> Any:
144    """Create a lightweight mass stand-in serving the given plugins for one feature."""
145
146    class DummyMass:
147        def get_providers_supporting_feature(
148            self,
149            requested: ProviderFeature,
150            priority: tuple[ProviderType, ...] = (),
151        ) -> list[Any]:
152            return list(providers) if requested == feature else []
153
154    mass = DummyMass()
155    for key, value in attrs.items():
156        setattr(mass, key, value)
157    return mass
158
159
160async def test_run_session_sets_completed_and_logs(caplog: Any) -> None:
161    """Complete a session and emit start/completion logs."""
162
163    class SuccessfulRuntime(DummyRuntime):
164        async def _run_show(
165            self,
166            session: SessionState,
167            station: dict[str, Any],
168        ) -> dict[str, Any]:
169            """Return a successful show run result."""
170            return {"ok": True}
171
172    runtime = SuccessfulRuntime()
173    session = SessionState(session_id="s1", station_id="station_a")
174    runtime._sessions[session.session_id] = session
175
176    with caplog.at_level(logging.INFO):
177        await runtime._run_session(session.session_id, {"id": "station_a"})
178
179    assert session.status == "completed"
180    assert session.result == {"ok": True}
181    assert any("AI Radio run started" in message for message in caplog.messages)
182    assert any("AI Radio run completed" in message for message in caplog.messages)
183
184
185async def test_run_session_sets_failed_state(caplog: Any) -> None:
186    """Fail a session and keep the error message in state."""
187    runtime = FailingRuntime()
188    session = SessionState(session_id="s2", station_id="station_b")
189    runtime._sessions[session.session_id] = session
190
191    with caplog.at_level(logging.ERROR):
192        await runtime._run_session(session.session_id, {"id": "station_b"})
193
194    assert session.status == "failed"
195    assert session.error == "boom"
196
197
198async def test_run_session_sets_failed_state_with_empty_exception_message() -> None:
199    """Store exception class name when failure has no message."""
200
201    class EmptyError(Exception):
202        """Exception with empty default message."""
203
204    class EmptyFailingRuntime(DummyRuntime):
205        async def _run_show(
206            self,
207            session: SessionState,
208            station: dict[str, Any],
209        ) -> dict[str, Any]:
210            raise EmptyError
211
212    runtime = EmptyFailingRuntime()
213    session = SessionState(session_id="s2b", station_id="station_b")
214    runtime._sessions[session.session_id] = session
215
216    await runtime._run_session(session.session_id, {"id": "station_b"})
217
218    assert session.status == "failed"
219    assert session.error == "EmptyError"
220
221
222async def test_run_session_sets_stopped_state_on_cancellation(caplog: Any) -> None:
223    """Mark session as stopped when runtime execution is cancelled."""
224
225    class CancelledRuntime(DummyRuntime):
226        async def _run_show(
227            self,
228            session: SessionState,
229            station: dict[str, Any],
230        ) -> dict[str, Any]:
231            raise asyncio.CancelledError
232
233    runtime = CancelledRuntime()
234    session = SessionState(session_id="s3", station_id="station_c")
235    runtime._sessions[session.session_id] = session
236
237    with caplog.at_level(logging.INFO), suppress(asyncio.CancelledError):
238        await runtime._run_session(session.session_id, {"id": "station_c"})
239
240    assert session.status == "stopped"
241    assert any("AI Radio run cancelled" in message for message in caplog.messages)
242
243
244async def test_run_session_reschedules_a_replan_for_the_session_queue() -> None:
245    """Re-arm the queue DJ for the session's queue once its show session ends."""
246
247    class ReplanTrackingRuntime(FailingRuntime):
248        def __init__(self) -> None:
249            super().__init__()
250            self.replanned_queue_ids: list[str] = []
251
252        def _schedule_replan(self, queue_id: str) -> None:
253            self.replanned_queue_ids.append(queue_id)
254
255    runtime = ReplanTrackingRuntime()
256    session = SessionState(session_id="s4", station_id="station_d")
257    session.queue_id = "queue_1"
258    runtime._sessions[session.session_id] = session
259
260    await runtime._run_session(session.session_id, {"id": "station_d"})
261
262    assert runtime.replanned_queue_ids == ["queue_1"]
263
264
265async def test_prepare_runtime_tokens_logs_unsupported_weather_provider(caplog: Any) -> None:
266    """Warn when weather placeholders are used with unsupported provider."""
267    runtime = DummyRuntime()
268    station = {
269        "sections": [
270            {
271                "id": "Weather_Short",
272                "type": "ai_text",
273                "prompt": "Forecast: <weather_hourly>",
274            }
275        ],
276        "section_order": [],
277    }
278    runtime.config = cast(
279        "Any",
280        StubConfig(
281            {
282                "weather_city": "Berlin",
283                "weather_country": "DE",
284                CONF_WEATHER_PROVIDER: "unsupported_provider",
285            }
286        ),
287    )
288
289    with caplog.at_level(logging.WARNING):
290        tokens = await runtime._prepare_runtime_tokens(station)
291
292    assert tokens == {}
293    assert any("Unsupported weather provider" in message for message in caplog.messages)
294
295
296def _weather_program() -> dict[str, Any]:
297    """Return a program whose only section references the hourly weather token."""
298    return {
299        "sections": [
300            {
301                "id": "Weather_Short",
302                "type": "ai_text",
303                "prompt": "Forecast: <weather_hourly>",
304            }
305        ],
306        "section_order": [],
307    }
308
309
310def _count_weather_fetches(runtime: DummyRuntime) -> list[str]:
311    """Replace the forecast lookup with a stub and return the list it records into."""
312    fetches: list[str] = []
313
314    async def _fetch(city: str, **_kwargs: Any) -> tuple[str, str]:
315        fetches.append(city)
316        return "12 degrees", "mild"
317
318    runtime._fetch_open_meteo_weather = _fetch  # type: ignore[method-assign, assignment]
319    return fetches
320
321
322async def test_prepare_runtime_tokens_reuses_the_cached_weather_within_the_ttl() -> None:
323    """A second pass inside the cache window reuses the tokens instead of refetching."""
324    runtime = DummyRuntime()
325    runtime.config = cast("Any", StubConfig({"weather_city": "Berlin", "weather_country": "DE"}))
326    fetches = _count_weather_fetches(runtime)
327
328    first = await runtime._prepare_runtime_tokens(_weather_program())
329    second = await runtime._prepare_runtime_tokens(_weather_program())
330
331    assert first == {"<weather_hourly>": "12 degrees", "<weather_daily>": "mild"}
332    assert second == first
333    assert fetches == ["Berlin"]
334
335
336async def test_prepare_runtime_tokens_refetches_the_weather_once_the_ttl_expired() -> None:
337    """An expired cache entry is refetched rather than served stale forever."""
338    runtime = DummyRuntime()
339    runtime.config = cast("Any", StubConfig({"weather_city": "Berlin", "weather_country": "DE"}))
340    fetches = _count_weather_fetches(runtime)
341
342    await runtime._prepare_runtime_tokens(_weather_program())
343    assert runtime._weather_tokens_cache is not None
344    fetched_at, tokens = runtime._weather_tokens_cache
345    runtime._weather_tokens_cache = (
346        fetched_at - runtime_module.WEATHER_TOKENS_CACHE_SECONDS - 1,
347        tokens,
348    )
349    await runtime._prepare_runtime_tokens(_weather_program())
350
351    assert fetches == ["Berlin", "Berlin"]
352
353
354def test_weather_strings_are_rounded_to_whole_numbers() -> None:
355    """A host reads the forecast out loud, so it says 19 degrees and never 19.2."""
356    runtime = DummyRuntime()
357    payload = {
358        "current": {
359            "time": "2026-08-10T09:00",
360            "temperature_2m": 19.2,
361            "apparent_temperature": 18.7,
362        },
363        "hourly": {
364            "time": ["2026-08-10T09:00", "2026-08-10T10:00"],
365            "temperature_2m": [19.2, 20.6],
366            "precipitation_probability": [12.4, 0],
367        },
368        "daily": {
369            "time": ["2026-08-10"],
370            "temperature_2m_min": [11.4],
371            "temperature_2m_max": [21.49],
372            "precipitation_probability_max": [30.6],
373        },
374    }
375
376    hourly, daily = runtime._format_weather_strings(payload)
377
378    assert hourly == (
379        "now 19C (feels 19C); 2026-08-10 09:00: 19C, rain 12%; 2026-08-10 10:00: 21C, rain 0%"
380    )
381    assert daily == "2026-08-10: 11-21C, rain 31%"
382
383
384async def test_prepare_runtime_tokens_ignores_missing_location(caplog: Any) -> None:
385    """Skip weather preparation when the configured location is incomplete."""
386    runtime = DummyRuntime()
387    station = {
388        "sections": [
389            {
390                "id": "Weather_Short",
391                "type": "ai_text",
392                "prompt": "Forecast: <weather_hourly>",
393            }
394        ],
395        "section_order": [],
396    }
397    runtime.config = cast("Any", StubConfig({"weather_city": "", "weather_country": "DE"}))
398
399    with caplog.at_level(logging.DEBUG):
400        tokens = await runtime._prepare_runtime_tokens(station)
401
402    assert tokens == {}
403    assert any("no location configured" in message for message in caplog.messages)
404
405
406def test_extract_location_reads_provider_config() -> None:
407    """Weather location comes from the provider config, not the station."""
408    runtime = DummyRuntime()
409    runtime.config = cast("Any", StubConfig({"weather_city": "Berlin", "weather_country": "DE"}))
410
411    assert runtime._extract_location() == ("Berlin", "DE")
412
413
414def test_extract_location_defaults_to_empty_when_unset() -> None:
415    """An unconfigured weather location resolves to empty strings, not an error."""
416    runtime = DummyRuntime()
417
418    assert runtime._extract_location() == ("", "")
419
420
421@pytest.mark.parametrize("timezone_value", ["Asia/Tokyo", "  Asia/Tokyo  "])
422def test_configured_now_uses_valid_configured_timezone(timezone_value: str) -> None:
423    """A valid configured IANA timezone name is honored, surrounding whitespace included."""
424    runtime = DummyRuntime()
425    runtime.config = cast("Any", StubConfig({"timezone": timezone_value}))
426
427    result = runtime._configured_now()
428
429    assert str(result.tzinfo) == "Asia/Tokyo"
430
431
432@pytest.mark.parametrize(
433    "timezone_value",
434    ["", "not-a-real-zone", "CEST", "../../etc/passwd"],
435)
436def test_configured_now_falls_back_when_timezone_blank_or_invalid(timezone_value: str) -> None:
437    """A blank or invalid configured timezone falls back to the host local time."""
438    runtime = DummyRuntime()
439    runtime.config = cast("Any", StubConfig({"timezone": timezone_value}))
440
441    result = runtime._configured_now()
442
443    assert result.utcoffset() == host_now().utcoffset()
444
445
446def test_plan_sections_ignores_invalid_optional_chance() -> None:
447    """Treat non-numeric OPTIONAL chance values as zero during planning."""
448    runtime = DummyRuntime()
449    station = {
450        "sections": [
451            {
452                "id": "Song_Transition",
453                "name": "Song Transition",
454                "type": "ai_text",
455                "prompt": "Transition from <prev_songinfo> to <next_songinfo>",
456            }
457        ],
458        "section_order": [
459            {
460                "when": "between_songs",
461                "flow": [
462                    {
463                        "OPTIONAL": {
464                            "section": "Song_Transition",
465                            "chance": "not-a-number",
466                        }
467                    }
468                ],
469            }
470        ],
471        "general": {"timezone": "UTC"},
472    }
473    tracks = [
474        {"name": "A", "artist": "Artist A", "songinfo": "Artist A - A", "duration": 180},
475        {"name": "B", "artist": "Artist B", "songinfo": "Artist B - B", "duration": 180},
476    ]
477
478    planned, _history = runtime._plan_sections(
479        session_id="sess",
480        tracks=tracks,
481        program=station,
482        track_index_offset=0,
483        minute_offset=0.0,
484        history_state={},
485        allowed_slot_when=["between_songs"],
486        runtime_tokens={},
487    )
488
489    assert planned == []
490
491
492async def test_generate_text_wraps_not_connected_error() -> None:
493    """Raise an actionable MusicAssistantError when the AI engine is disconnected."""
494
495    class NotConnected(Exception):
496        """Match hass_client NotConnected exception name."""
497
498    plugin = _create_ai_plugin("hass_1", "ai_task.default")
499    plugin.ai_query = AsyncMock(side_effect=NotConnected)
500    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
501    _set_runtime_mass(
502        runtime,
503        _create_engine_mass(
504            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="en_US")
505        ),
506    )
507
508    with pytest.raises(MusicAssistantError) as error:
509        await runtime._generate_text(
510            instructions="test",
511            prompt="test prompt",
512            web_mode="disabled",
513        )
514    assert "not connected" in str(error.value).lower()
515    assert "hass_1/ai_task.default" in str(error.value)
516
517
518async def test_generate_text_fails_the_section_when_the_engine_stalls(
519    monkeypatch: pytest.MonkeyPatch,
520) -> None:
521    """A stalled AI engine fails the section instead of hanging the session."""
522    monkeypatch.setattr("music_assistant.providers.ai_radio.runtime.AI_QUERY_TIMEOUT_SECONDS", 0.01)
523
524    async def _answers_too_late(*_args: Any, **_kwargs: Any) -> str:
525        await asyncio.sleep(5)
526        return "section text"
527
528    plugin = _create_ai_plugin("hass_1", "ai_task.default")
529    plugin.ai_query = AsyncMock(side_effect=_answers_too_late)
530    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
531    _set_runtime_mass(
532        runtime,
533        _create_engine_mass(
534            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="en_US")
535        ),
536    )
537
538    with pytest.raises(MusicAssistantError) as error:
539        await runtime._generate_text(
540            instructions="test",
541            prompt="test prompt",
542            web_mode="disabled",
543        )
544    assert "did not respond within" in str(error.value)
545
546
547async def test_generate_text_reports_an_engine_side_timeout_as_a_query_failure() -> None:
548    """A timeout raised by the engine itself is reported as a query failure, not our cap."""
549    plugin = _create_ai_plugin("hass_1", "ai_task.default")
550    plugin.ai_query = AsyncMock(side_effect=TimeoutError)
551    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
552    _set_runtime_mass(
553        runtime,
554        _create_engine_mass(
555            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="en_US")
556        ),
557    )
558
559    with pytest.raises(MusicAssistantError) as error:
560        await runtime._generate_text(
561            instructions="test",
562            prompt="test prompt",
563            web_mode="disabled",
564        )
565    assert "query failed: TimeoutError" in str(error.value)
566
567
568async def test_generate_text_asks_for_the_system_locale_language() -> None:
569    """The AI query states the server locale so sections are written in that language."""
570    plugin = _create_ai_plugin("hass_1", "ai_task.default")
571    plugin.ai_query = AsyncMock(return_value="section text")
572    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
573    _set_runtime_mass(
574        runtime,
575        _create_engine_mass(
576            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="nl_NL")
577        ),
578    )
579
580    await runtime._generate_text(
581        instructions="test",
582        prompt="test prompt",
583        web_mode="disabled",
584    )
585
586    assert "nl_NL" in plugin.ai_query.await_args.args[0]
587    assert plugin.ai_query.await_args.kwargs == {"engine_id": "ai_task.default"}
588
589
590@pytest.mark.parametrize("general", [{"instructions": "Host personality: minimal DJ."}, {}])
591async def test_generate_text_always_states_the_pronunciation_rules(
592    general: dict[str, Any],
593) -> None:
594    """Every query carries the TTS pronunciation rules, with or without station instructions."""
595    plugin = _create_ai_plugin("hass_1", "ai_task.default")
596    plugin.ai_query = AsyncMock(return_value="section text")
597    runtime = DummyRuntime({CONF_AI_ENGINE: "hass_1/ai_task.default"})
598    _set_runtime_mass(
599        runtime,
600        _create_engine_mass(
601            ProviderFeature.AI_QUERY, plugin, metadata=SimpleNamespace(locale="en_US")
602        ),
603    )
604
605    await runtime._generate_text(
606        instructions=str(general.get("instructions", "")), prompt="test prompt", web_mode="allow"
607    )
608
609    assert TTS_PRONUNCIATION_INSTRUCTIONS in plugin.ai_query.await_args.args[0]
610
611
612def test_resolve_placeholders_keeps_time_and_weather_deferred() -> None:
613    """Static track placeholders resolve at plan time; time and weather stay deferred."""
614    runtime = DummyRuntime()
615    _set_runtime_mass(runtime, SimpleNamespace(metadata=SimpleNamespace(locale="en_US")))
616    tracks = [
617        {"index": 0, "songinfo": "A - One", "duration": 200},
618        {"index": 1, "songinfo": "B - Two", "duration": 200},
619    ]
620    slot = Slot(
621        when="between_songs",
622        at_index=1,
623        prev_index=0,
624        next_index=1,
625        very_next_index=None,
626        minute_mark=3.3,
627    )
628
629    static, deferred = runtime._resolve_placeholders(
630        program={},
631        tracks=tracks,
632        slot=slot,
633        runtime_tokens={"<weather_hourly>": "12 degrees"},
634    )
635
636    assert static["<prev_songinfo>"] == "A - One"
637    assert static["<next_songinfo>"] == "B - Two"
638    assert "<timestamp>" not in static
639    assert "<weather_hourly>" not in static
640    assert deferred["<weather_hourly>"] == "12 degrees"
641    assert "<timestamp>" in deferred
642
643
644def test_plan_sections_leaves_deferred_tokens_in_the_prompt() -> None:
645    """A planned section's prompt keeps its deferred tokens verbatim."""
646    runtime = DummyRuntime()
647    _set_runtime_mass(runtime, SimpleNamespace(metadata=SimpleNamespace(locale="en_US")))
648    station = {
649        "sections": [
650            {
651                "id": "Weather",
652                "name": "Weather",
653                "prompt": "It is <timestamp>. Weather: <weather_hourly>. Next: <next_songinfo>.",
654                "constraints": {"max_chars": 300},
655                "web_search": "disabled",
656            }
657        ],
658        "section_order": [{"when": "between_songs", "flow": [{"MUST": "Weather"}]}],
659    }
660    tracks = [
661        {"index": 0, "songinfo": "A - One", "duration": 200},
662        {"index": 1, "songinfo": "B - Two", "duration": 200},
663    ]
664
665    planned, _history = runtime._plan_sections(
666        session_id="sess",
667        tracks=tracks,
668        program=station,
669        track_index_offset=0,
670        minute_offset=0.0,
671        history_state={},
672        allowed_slot_when=["between_songs"],
673        runtime_tokens={"<weather_hourly>": "12 degrees"},
674    )
675
676    assert planned
677    prompt = planned[0].prompt
678    assert "<timestamp>" in prompt
679    assert "<weather_hourly>" in prompt
680    assert "B - Two" in prompt
681
682
683def _weather_guarded_station() -> dict[str, Any]:
684    """Return a station whose only section requires the weather-hourly token to be present."""
685    return {
686        "sections": [
687            {
688                "id": "Weather",
689                "name": "Weather",
690                "type": "ai_text",
691                "web_search": "disabled",
692                "prompt": "Current weather: <weather_hourly>.",
693                "constraints": {"max_chars": 200},
694            }
695        ],
696        "section_order": [
697            {
698                "when": "between_songs",
699                "flow": [
700                    {
701                        "OPTIONAL": {
702                            "section": "Weather",
703                            "chance": 100,
704                            "guards": {"require_placeholders_present": ["<weather_hourly>"]},
705                        }
706                    }
707                ],
708            }
709        ],
710    }
711
712
713def test_plan_sections_suppresses_section_when_required_placeholder_is_missing() -> None:
714    """A guarded section plans zero entries when its required placeholder never resolved."""
715    runtime = DummyRuntime()
716    _set_runtime_mass(runtime, SimpleNamespace(metadata=SimpleNamespace(locale="en_US")))
717    tracks = [
718        {"index": 0, "songinfo": "A - One", "duration": 200},
719        {"index": 1, "songinfo": "B - Two", "duration": 200},
720    ]
721
722    planned, _history = runtime._plan_sections(
723        session_id="sess",
724        tracks=tracks,
725        program=_weather_guarded_station(),
726        track_index_offset=0,
727        minute_offset=0.0,
728        history_state={},
729        allowed_slot_when=["between_songs"],
730        runtime_tokens={},
731    )
732
733    assert planned == []
734
735
736def test_plan_sections_includes_section_when_required_placeholder_is_present() -> None:
737    """The same guarded section plans normally once its required placeholder resolved."""
738    runtime = DummyRuntime()
739    _set_runtime_mass(runtime, SimpleNamespace(metadata=SimpleNamespace(locale="en_US")))
740    tracks = [
741        {"index": 0, "songinfo": "A - One", "duration": 200},
742        {"index": 1, "songinfo": "B - Two", "duration": 200},
743    ]
744
745    planned, _history = runtime._plan_sections(
746        session_id="sess",
747        tracks=tracks,
748        program=_weather_guarded_station(),
749        track_index_offset=0,
750        minute_offset=0.0,
751        history_state={},
752        allowed_slot_when=["between_songs"],
753        runtime_tokens={"<weather_hourly>": "12 degrees"},
754    )
755
756    assert len(planned) == 1
757    assert planned[0].section_id == "Weather"
758
759
760def _stub_track(item_id: str) -> Track:
761    """Build a minimal Track with one available ProviderMapping, for build_queue_item."""
762    return Track(
763        item_id=item_id,
764        provider="library",
765        name=f"Track {item_id}",
766        provider_mappings={
767            ProviderMapping(
768                item_id=item_id,
769                provider_domain="library",
770                provider_instance="library",
771            )
772        },
773    )
774
775
776def test_compose_queue_items_places_clips_at_planned_indices() -> None:
777    """Clips are interleaved at their planned indices, carrying their render state."""
778    runtime = DummyRuntime()
779    _set_runtime_mass(runtime, SimpleNamespace())
780    tracks = [
781        {"index": 0, "uri": "library://track/1", "media_item": _stub_track("1")},
782        {"index": 1, "uri": "library://track/2", "media_item": _stub_track("2")},
783    ]
784    sections = [
785        PlannedSection(
786            order=0,
787            clip_id="sess_000",
788            section_id="Intro",
789            section_name="Intro",
790            when="start_of_playlist",
791            insert_at_index=0,
792            prompt="hello <timestamp>",
793            max_chars=200,
794            web_search_mode="disabled",
795        ),
796        PlannedSection(
797            order=1,
798            clip_id="sess_001",
799            section_id="Between",
800            section_name="Between",
801            when="between_songs",
802            insert_at_index=1,
803            prompt="middle <weather_hourly>",
804            max_chars=200,
805            web_search_mode="allow",
806        ),
807    ]
808
809    items = runtime._compose_queue_items(
810        queue_id="player_a",
811        session=SessionState(session_id="sess", station_id="st"),
812        program={"id": "st"},
813        tracks=tracks,
814        sections=sections,
815    )
816
817    assert [item.media_item.item_id for item in items if item.media_item is not None] == [
818        "sess_000",
819        "1",
820        "sess_001",
821        "2",
822    ]
823    intro = items[0]
824    assert intro.media_item is not None
825    assert intro.media_item.media_type == MediaType.SOUND_EFFECT
826    assert intro.extra_attributes[ATTR_PROMPT] == "hello <timestamp>"
827    assert intro.extra_attributes[ATTR_SESSION_ID] == "sess"
828    assert intro.extra_attributes[ATTR_STATION_ID] == "st"
829    assert intro.extra_attributes[ATTR_MAX_CHARS] == 200
830    assert items[2].extra_attributes[ATTR_WEB_SEARCH_MODE] == "allow"
831    # the section name travels as the item's own name, not as an attribute
832    assert intro.name == "Intro"
833    assert "ai_radio_section_name" not in intro.extra_attributes
834    # track items carry no AI Radio state
835    assert items[1].extra_attributes == {}
836
837
838def test_build_program_merges_host_into_station() -> None:
839    """The merged program carries the host's persona, sections and section_order."""
840    runtime = DummyRuntime()
841    runtime._sections = {
842        "Song_Transition": {
843            "id": "Song_Transition",
844            "name": "Song Transition",
845            "type": "ai_text",
846            "prompt": "Prompt",
847            "web_search": "disabled",
848        }
849    }
850    host = {
851        "id": "rick",
852        "name": "Rick",
853        "instructions": "Persona.",
854        "tts_engine": "engine-1",
855        "section_ids": ["Song_Transition"],
856        "section_order": [{"when": "between_songs", "flow": [{"MUST": "Song_Transition"}]}],
857        "merge_section_id": "",
858    }
859    station = {
860        "id": "station_a",
861        "name": "Station A",
862        "source_playlist_id": "p1",
863        "source_playlist_provider": "library",
864        "default_player_id": "",
865        "max_duration_minutes": 0.0,
866        "shuffle_source_tracks": True,
867        "host_id": "rick",
868    }
869
870    program = runtime._build_program(station, host)
871
872    assert program["instructions"] == "Persona."
873    assert program["tts_engine"] == "engine-1"
874    assert [s["id"] for s in program["sections"]] == ["Song_Transition"]
875    assert program["section_order"] == host["section_order"]
876    assert program["source_playlist_id"] == "p1"
877
878
879def test_clip_item_carries_host_id() -> None:
880    """A planned clip's queue item stamps both the station id and the host id."""
881    runtime = DummyRuntime()
882    section = PlannedSection(
883        order=0,
884        clip_id="sess_000",
885        section_id="Song_Transition",
886        section_name="Song Transition",
887        when="between_songs",
888        insert_at_index=1,
889        prompt="p",
890        max_chars=0,
891        web_search_mode="disabled",
892    )
893    program = {"id": "station_a", "host_id": "rick"}
894
895    item = runtime._section_to_clip_item("queue-1", "sess", program, section)
896
897    assert item.extra_attributes[ATTR_HOST_ID] == "rick"
898    assert item.extra_attributes[ATTR_SESSION_ID] == "sess"
899
900
901async def test_get_ai_engine_requires_a_configured_selection() -> None:
902    """Without a stored selection no engine is picked, so the run fails with a clear error."""
903    runtime = DummyRuntime()
904    _set_runtime_mass(
905        runtime, _create_engine_mass(ProviderFeature.AI_QUERY, _create_ai_plugin("hass_1", "one"))
906    )
907
908    with pytest.raises(MusicAssistantError, match="No AI engine available"):
909        await runtime._get_ai_engine()
910
911
912async def test_get_ai_engine_uses_the_configured_selection() -> None:
913    """A configured engine uid wins over the first available engine."""
914    high_priority = _create_ai_plugin("zz_high", "engine")
915    low_priority = _create_ai_plugin("aa_low", "engine")
916    runtime = DummyRuntime({CONF_AI_ENGINE: "aa_low/engine"})
917    _set_runtime_mass(
918        runtime, _create_engine_mass(ProviderFeature.AI_QUERY, high_priority, low_priority)
919    )
920
921    assert (await runtime._get_ai_engine()).uid == "aa_low/engine"
922
923
924async def test_get_ai_engine_refuses_a_configured_engine_that_disappeared() -> None:
925    """A concrete AI selection is never silently replaced by another available engine."""
926    runtime = DummyRuntime({CONF_AI_ENGINE: "gone/engine"})
927    _set_runtime_mass(
928        runtime, _create_engine_mass(ProviderFeature.AI_QUERY, _create_ai_plugin("hass_1", "one"))
929    )
930
931    with pytest.raises(MusicAssistantError, match="No AI engine available"):
932        await runtime._get_ai_engine()
933
934
935async def test_get_tts_engine_uses_the_configured_selection() -> None:
936    """The stored TTS uid selects its engine, whatever order the plugins are served in."""
937    high_priority = _create_tts_plugin("zz_high", "engine")
938    low_priority = _create_tts_plugin("aa_low", "engine")
939    runtime = DummyRuntime({CONF_TTS_ENGINE: "aa_low/engine"})
940    _set_runtime_mass(
941        runtime, _create_engine_mass(ProviderFeature.TTS, high_priority, low_priority)
942    )
943
944    assert (await runtime._get_tts_engine()).uid == "aa_low/engine"
945
946
947async def test_get_tts_engine_refuses_a_configured_engine_that_disappeared() -> None:
948    """A concrete TTS selection is never silently replaced by another available engine."""
949    runtime = DummyRuntime({CONF_TTS_ENGINE: "gone/engine"})
950    _set_runtime_mass(
951        runtime, _create_engine_mass(ProviderFeature.TTS, _create_tts_plugin("hass_1", "one"))
952    )
953
954    with pytest.raises(MusicAssistantError, match="No text-to-speech engine available"):
955        await runtime._get_tts_engine()
956
957
958async def test_get_tts_engine_falls_back_to_provider_selection_when_host_uid_is_unresolvable(
959    caplog: Any,
960) -> None:
961    """A host engine_uid that no longer resolves falls back to the provider's TTS selection."""
962    runtime = DummyRuntime({CONF_TTS_ENGINE: "aa_low/engine"})
963    _set_runtime_mass(
964        runtime, _create_engine_mass(ProviderFeature.TTS, _create_tts_plugin("aa_low", "engine"))
965    )
966
967    with caplog.at_level(logging.WARNING):
968        engine = await runtime._get_tts_engine("gone/engine")
969
970    assert engine.uid == "aa_low/engine"
971    assert any("unavailable" in message for message in caplog.messages)
972
973
974def _show_mass_stub(**handlers: Any) -> SimpleNamespace:
975    """
976    Build a minimal mass stub for exercising _run_show.
977
978    Any player_queues/players/music/metadata handler not passed gets a no-op default.
979    """
980
981    async def _noop_async(*_args: Any, **_kwargs: Any) -> None:
982        return None
983
984    def _noop_sync(*_args: Any, **_kwargs: Any) -> None:
985        return None
986
987    def _noop_get_active_queue(_player_id: str) -> Any:
988        return None
989
990    def _noop_get_player(_player_id: str) -> Any:
991        return object()
992
993    def _noop_items(_queue_id: str, limit: int = 500, offset: int = 0) -> list[Any]:  # noqa: ARG001
994        return []
995
996    subscribers: list[Callable[[Any], None]] = []
997
998    def _recording_subscribe(
999        cb_func: Callable[[Any], None],
1000        event_filter: Any = None,  # noqa: ARG001
1001        id_filter: Any = None,  # noqa: ARG001
1002    ) -> Callable[[], None]:
1003        subscribers.append(cb_func)
1004
1005        def _unsubscribe() -> None:
1006            subscribers.remove(cb_func)
1007
1008        return _unsubscribe
1009
1010    def _emit_queue_updated(queue_id: str) -> None:
1011        event = MassEvent(event=EventType.QUEUE_UPDATED, object_id=queue_id)
1012        for cb_func in subscribers:
1013            cb_func(event)
1014
1015    def _emit_player_removed(player_id: str) -> None:
1016        event = MassEvent(event=EventType.PLAYER_REMOVED, object_id=player_id)
1017        for cb_func in subscribers:
1018            cb_func(event)
1019
1020    player_queues = SimpleNamespace(
1021        clear=handlers.get("clear", _noop_sync),
1022        get=handlers.get("get", lambda _queue_id: None),
1023        get_active_queue=handlers.get("get_active_queue", _noop_get_active_queue),
1024        set_shuffle=handlers.get("set_shuffle", _noop_async),
1025        load=handlers.get("load", _noop_async),
1026        play_index=handlers.get("play_index", _noop_async),
1027        items=handlers.get("items", _noop_items),
1028        signal_update=handlers.get("signal_update", _noop_sync),
1029        stop=handlers.get("stop", _noop_async),
1030    )
1031    return SimpleNamespace(
1032        player_queues=player_queues,
1033        players=SimpleNamespace(get_player=handlers.get("get_player", _noop_get_player)),
1034        music=SimpleNamespace(playlists=handlers.get("playlists", SimpleNamespace())),
1035        metadata=SimpleNamespace(locale=handlers.get("locale", "en_US")),
1036        create_task=handlers.get("create_task", _noop_sync),
1037        subscribe=handlers.get("subscribe", _recording_subscribe),
1038        emit_queue_updated=_emit_queue_updated,
1039        emit_player_removed=_emit_player_removed,
1040    )
1041
1042
1043def _stub_queue(state: PlaybackState, current_index: int | None) -> SimpleNamespace:
1044    """Build a mutable queue stand-in exposing the fields _await_show_end reads."""
1045    return SimpleNamespace(state=state, current_index=current_index)
1046
1047
1048def _stub_clip_queue_item(clip_id: str, session_id: str) -> SimpleNamespace:
1049    """Build a queue item stand-in whose extra_attributes carry a session id."""
1050    return SimpleNamespace(extra_attributes={ATTR_SESSION_ID: session_id}, item_id=clip_id)
1051
1052
1053def _recording_set_shuffle(log: list[str]) -> Callable[[str, bool], Awaitable[None]]:
1054    """Return an async set_shuffle stub that appends "set_shuffle" to the given call-order log."""
1055
1056    async def _set_shuffle(_queue_id: str, _shuffle_enabled: bool) -> None:
1057        log.append("set_shuffle")
1058
1059    return _set_shuffle
1060
1061
1062def _show_station() -> dict[str, Any]:
1063    """Return a station config for _run_show tests, whose section_order yields clips."""
1064    return {
1065        "id": "st",
1066        "name": "Show Station",
1067        "default_player_id": "living_room",
1068        "source_playlist_id": "playlist-1",
1069        "source_playlist_provider": "library",
1070        "shuffle_source_tracks": False,
1071        "general": {"timezone": "UTC"},
1072        "sections": [
1073            {
1074                "id": "Song_Introduction_Start",
1075                "name": "Intro",
1076                "type": "ai_text",
1077                "web_search": "disabled",
1078                "prompt": "Welcome, next up is <next_songinfo>.",
1079                "constraints": {"max_chars": 200},
1080            },
1081            {
1082                "id": "Song_Transition",
1083                "name": "Transition",
1084                "type": "ai_text",
1085                "web_search": "disabled",
1086                "prompt": "From <prev_songinfo> to <next_songinfo>.",
1087                "constraints": {"max_chars": 200},
1088            },
1089        ],
1090        "section_order": [
1091            {"when": "start_of_playlist", "flow": [{"MUST": "Song_Introduction_Start"}]},
1092            {"when": "between_songs", "flow": [{"MUST": "Song_Transition"}]},
1093        ],
1094    }
1095
1096
1097class ShowRuntime(DummyRuntime):
1098    """Runtime harness exercising the real _run_show with stubbed track sourcing."""
1099
1100    async def _fetch_source_tracks(
1101        self, station: dict[str, Any]
1102    ) -> tuple[list[dict[str, Any]], str]:
1103        """Return two fixed tracks, each carrying its resolved media item."""
1104        return [
1105            {"index": 0, "songinfo": "A - One", "duration": 200, "media_item": _stub_track("1")},
1106            {"index": 1, "songinfo": "B - Two", "duration": 200, "media_item": _stub_track("2")},
1107        ], "Source Playlist"
1108
1109    async def _prepare_runtime_tokens(self, station: dict[str, Any]) -> dict[str, str]:
1110        """Skip the weather lookup; runtime tokens are irrelevant to these tests."""
1111        return {}
1112
1113
1114class ShowRuntimeWithDJ(AIRadioQueueDJMixin, AIRadioStorageMixin, ShowRuntime):
1115    """ShowRuntime harness that also carries sticky queue DJ state."""
1116
1117    def __init__(self, tmp_path: Path) -> None:
1118        """Initialize show runtime state plus queue DJ bookkeeping."""
1119        super().__init__()
1120        self._hosts: dict[str, dict[str, Any]] = {
1121            "rick": {"id": "rick", "name": "Rick", "instructions": "x", "tts_engine": ""},
1122        }
1123        self._dj_queues: dict[str, Any] = {}
1124        self._dj_file = tmp_path / "queue_dj.json"
1125        self._dj_lock = asyncio.Lock()
1126        self._unloading = False
1127
1128
1129def _recording_create_task(scheduled: list[str]) -> Callable[..., None]:
1130    """Return a create_task stub that records the task id and discards the coroutine."""
1131
1132    def _create_task(coro: Any, task_id: str | None = None, **_kwargs: Any) -> None:
1133        if task_id:
1134            scheduled.append(task_id)
1135        coro.close()
1136
1137    return _create_task
1138
1139
1140async def test_run_show_loads_the_whole_show_then_plays_index_zero() -> None:
1141    """The show is loaded in one call, fully stamped, before playback is started."""
1142    runtime = ShowRuntime()
1143    call_order: list[str] = []
1144    loaded: list[tuple[Any, dict[str, Any]]] = []
1145
1146    async def _load(_queue_id: str, queue_items: list[Any], **kwargs: Any) -> None:
1147        call_order.append("load")
1148        # snapshot media_item + extra_attributes now: asserting on the live queue_item
1149        # objects after _run_show returns would pass even if stamping happened later
1150        loaded.extend((item.media_item, dict(item.extra_attributes)) for item in queue_items)
1151        assert kwargs["shuffle"] is False
1152        assert kwargs["keep_remaining"] is False
1153        assert kwargs["keep_played"] is False
1154
1155    async def _play_index(_queue_id: str, index: int) -> None:
1156        call_order.append(f"play_index:{index}")
1157
1158    _set_runtime_mass(
1159        runtime,
1160        _show_mass_stub(
1161            load=_load,
1162            play_index=_play_index,
1163            clear=lambda _queue_id: call_order.append("clear"),
1164            set_shuffle=_recording_set_shuffle(call_order),
1165        ),
1166    )
1167
1168    await runtime._run_show(SessionState(session_id="sess", station_id="st"), _show_station())
1169
1170    assert call_order == ["clear", "set_shuffle", "load", "play_index:0"]
1171    clips = [
1172        (media_item, attrs)
1173        for media_item, attrs in loaded
1174        if media_item.media_type == MediaType.SOUND_EFFECT
1175    ]
1176    assert clips
1177    # every clip was already fully stamped at the moment load() was called
1178    assert all(attrs[ATTR_PROMPT] for _media_item, attrs in clips)
1179    assert all(attrs[ATTR_SESSION_ID] == "sess" for _media_item, attrs in clips)
1180
1181
1182async def test_run_show_targets_active_group_queue() -> None:
1183    """Queue and start the show on the active (group) queue when the player is grouped."""
1184    runtime = ShowRuntime()
1185    clear_calls: list[str] = []
1186    load_queue_ids: list[str] = []
1187    play_index_queue_ids: list[str] = []
1188
1189    async def _load(queue_id: str, **_kwargs: Any) -> None:
1190        load_queue_ids.append(queue_id)
1191
1192    async def _play_index(queue_id: str, _index: int) -> None:
1193        play_index_queue_ids.append(queue_id)
1194
1195    _set_runtime_mass(
1196        runtime,
1197        _show_mass_stub(
1198            get_active_queue=lambda _player_id: SimpleNamespace(queue_id="group_1"),
1199            clear=clear_calls.append,
1200            load=_load,
1201            play_index=_play_index,
1202        ),
1203    )
1204    session = SessionState(session_id="s1", station_id="st")
1205
1206    result = await runtime._run_show(session, _show_station())
1207
1208    assert result["queue_id"] == "group_1"
1209    assert clear_calls == ["group_1"]
1210    assert load_queue_ids == ["group_1"]
1211    assert play_index_queue_ids == ["group_1"]
1212    assert session.queue_id == "group_1"
1213
1214
1215async def test_run_show_clears_the_queues_sticky_dj(tmp_path: Path) -> None:
1216    """Starting a show drops that queue's existing sticky DJ assignment."""
1217    runtime = ShowRuntimeWithDJ(tmp_path)
1218    _set_runtime_mass(runtime, _show_mass_stub())
1219    await runtime.set_queue_dj("living_room", "rick")
1220    assert "living_room" in runtime._dj_queues
1221
1222    await runtime._run_show(SessionState(session_id="sess", station_id="st"), _show_station())
1223
1224    assert "living_room" not in runtime._dj_queues
1225    persisted = json.loads(runtime._dj_file.read_text())
1226    assert persisted["queues"] == {}
1227
1228
1229async def test_run_show_clears_the_dj_on_the_resolved_group_queue(tmp_path: Path) -> None:
1230    """A grouped player's DJ is cleared on the active (group) queue, not the raw player id."""
1231    runtime = ShowRuntimeWithDJ(tmp_path)
1232    _set_runtime_mass(
1233        runtime,
1234        _show_mass_stub(get_active_queue=lambda _player_id: SimpleNamespace(queue_id="group_1")),
1235    )
1236    # a stale assignment on the raw player id must survive untouched: the show never
1237    # played there, only on the resolved group queue
1238    await runtime.set_queue_dj("living_room", "rick")
1239    await runtime.set_queue_dj("group_1", "rick")
1240
1241    await runtime._run_show(SessionState(session_id="s1", station_id="st"), _show_station())
1242
1243    assert "group_1" not in runtime._dj_queues
1244    assert "living_room" in runtime._dj_queues
1245
1246
1247async def test_run_session_finally_replans_a_dj_armed_mid_show(tmp_path: Path) -> None:
1248    """A DJ armed via the menu while a show plays still gets scheduled once the show ends."""
1249    runtime = ShowRuntimeWithDJ(tmp_path)
1250    scheduled: list[str] = []
1251    _set_runtime_mass(runtime, _show_mass_stub(create_task=_recording_create_task(scheduled)))
1252    session = SessionState(session_id="sess", station_id="st", queue_id="living_room")
1253    runtime._sessions[session.session_id] = session
1254
1255    # arming mid-show already requested a pass; that pass would drain against the running-show
1256    # guard in _replan_queue and clear replan_pending without planning anything, so reset it
1257    # here to isolate the finally block's own request instead of piggybacking on this one
1258    await runtime.set_queue_dj("living_room", "rick")
1259    runtime._dj_queues["living_room"].replan_pending = False
1260    scheduled.clear()
1261
1262    async def _run_show_stub(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
1263        raise RuntimeError("show over")
1264
1265    runtime._run_show = _run_show_stub  # type: ignore[method-assign]
1266
1267    await runtime._run_session(session.session_id, {"id": "st"})
1268
1269    assert scheduled == ["ai_radio_dj_replan_living_room"]
1270    assert runtime._dj_queues["living_room"].ready is True
1271
1272
1273async def test_run_show_ends_as_stopped_when_the_user_stops_the_queue() -> None:
1274    """A queue stopped part-way through the show ends the run as a user stop."""
1275    runtime = DummyRuntime()
1276    session = SessionState(session_id="sess", station_id="st")
1277    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=0)
1278    mass = _show_mass_stub(
1279        get=lambda _queue_id: queue,
1280        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1281            _stub_clip_queue_item("sess_000", session_id="sess")
1282        ],
1283    )
1284    _set_runtime_mass(runtime, mass)
1285
1286    task = asyncio.create_task(
1287        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1288    )
1289    await asyncio.sleep(0)
1290    assert not task.done()
1291
1292    queue.state = PlaybackState.IDLE
1293    mass.emit_queue_updated("player_a")
1294
1295    assert await asyncio.wait_for(task, timeout=1) == "queue_stopped"
1296
1297
1298async def test_run_show_ends_as_exhausted_when_the_show_plays_out() -> None:
1299    """Reaching the last enqueued entry ends the run as a normal completion."""
1300    runtime = DummyRuntime()
1301    session = SessionState(session_id="sess", station_id="st")
1302    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=5)
1303    mass = _show_mass_stub(
1304        get=lambda _queue_id: queue,
1305        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1306            _stub_clip_queue_item("sess_000", session_id="sess")
1307        ],
1308    )
1309    _set_runtime_mass(runtime, mass)
1310
1311    task = asyncio.create_task(
1312        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1313    )
1314    await asyncio.sleep(0)
1315
1316    queue.state = PlaybackState.IDLE
1317    mass.emit_queue_updated("player_a")
1318
1319    assert await asyncio.wait_for(task, timeout=1) == "source_exhausted"
1320
1321
1322async def test_run_show_ignores_an_idle_queue_before_playback_starts() -> None:
1323    """A queue that has not started yet is not mistaken for a stopped one."""
1324    runtime = DummyRuntime()
1325    session = SessionState(session_id="sess", station_id="st")
1326    queue = _stub_queue(state=PlaybackState.IDLE, current_index=None)
1327    mass = _show_mass_stub(
1328        get=lambda _queue_id: queue,
1329        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1330            _stub_clip_queue_item("sess_000", session_id="sess")
1331        ],
1332    )
1333    _set_runtime_mass(runtime, mass)
1334
1335    task = asyncio.create_task(
1336        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1337    )
1338    mass.emit_queue_updated("player_a")
1339    await asyncio.sleep(0)
1340
1341    assert not task.done()
1342    task.cancel()
1343
1344
1345async def test_run_show_keeps_a_paused_queue_on_air() -> None:
1346    """A paused queue keeps the show running."""
1347    runtime = DummyRuntime()
1348    session = SessionState(session_id="sess", station_id="st")
1349    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=1)
1350    mass = _show_mass_stub(
1351        get=lambda _queue_id: queue,
1352        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1353            _stub_clip_queue_item("sess_000", session_id="sess")
1354        ],
1355    )
1356    _set_runtime_mass(runtime, mass)
1357
1358    task = asyncio.create_task(
1359        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1360    )
1361    await asyncio.sleep(0)
1362
1363    queue.state = PlaybackState.PAUSED
1364    mass.emit_queue_updated("player_a")
1365    await asyncio.sleep(0)
1366
1367    assert not task.done()
1368    task.cancel()
1369
1370
1371async def test_run_show_ends_when_the_queue_no_longer_holds_its_clips() -> None:
1372    """A queue cleared or taken over by other playback ends the run."""
1373    runtime = DummyRuntime()
1374    session = SessionState(session_id="sess", station_id="st")
1375    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=1)
1376    queue_items = [_stub_clip_queue_item("sess_000", session_id="sess")]
1377    mass = _show_mass_stub(
1378        get=lambda _queue_id: queue,
1379        items=lambda _queue_id, limit=500, offset=0: queue_items[offset : offset + limit],
1380    )
1381    _set_runtime_mass(runtime, mass)
1382
1383    task = asyncio.create_task(
1384        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1385    )
1386    await asyncio.sleep(0)
1387
1388    queue_items.clear()
1389    mass.emit_queue_updated("player_a")
1390
1391    assert await asyncio.wait_for(task, timeout=1) == "queue_stopped"
1392
1393
1394async def test_run_show_ends_when_its_player_is_removed() -> None:
1395    """Removing the target player must not pin the session's slot forever."""
1396    runtime = DummyRuntime()
1397    session = SessionState(session_id="sess", station_id="st")
1398    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=1)
1399    queue_holder: list[Any] = [queue]
1400    mass = _show_mass_stub(
1401        get=lambda _queue_id: queue_holder[0],
1402        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1403            _stub_clip_queue_item("sess_000", session_id="sess")
1404        ],
1405    )
1406    _set_runtime_mass(runtime, mass)
1407
1408    task = asyncio.create_task(
1409        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1410    )
1411    await asyncio.sleep(0)
1412    assert not task.done()
1413
1414    # on_player_remove pops the queue data before PLAYER_REMOVED is signaled
1415    queue_holder[0] = None
1416    mass.emit_player_removed("player_a")
1417
1418    assert await asyncio.wait_for(task, timeout=1) == "queue_stopped"
1419
1420
1421async def test_run_show_keeps_waiting_when_the_show_has_no_clips_to_lose() -> None:
1422    """A clip-free show is not mistaken for one whose clips got cleared out."""
1423    runtime = DummyRuntime()
1424    session = SessionState(session_id="sess", station_id="st")
1425    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=0)
1426    # track-only queue: no item carries ATTR_SESSION_ID, exactly like a show whose
1427    # section rules never selected anything to insert
1428    mass = _show_mass_stub(
1429        get=lambda _queue_id: queue,
1430        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1431            SimpleNamespace(extra_attributes={})
1432        ],
1433    )
1434    _set_runtime_mass(runtime, mass)
1435
1436    task = asyncio.create_task(
1437        runtime._await_show_end(session, "player_a", last_index=5, has_clips=False)
1438    )
1439    await asyncio.sleep(0)
1440    assert not task.done()
1441
1442    queue.current_index = 5
1443    queue.state = PlaybackState.IDLE
1444    mass.emit_queue_updated("player_a")
1445
1446    assert await asyncio.wait_for(task, timeout=1) == "source_exhausted"
1447
1448
1449async def test_await_show_end_fails_when_playback_never_starts(
1450    monkeypatch: pytest.MonkeyPatch,
1451) -> None:
1452    """A show whose playback never starts is declared failed instead of waiting forever."""
1453    monkeypatch.setattr(
1454        "music_assistant.providers.ai_radio.runtime.SHOW_START_TIMEOUT_SECONDS", 0.05
1455    )
1456    runtime = DummyRuntime()
1457    session = SessionState(session_id="sess", station_id="st")
1458    queue = _stub_queue(state=PlaybackState.IDLE, current_index=None)
1459    mass = _show_mass_stub(
1460        get=lambda _queue_id: queue,
1461        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1462            _stub_clip_queue_item("sess_000", session_id="sess")
1463        ],
1464    )
1465    _set_runtime_mass(runtime, mass)
1466
1467    with pytest.raises(MusicAssistantError, match="did not start"):
1468        await runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1469
1470
1471async def test_await_show_end_does_not_time_out_once_playback_starts(
1472    monkeypatch: pytest.MonkeyPatch,
1473) -> None:
1474    """Playback starting before the start-timeout elapses lets the show proceed as normal."""
1475    monkeypatch.setattr(
1476        "music_assistant.providers.ai_radio.runtime.SHOW_START_TIMEOUT_SECONDS", 0.2
1477    )
1478    runtime = DummyRuntime()
1479    session = SessionState(session_id="sess", station_id="st")
1480    queue = _stub_queue(state=PlaybackState.IDLE, current_index=None)
1481    mass = _show_mass_stub(
1482        get=lambda _queue_id: queue,
1483        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1484            _stub_clip_queue_item("sess_000", session_id="sess")
1485        ],
1486    )
1487    _set_runtime_mass(runtime, mass)
1488
1489    task = asyncio.create_task(
1490        runtime._await_show_end(session, "player_a", last_index=5, has_clips=True)
1491    )
1492    await asyncio.sleep(0)
1493    assert not task.done()
1494
1495    queue.state = PlaybackState.PLAYING
1496    queue.current_index = 0
1497    mass.emit_queue_updated("player_a")
1498    await asyncio.sleep(0)
1499    assert not task.done()
1500
1501    queue.current_index = 5
1502    queue.state = PlaybackState.IDLE
1503    mass.emit_queue_updated("player_a")
1504
1505    assert await asyncio.wait_for(task, timeout=1) == "source_exhausted"
1506
1507
1508def test_passes_optional_guards_handles_non_numeric_guard_values() -> None:
1509    """Treat non-numeric guard values as disabled instead of raising ValueError."""
1510    runtime = DummyRuntime()
1511    slot = Slot(
1512        when="between_songs",
1513        at_index=1,
1514        prev_index=0,
1515        next_index=1,
1516        very_next_index=2,
1517        minute_mark=5.0,
1518    )
1519
1520    result = runtime._passes_optional_guards(
1521        section_id="Weather_Short",
1522        guards={"min_gap_songs": "abc", "max_per_60min": "xyz"},
1523        history={},
1524        slot=slot,
1525        tracks=[{}, {}, {}],
1526        placeholders={},
1527        track_index_offset=0,
1528        minute_offset=0.0,
1529    )
1530
1531    assert result is True
1532
1533
1534async def test_fetch_source_tracks_skips_tracks_with_no_resolvable_uri(caplog: Any) -> None:
1535    """Skip and warn about source tracks with no resolvable uri instead of queuing a dead entry."""
1536
1537    class DummyPlaylist:
1538        name = "Source Playlist"
1539
1540    class DummyPlaylistsController:
1541        def __init__(self, tracks: list[Any]) -> None:
1542            self._tracks = tracks
1543
1544        async def get(self, playlist_id: str, provider: str) -> Any:
1545            return DummyPlaylist()
1546
1547        async def tracks(self, playlist_id: str, provider: str) -> Any:
1548            for track in self._tracks:
1549                yield track
1550
1551    class DummyTrack:
1552        def __init__(self, item_id: str, name: str, uri: str = "") -> None:
1553            self.item_id = item_id
1554            self.name = name
1555            self.artists: list[Any] = []
1556            self.duration = 180
1557            self.uri = uri
1558            self.provider_mappings: list[Any] = []
1559
1560    good_track_1 = DummyTrack("1", "Track One", uri="library://track/1")
1561    unresolvable_track = DummyTrack("2", "Track Two")
1562    good_track_2 = DummyTrack("3", "Track Three", uri="library://track/3")
1563
1564    class DummyMusic:
1565        playlists = DummyPlaylistsController([good_track_1, unresolvable_track, good_track_2])
1566
1567    class DummyMass:
1568        music = DummyMusic()
1569
1570    runtime = DummyRuntime()
1571    _set_runtime_mass(runtime, DummyMass())
1572    station = {"source_playlist_id": "playlist-1", "source_playlist_provider": "library"}
1573
1574    with caplog.at_level(logging.WARNING):
1575        tracks, playlist_name = await runtime._fetch_source_tracks(station)
1576
1577    assert playlist_name == "Source Playlist"
1578    assert [track["item_id"] for track in tracks] == ["1", "3"]
1579    assert [track["index"] for track in tracks] == [0, 1]
1580    # the resolved media item travels on the normalized dict, unchanged
1581    assert [track["media_item"] for track in tracks] == [good_track_1, good_track_2]
1582    assert any("Track Two" in record.message for record in caplog.records)
1583
1584
1585def test_apply_source_shuffle_returns_unchanged_when_disabled() -> None:
1586    """Leave the source list untouched when the station does not request shuffling."""
1587    runtime = DummyRuntime()
1588    tracks = [{"index": 0, "uri": "a"}, {"index": 1, "uri": "b"}]
1589    station = {"shuffle_source_tracks": False}
1590
1591    result = runtime._apply_source_shuffle(tracks, station)
1592
1593    assert result is tracks
1594
1595
1596def test_apply_source_shuffle_reorders_and_records_source_index(
1597    monkeypatch: pytest.MonkeyPatch,
1598) -> None:
1599    """Shuffle every track into a new order while keeping all of them and their origin."""
1600    # capture the real class before patching it away: runtime.random is the same shared
1601    # stdlib module object, so the lambda below would otherwise re-look-up itself
1602    original_random_cls = random.Random
1603    monkeypatch.setattr(
1604        "music_assistant.providers.ai_radio.runtime.random.Random",
1605        lambda: original_random_cls(1234),
1606    )
1607    runtime = DummyRuntime()
1608    tracks = [{"uri": f"track/{i}"} for i in range(5)]
1609    station = {"shuffle_source_tracks": True}
1610
1611    result = runtime._apply_source_shuffle(tracks, station)
1612
1613    assert [track["index"] for track in result] == list(range(len(tracks)))
1614    assert {track["uri"] for track in result} == {track["uri"] for track in tracks}
1615    for track in result:
1616        assert tracks[track["source_index"]]["uri"] == track["uri"]
1617    # a seeded shuffle must actually reorder, not silently pass through in place
1618    assert [track["source_index"] for track in result] != list(range(len(tracks)))
1619
1620
1621def test_apply_track_duration_limit_keeps_prefix_of_given_order() -> None:
1622    """Truncate to the playtime cap by walking the given order, no shuffling."""
1623    runtime = DummyRuntime()
1624    tracks = [
1625        {"uri": "a", "duration": 120},
1626        {"uri": "b", "duration": 120},
1627        {"uri": "c", "duration": 120},
1628        {"uri": "d", "duration": 120},
1629    ]
1630    station = {"max_duration_minutes": 3}
1631
1632    result = runtime._apply_track_duration_limit(tracks, station)
1633
1634    assert [track["uri"] for track in result] == ["a", "b"]
1635    assert [track["index"] for track in result] == [0, 1]
1636    assert [track["source_index"] for track in result] == [0, 1]
1637
1638
1639def test_apply_track_duration_limit_zero_cap_is_noop() -> None:
1640    """A cap of 0 disables truncation entirely."""
1641    runtime = DummyRuntime()
1642    tracks = [{"uri": "a", "duration": 120}, {"uri": "b", "duration": 120}]
1643    station = {"max_duration_minutes": 0}
1644
1645    result = runtime._apply_track_duration_limit(tracks, station)
1646
1647    assert result is tracks
1648
1649
1650async def test_run_show_disables_shuffle_before_load() -> None:
1651    """Disable queue shuffle before the items are loaded, so sections keep their planned order."""
1652    runtime = ShowRuntime()
1653    call_order: list[str] = []
1654    set_shuffle_calls: list[tuple[str, bool]] = []
1655
1656    async def _set_shuffle(queue_id: str, shuffle_enabled: bool) -> None:
1657        set_shuffle_calls.append((queue_id, shuffle_enabled))
1658        call_order.append("set_shuffle")
1659
1660    async def _load(_queue_id: str, **_kwargs: Any) -> None:
1661        call_order.append("load")
1662
1663    _set_runtime_mass(runtime, _show_mass_stub(set_shuffle=_set_shuffle, load=_load))
1664    session = SessionState(session_id="s1", station_id="st")
1665
1666    await runtime._run_show(session, _show_station())
1667
1668    assert set_shuffle_calls == [("living_room", False)]
1669    assert call_order.index("set_shuffle") < call_order.index("load")
1670
1671
1672async def test_run_show_stays_running_while_the_queue_plays_and_stop_cancels_it() -> None:
1673    """The session stays 'running' for as long as the show plays; a stop cancels it mid-show."""
1674    runtime = ShowRuntime()
1675    queue = _stub_queue(state=PlaybackState.PLAYING, current_index=0)
1676    unsubscribed = False
1677
1678    def _subscribe(
1679        cb_func: Callable[[Any], None],  # noqa: ARG001
1680        event_filter: Any = None,  # noqa: ARG001
1681        id_filter: Any = None,  # noqa: ARG001
1682    ) -> Callable[[], None]:
1683        def _unsubscribe() -> None:
1684            nonlocal unsubscribed
1685            unsubscribed = True
1686
1687        return _unsubscribe
1688
1689    mass = _show_mass_stub(
1690        get=lambda _queue_id: queue,
1691        items=lambda _queue_id, limit=500, offset=0: [  # noqa: ARG005
1692            _stub_clip_queue_item("sess_000", session_id="s1")
1693        ],
1694        subscribe=_subscribe,
1695    )
1696    _set_runtime_mass(runtime, mass)
1697    session = SessionState(session_id="s1", station_id="st")
1698    runtime._sessions[session.session_id] = session
1699
1700    task = asyncio.create_task(runtime._run_session(session.session_id, _show_station()))
1701    await asyncio.sleep(0)
1702    await asyncio.sleep(0)
1703
1704    # the show is on air: the session must stay running, exactly like start_run's
1705    # max-concurrent-runs and station-already-active guards require
1706    assert session.status == "running"
1707    assert not task.done()
1708
1709    # this is what stop_run does to end a run mid-show
1710    task.cancel()
1711    with pytest.raises(asyncio.CancelledError):
1712        await task
1713
1714    assert session.status == "stopped"
1715    assert unsubscribed
1716
1717
1718async def test_run_show_binds_the_session_to_the_target_queue() -> None:
1719    """Record the queue a show plays on so stopping the show can stop it."""
1720    runtime = ShowRuntime()
1721    _set_runtime_mass(runtime, _show_mass_stub())
1722    session = SessionState(session_id="s1", station_id="st")
1723
1724    await runtime._run_show(session, _show_station())
1725
1726    assert session.queue_id == "living_room"
1727
1728
1729async def test_run_session_reports_a_queue_stop_as_stopped() -> None:
1730    """Report a run that ended because the queue was stopped as stopped, not completed."""
1731
1732    class QueueStoppedRuntime(AIRadioRuntimeMixin):
1733        def __init__(self) -> None:
1734            self.logger = logging.getLogger("tests.ai_radio.runtime.queue_stopped")
1735            self._sessions: dict[str, SessionState] = {}
1736
1737        async def _run_show(self, session: SessionState, station: dict[str, Any]) -> dict[str, Any]:
1738            return {"ended_reason": "queue_stopped"}
1739
1740    runtime = QueueStoppedRuntime()
1741    session = SessionState(session_id="s1", station_id="st")
1742    runtime._sessions[session.session_id] = session
1743
1744    await runtime._run_session(session.session_id, {"id": "st"})
1745
1746    assert session.status == "stopped"
1747    assert session.ended_at is not None
1748