music-assistant-server

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