music-assistant-server

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