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