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