music-assistant-server

1"""Unit tests for AI Radio just-in-time clip rendering."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from pathlib import Path
8from types import SimpleNamespace
9from typing import Any, cast
10from unittest.mock import AsyncMock, MagicMock
11
12import pytest
13from music_assistant_models.enums import ContentType, MediaType, StreamType
14from music_assistant_models.errors import (
15    InvalidDataError,
16    MediaNotFoundError,
17    MusicAssistantError,
18)
19from music_assistant_models.media_items import AudioFormat, ProviderMapping, SoundEffect
20from music_assistant_models.queue_item import QueueItem
21
22from music_assistant.helpers.tags import AudioTags
23from music_assistant.models.plugin import PluginProvider, TTSEngine
24from music_assistant.providers.ai_radio.constants import (
25    ATTR_HOST_ID,
26    ATTR_MAX_CHARS,
27    ATTR_PROMPT,
28    ATTR_RENDERED_TEXT,
29    ATTR_SESSION_ID,
30    ATTR_STATION_ID,
31    ATTR_WEB_SEARCH_MODE,
32    CLIP_STREAMDETAILS_EXPIRATION,
33    DEFAULT_LLM_INSTRUCTIONS,
34)
35from music_assistant.providers.ai_radio.models import SessionState
36from music_assistant.providers.ai_radio.rendering import AIRadioRenderMixin
37
38
39class DummyRenderer(AIRadioRenderMixin):
40    """Minimal harness exposing the render path."""
41
42    domain = "ai_radio"
43    instance_id = "ai_radio--test"
44
45    def __init__(self) -> None:
46        """Initialize the harness with recording stubs."""
47        self.logger = logging.getLogger("tests.ai_radio.rendering")
48        self._sessions: dict[str, Any] = {}
49        self._hosts: dict[str, dict[str, Any]] = {}
50        self.llm_prompts: list[str] = []
51        self.tts_texts: list[str] = []
52        self.tts_options: list[dict[str, Any] | None] = []
53        self.weather_calls = 0
54        self.fail_generation = False
55
56    def _configured_now(self) -> Any:
57        return __import__("datetime").datetime(2026, 7, 30, 18, 30)
58
59    async def _generate_text(
60        self, instructions: str, prompt: str, web_mode: str, language: str | None = None
61    ) -> str:
62        # a real suspension point so concurrent callers actually interleave under
63        # asyncio.gather, otherwise the lock in get_stream_details is never exercised
64        await asyncio.sleep(0)
65        if self.fail_generation:
66            raise RuntimeError("llm down")
67        self.llm_prompts.append(prompt)
68        return "Good evening, it is warm out."
69
70    async def _prepare_weather_tokens(self) -> dict[str, str]:
71        self.weather_calls += 1
72        return {"<weather_hourly>": f"fresh weather {self.weather_calls}"}
73
74    async def _render_tts_media(
75        self,
76        text: str,
77        engine_uid: str | None = None,
78        language: str | None = None,
79        options: dict[str, Any] | None = None,
80    ) -> tuple[str, StreamType, AudioFormat]:
81        self.tts_texts.append(text)
82        self.tts_options.append(options)
83        return (
84            f"http://ha.invalid/api/tts_proxy/{len(self.tts_texts)}.mp3",
85            StreamType.HTTP,
86            AudioFormat(content_type=ContentType.MP3),
87        )
88
89    async def _probe_duration(self, path: str) -> int | None:
90        return 9
91
92
93class RealTtsRenderer(DummyRenderer):
94    """Harness that exercises the mixin's real TTS path instead of the DummyRenderer stub."""
95
96    _render_tts_media = AIRadioRenderMixin._render_tts_media
97
98
99def _tts_renderer(path: str, audio_format: AudioFormat | None = None) -> RealTtsRenderer:
100    """Build a renderer whose TTS engine returns StreamDetails carrying the given path."""
101    renderer = RealTtsRenderer()
102    plugin = MagicMock(spec=PluginProvider)
103    plugin.instance_id = "hass_1"
104    plugin.get_tts_message = AsyncMock(
105        return_value=SimpleNamespace(
106            path=path, audio_format=audio_format or AudioFormat(content_type=ContentType.UNKNOWN)
107        )
108    )
109    engine = TTSEngine(id="tts.cloud", name="Cloud", provider=plugin)
110    cast("Any", renderer)._get_tts_engine = AsyncMock(return_value=engine)
111    return renderer
112
113
114def _clip_item(clip_id: str, queue_id: str = "player_a", **overrides: Any) -> QueueItem:
115    """Build a queue item for a pending AI Radio clip."""
116    attributes: dict[str, Any] = {
117        ATTR_SESSION_ID: "sess",
118        ATTR_STATION_ID: "st",
119        ATTR_PROMPT: "It is <timestamp>. Weather: <weather_hourly>.",
120        ATTR_MAX_CHARS: 300,
121        ATTR_WEB_SEARCH_MODE: "disabled",
122    }
123    attributes.update(overrides)
124    media_item = SoundEffect(
125        item_id=clip_id,
126        provider="ai_radio--test",
127        name="Weather",
128        provider_mappings={
129            ProviderMapping(
130                item_id=clip_id,
131                provider_domain="ai_radio",
132                provider_instance="ai_radio--test",
133            )
134        },
135    )
136    return QueueItem(
137        queue_id=queue_id,
138        queue_item_id=f"qi_{clip_id}",
139        name="Weather",
140        duration=None,
141        media_item=media_item,
142        extra_attributes=attributes,
143    )
144
145
146def _attach_queues(renderer: DummyRenderer, queues: dict[str, list[QueueItem]]) -> list[bool]:
147    """Wire a minimal player_queues stub and return the signal_update call log."""
148    signals: list[bool] = []
149    cast("Any", renderer).mass = SimpleNamespace(
150        player_queues=SimpleNamespace(
151            all=lambda: tuple(SimpleNamespace(queue_id=queue_id) for queue_id in queues),
152            items=lambda queue_id, limit=500, offset=0: queues.get(queue_id, [])[
153                offset : offset + limit
154            ],
155            signal_update=lambda _queue_id, items_changed=False: signals.append(items_changed),
156        ),
157        metadata=SimpleNamespace(locale="en_US"),
158    )
159    return signals
160
161
162def _attach_queue(renderer: DummyRenderer, items: list[QueueItem]) -> list[bool]:
163    """Wire a single-queue player_queues stub and return the signal_update call log."""
164    return _attach_queues(renderer, {"player_a": items})
165
166
167async def test_render_resolves_deferred_placeholders_at_render_time() -> None:
168    """The prompt sent to the LLM carries render-time weather, not plan-time."""
169    renderer = DummyRenderer()
170    item = _clip_item("sess_001")
171    _attach_queue(renderer, [item])
172
173    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
174
175    assert renderer.weather_calls == 1
176    assert "fresh weather 1" in renderer.llm_prompts[0]
177    assert "<timestamp>" not in renderer.llm_prompts[0]
178    assert streamdetails.media_type == MediaType.SOUND_EFFECT
179    assert streamdetails.stream_type == StreamType.HTTP
180    assert streamdetails.duration == 9
181    assert streamdetails.expiration == 60
182    assert streamdetails.can_seek is False
183    assert streamdetails.allow_seek is False
184
185
186async def test_render_caches_the_script_and_the_minted_media() -> None:
187    """A second render within the cache window reuses both the stored script and media."""
188    renderer = DummyRenderer()
189    item = _clip_item("sess_001")
190    signals = _attach_queue(renderer, [item])
191
192    first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
193    second = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
194
195    assert len(renderer.llm_prompts) == 1
196    assert renderer.tts_texts == ["Good evening, it is warm out."]
197    assert first.path == second.path
198    assert item.extra_attributes[ATTR_RENDERED_TEXT] == "Good evening, it is warm out."
199    assert signals == [True]
200
201
202async def test_concurrent_renders_call_the_llm_once() -> None:
203    """Two simultaneous requests for one clip render a single script."""
204    renderer = DummyRenderer()
205    _attach_queue(renderer, [_clip_item("sess_001")])
206
207    await asyncio.gather(
208        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
209        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
210    )
211
212    assert len(renderer.llm_prompts) == 1
213
214
215async def test_concurrent_renders_mint_the_clip_only_once() -> None:
216    """Three simultaneous requests for one clip share a single minted TTS render."""
217    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
218    _attach_queue(renderer, [_clip_item("sess_001")])
219
220    results = await asyncio.gather(
221        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
222        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
223        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
224    )
225
226    engine = cast("Any", renderer)._get_tts_engine.return_value
227    engine.provider.get_tts_message.assert_awaited_once()
228    assert len({result.path for result in results}) == 1
229
230
231async def test_cached_media_remints_once_it_expires() -> None:
232    """A render requested after the cache window elapses mints a fresh clip."""
233    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
234    _attach_queue(renderer, [_clip_item("sess_001")])
235
236    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
237    cached = cast("Any", renderer)._media_cache["sess_001"]
238    cached.minted_at -= CLIP_STREAMDETAILS_EXPIRATION + 1
239
240    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
241
242    engine = cast("Any", renderer)._get_tts_engine.return_value
243    assert engine.provider.get_tts_message.await_count == 2
244
245
246async def test_a_late_cache_hit_expires_with_the_url_it_serves() -> None:
247    """A hit late in the window hands out the url's remaining life, not a fresh full window."""
248    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
249    _attach_queue(renderer, [_clip_item("sess_001")])
250
251    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
252    cached = cast("Any", renderer)._media_cache["sess_001"]
253    cached.minted_at -= 45
254
255    late = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
256
257    engine = cast("Any", renderer)._get_tts_engine.return_value
258    assert engine.provider.get_tts_message.await_count == 1
259    assert 14 <= late.expiration <= 15
260
261
262async def test_a_cache_hit_with_no_useful_life_left_remints() -> None:
263    """A hit in the last seconds of the window mints again instead of serving a dying url."""
264    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
265    _attach_queue(renderer, [_clip_item("sess_001")])
266
267    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
268    cached = cast("Any", renderer)._media_cache["sess_001"]
269    cached.minted_at -= CLIP_STREAMDETAILS_EXPIRATION - 2
270
271    fresh = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
272
273    engine = cast("Any", renderer)._get_tts_engine.return_value
274    assert engine.provider.get_tts_message.await_count == 2
275    assert fresh.expiration == CLIP_STREAMDETAILS_EXPIRATION
276
277
278async def test_expired_cache_entries_are_pruned_on_the_next_mint() -> None:
279    """Minting a clip drops the entries whose urls died, so the cache cannot grow forever."""
280    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
281    _attach_queue(renderer, [_clip_item("sess_001"), _clip_item("sess_002")])
282
283    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
284    media_cache = cast("Any", renderer)._media_cache
285    media_cache["sess_001"].minted_at -= CLIP_STREAMDETAILS_EXPIRATION + 1
286    await renderer.get_stream_details("sess_002", MediaType.SOUND_EFFECT)
287
288    assert set(media_cache) == {"sess_002"}
289
290
291async def test_render_tts_media_passes_the_locale_as_language() -> None:
292    """The DJ script's locale reaches the TTS engine as a hyphenated language code."""
293    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
294    _attach_queue(renderer, [_clip_item("sess_001")])
295
296    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
297
298    engine = cast("Any", renderer)._get_tts_engine.return_value
299    engine.provider.get_tts_message.assert_awaited_once_with(
300        "Good evening, it is warm out.", language="en-US", engine_id="tts.cloud", options={}
301    )
302
303
304async def test_render_tts_media_falls_back_without_language_on_rejection() -> None:
305    """An engine that rejects the requested language is retried once without it."""
306    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
307    _attach_queue(renderer, [_clip_item("sess_001")])
308    engine = cast("Any", renderer)._get_tts_engine.return_value
309    engine.provider.get_tts_message = AsyncMock(
310        side_effect=[
311            Exception("unsupported language"),
312            SimpleNamespace(
313                path="http://example.test/api/tts_proxy/abc123.mp3",
314                audio_format=AudioFormat(content_type=ContentType.MP3),
315            ),
316        ]
317    )
318
319    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
320
321    assert streamdetails.path == "http://example.test/api/tts_proxy/abc123.mp3"
322    assert engine.provider.get_tts_message.await_count == 2
323    first_call, second_call = engine.provider.get_tts_message.await_args_list
324    assert first_call.kwargs["language"] == "en-US"
325    assert second_call.kwargs["language"] is None
326
327
328async def test_render_tts_media_does_not_retry_after_a_timeout_style_failure() -> None:
329    """A structured MusicAssistantError is not a language rejection, so it skips the retry."""
330    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
331    _attach_queue(renderer, [_clip_item("sess_001")])
332    engine = cast("Any", renderer)._get_tts_engine.return_value
333    engine.provider.get_tts_message = AsyncMock(
334        side_effect=MusicAssistantError("engine did not respond within 5s")
335    )
336
337    with pytest.raises(MediaNotFoundError):
338        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
339
340    engine.provider.get_tts_message.assert_awaited_once()
341
342
343async def test_clip_is_found_in_the_owning_sessions_queue() -> None:
344    """The session registry points the lookup at the queue that holds the clip."""
345    renderer = DummyRenderer()
346    session = SessionState(session_id="sess", station_id="st", queue_id="player_b")
347    renderer._sessions = {"sess": session}
348    _attach_queues(
349        renderer,
350        {"player_a": [], "player_b": [_clip_item("sess_001", queue_id="player_b")]},
351    )
352
353    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
354
355    assert streamdetails.item_id == "sess_001"
356
357
358async def test_clip_is_found_by_scanning_every_queue_after_a_restart() -> None:
359    """With the session registry gone, the clip is still located in its persisted queue."""
360    renderer = DummyRenderer()
361    _attach_queues(
362        renderer,
363        {"player_a": [], "player_b": [_clip_item("sess_001", queue_id="player_b")]},
364    )
365
366    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
367
368    assert streamdetails.item_id == "sess_001"
369
370
371async def test_unknown_clip_raises_media_not_found() -> None:
372    """A clip id that is not in any queue is reported as missing media."""
373    renderer = DummyRenderer()
374    _attach_queue(renderer, [_clip_item("sess_001")])
375
376    with pytest.raises(MediaNotFoundError):
377        await renderer.get_stream_details("sess_999", MediaType.SOUND_EFFECT)
378
379
380async def test_clip_without_prompt_raises_media_not_found() -> None:
381    """A clip whose attributes were lost is reported as missing media."""
382    renderer = DummyRenderer()
383    item = _clip_item("sess_001")
384    item.extra_attributes.pop(ATTR_PROMPT)
385    _attach_queue(renderer, [item])
386
387    with pytest.raises(MediaNotFoundError):
388        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
389
390
391async def test_llm_failure_raises_media_not_found() -> None:
392    """An LLM failure surfaces as missing media so the core skips the clip."""
393    renderer = DummyRenderer()
394    renderer.fail_generation = True
395    _attach_queue(renderer, [_clip_item("sess_001")])
396
397    with pytest.raises(MediaNotFoundError):
398        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
399
400
401async def test_render_failure_increments_the_session_skip_counter() -> None:
402    """A skipped clip is recorded on its session without failing the run."""
403    renderer = DummyRenderer()
404    session = SessionState(session_id="sess", station_id="st")
405    renderer._sessions = {"sess": session}
406    _attach_queue(renderer, [_clip_item("sess_001")])
407    renderer.fail_generation = True
408
409    with pytest.raises(MediaNotFoundError):
410        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
411
412    assert session.skipped_sections == 1
413    assert session.last_render_error
414
415
416async def test_tts_failure_raises_media_not_found_and_records_skip() -> None:
417    """A TTS failure surfaces as missing media and is recorded on the owning session."""
418
419    class UnspeakableRenderer(DummyRenderer):
420        async def _render_tts_media(
421            self,
422            text: str,
423            engine_uid: str | None = None,
424            language: str | None = None,
425            options: dict[str, Any] | None = None,
426        ) -> tuple[str, StreamType, AudioFormat]:
427            raise RuntimeError("tts down")
428
429    renderer = UnspeakableRenderer()
430    session = SessionState(session_id="sess", station_id="st")
431    renderer._sessions = {"sess": session}
432    _attach_queue(renderer, [_clip_item("sess_001")])
433
434    with pytest.raises(MediaNotFoundError):
435        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
436
437    assert session.skipped_sections == 1
438    assert session.last_render_error
439
440
441async def test_probe_failure_is_not_fatal() -> None:
442    """A failed duration probe yields streamdetails without a duration."""
443
444    class UnprobableRenderer(DummyRenderer):
445        async def _probe_duration(self, path: str) -> int | None:
446            return None
447
448    renderer = UnprobableRenderer()
449    _attach_queue(renderer, [_clip_item("sess_001")])
450
451    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
452
453    assert streamdetails.duration is None
454
455
456class RealProbeRenderer(DummyRenderer):
457    """Harness that exercises the mixin's real duration probe instead of the stub."""
458
459    _probe_duration = AIRadioRenderMixin._probe_duration
460
461
462def _failing_probe(message: str, monkeypatch: pytest.MonkeyPatch) -> RealProbeRenderer:
463    """Build a renderer whose duration probe fails with the given ffprobe message."""
464
465    async def _raise(*_args: Any, **_kwargs: Any) -> AudioTags:
466        raise InvalidDataError(message)
467
468    monkeypatch.setattr("music_assistant.providers.ai_radio.rendering.async_parse_tags", _raise)
469    renderer = RealProbeRenderer()
470    renderer._sessions = {"sess": SessionState(session_id="sess", station_id="st")}
471    _attach_queue(renderer, [_clip_item("sess_001")])
472    return renderer
473
474
475async def test_tts_server_error_fails_the_clip_with_an_actionable_message(
476    monkeypatch: pytest.MonkeyPatch,
477) -> None:
478    """An engine that hands out a URL it cannot render fails the clip, not the playback."""
479    renderer = _failing_probe(
480        "Unable to retrieve info for http://ha.invalid/api/tts_proxy/1.mp3 "
481        "(Server returned 5XX Server Error reply)",
482        monkeypatch,
483    )
484
485    with pytest.raises(MediaNotFoundError):
486        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
487
488    session = renderer._sessions["sess"]
489    assert session.skipped_sections == 1
490    assert "enough credit" in session.last_render_error
491
492
493async def test_unmeasurable_clip_still_plays(monkeypatch: pytest.MonkeyPatch) -> None:
494    """A probe that only fails to measure the audio leaves the clip playable."""
495    renderer = _failing_probe(
496        "Unable to retrieve info for http://ha.invalid/api/tts_proxy/1.mp3 "
497        "(Invalid or unsupported media file)",
498        monkeypatch,
499    )
500
501    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
502
503    assert streamdetails.duration is None
504    assert renderer._sessions["sess"].skipped_sections == 0
505
506
507async def test_generate_script_uses_host_instructions() -> None:
508    """The prompt sent to the LLM carries the resolved host's persona instructions."""
509    renderer = DummyRenderer()
510    renderer._hosts = {"rick": {"id": "rick", "instructions": "Persona text.", "tts_engine": ""}}
511    captured: dict[str, str] = {}
512
513    async def fake_generate_text(
514        instructions: str,
515        prompt: str,  # noqa: ARG001
516        web_mode: str,  # noqa: ARG001
517        language: str | None = None,  # noqa: ARG001
518    ) -> str:
519        captured["instructions"] = instructions
520        return "script"
521
522    cast("Any", renderer)._generate_text = fake_generate_text
523    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
524
525    text = await renderer._generate_script(item, "p", "clip_1")
526
527    assert text == "script"
528    assert captured["instructions"] == "Persona text."
529
530
531async def test_generate_script_falls_back_to_default_instructions() -> None:
532    """A clip whose host is gone by render time still generates, using the default persona."""
533    renderer = DummyRenderer()
534    renderer._hosts = {}
535    captured: dict[str, str] = {}
536
537    async def fake_generate_text(
538        instructions: str,
539        prompt: str,  # noqa: ARG001
540        web_mode: str,  # noqa: ARG001
541        language: str | None = None,  # noqa: ARG001
542    ) -> str:
543        # mirrors the empty-to-default fallback the real _generate_text applies (runtime.py)
544        captured["instructions"] = instructions.strip() or DEFAULT_LLM_INSTRUCTIONS
545        return "script"
546
547    cast("Any", renderer)._generate_text = fake_generate_text
548    item = _clip_item("sess_001", **{ATTR_HOST_ID: "gone", ATTR_PROMPT: "p"})
549
550    await renderer._generate_script(item, "p", "clip_1")
551
552    assert captured["instructions"] == DEFAULT_LLM_INSTRUCTIONS
553
554
555async def test_generate_script_forwards_the_hosts_language() -> None:
556    """The host's configured language reaches _generate_text, ready to override the locale."""
557    renderer = DummyRenderer()
558    renderer._hosts = {
559        "rick": {
560            "id": "rick",
561            "instructions": "Persona text.",
562            "tts_engine": "",
563            "language": "fr_FR",
564        }
565    }
566    captured: dict[str, str | None] = {}
567
568    async def fake_generate_text(
569        instructions: str,  # noqa: ARG001
570        prompt: str,  # noqa: ARG001
571        web_mode: str,  # noqa: ARG001
572        language: str | None = None,
573    ) -> str:
574        captured["language"] = language
575        return "script"
576
577    cast("Any", renderer)._generate_text = fake_generate_text
578    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
579
580    await renderer._generate_script(item, "p", "clip_1")
581
582    assert captured["language"] == "fr_FR"
583
584
585async def test_generate_script_forwards_empty_language_when_host_has_none() -> None:
586    """A host with no configured language forwards an empty string, not None."""
587    renderer = DummyRenderer()
588    renderer._hosts = {"rick": {"id": "rick", "instructions": "Persona text.", "tts_engine": ""}}
589    captured: dict[str, str | None] = {}
590
591    async def fake_generate_text(
592        instructions: str,  # noqa: ARG001
593        prompt: str,  # noqa: ARG001
594        web_mode: str,  # noqa: ARG001
595        language: str | None = None,
596    ) -> str:
597        captured["language"] = language
598        return "script"
599
600    cast("Any", renderer)._generate_text = fake_generate_text
601    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
602
603    await renderer._generate_script(item, "p", "clip_1")
604
605    assert captured["language"] == ""
606
607
608async def test_render_tts_media_prefers_the_hosts_language_over_the_locale() -> None:
609    """A host's configured language reaches the TTS engine, overriding the server locale."""
610    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
611    renderer._hosts = {"rick": {"id": "rick", "tts_engine": "", "language": "fr_FR"}}
612    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
613
614    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
615
616    engine = cast("Any", renderer)._get_tts_engine.return_value
617    engine.provider.get_tts_message.assert_awaited_once_with(
618        "Good evening, it is warm out.", language="fr-FR", engine_id="tts.cloud", options={}
619    )
620
621
622async def test_render_tts_media_falls_back_to_locale_when_host_language_is_empty() -> None:
623    """A host with no configured language falls back to the server locale for the TTS call."""
624    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
625    renderer._hosts = {"rick": {"id": "rick", "tts_engine": ""}}
626    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
627
628    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
629
630    engine = cast("Any", renderer)._get_tts_engine.return_value
631    engine.provider.get_tts_message.assert_awaited_once_with(
632        "Good evening, it is warm out.", language="en-US", engine_id="tts.cloud", options={}
633    )
634
635
636async def test_resolve_deferred_placeholders_skips_weather_without_token() -> None:
637    """A prompt with no weather placeholder never triggers a weather fetch."""
638    renderer = DummyRenderer()
639
640    values = await renderer._resolve_deferred_placeholders("Just plain text, no tokens.")
641
642    assert renderer.weather_calls == 0
643    assert values["<weather_hourly>"] == ""
644    assert values["<weather_daily>"] == ""
645
646
647async def test_mint_clip_media_resolves_host_tts_engine() -> None:
648    """A clip whose host declares a tts_engine reaches _get_tts_engine with that override."""
649    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
650    renderer._hosts = {"rick": {"id": "rick", "tts_engine": "tts.rick_voice"}}
651    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
652    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
653
654    await renderer._mint_clip_media(item, "hello world", "clip_1")
655
656    cast("Any", renderer)._get_tts_engine.assert_awaited_once_with("tts.rick_voice")
657
658
659async def test_mint_clip_media_forwards_the_hosts_options() -> None:
660    """A host's configured TTS options are forwarded into the render call."""
661    renderer = DummyRenderer()
662    renderer._hosts = {
663        "rick": {
664            "id": "rick",
665            "tts_engine": "",
666            "options": {"voice": "en_US-lessac-medium", "length_scale": 1.2},
667        }
668    }
669    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
670    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
671
672    await renderer._mint_clip_media(item, "hello world", "clip_1")
673
674    assert renderer.tts_options == [{"voice": "en_US-lessac-medium", "length_scale": 1.2}]
675
676
677async def test_mint_clip_media_sends_no_options_for_a_host_without_any() -> None:
678    """A host with no configured options forwards an empty dict, not None."""
679    renderer = DummyRenderer()
680    renderer._hosts = {"rick": {"id": "rick", "tts_engine": ""}}
681    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
682    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
683
684    await renderer._mint_clip_media(item, "hello world", "clip_1")
685
686    assert renderer.tts_options == [{}]
687
688
689async def test_render_tts_media_forwards_the_hosts_tts_options() -> None:
690    """A host's configured TTS options reach the engine's get_tts_message call."""
691    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
692    renderer._hosts = {
693        "rick": {
694            "id": "rick",
695            "tts_engine": "",
696            "options": {"voice": "en_US-lessac-medium", "length_scale": 1.2},
697        }
698    }
699    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
700
701    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
702
703    engine = cast("Any", renderer)._get_tts_engine.return_value
704    engine.provider.get_tts_message.assert_awaited_once_with(
705        "Good evening, it is warm out.",
706        language="en-US",
707        engine_id="tts.cloud",
708        options={"voice": "en_US-lessac-medium", "length_scale": 1.2},
709    )
710
711
712async def test_render_tts_media_streams_a_url_over_http() -> None:
713    """A TTS engine returning a proxy URL yields an HTTP stream with the MP3 default format."""
714    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
715
716    path, stream_type, audio_format = await renderer._render_tts_media("hello world")
717
718    assert path == "http://example.test/api/tts_proxy/abc123.mp3"
719    assert stream_type == StreamType.HTTP
720    assert audio_format.content_type == ContentType.MP3
721    engine = cast("Any", renderer)._get_tts_engine.return_value
722    # the provider-scoped engine.id, never engine.uid, and never omitted
723    engine.provider.get_tts_message.assert_awaited_once_with(
724        "hello world", language=None, engine_id="tts.cloud", options=None
725    )
726
727
728async def test_render_tts_media_streams_a_local_file_from_disk(tmp_path: Path) -> None:
729    """A TTS engine that renders to disk is played as a local file rather than fetched."""
730    clip = tmp_path / "section.mp3"
731    clip.write_bytes(b"")
732    renderer = _tts_renderer(str(clip))
733    _attach_queue(renderer, [_clip_item("sess_001")])
734
735    path, stream_type, _ = await renderer._render_tts_media("hello world")
736
737    assert path == str(clip)
738    assert stream_type == StreamType.LOCAL_FILE
739
740
741async def test_render_tts_media_keeps_a_declared_audio_format(tmp_path: Path) -> None:
742    """A TTS engine that declares its own format has it carried into the clip streamdetails."""
743    clip = tmp_path / "section.wav"
744    clip.write_bytes(b"")
745    renderer = _tts_renderer(str(clip), AudioFormat(content_type=ContentType.WAV))
746
747    _, _, audio_format = await renderer._render_tts_media("hello world")
748
749    assert audio_format.content_type == ContentType.WAV
750
751
752@pytest.mark.parametrize("path", ["", "section.mp3", "/does/not/exist.mp3"])
753async def test_render_tts_media_rejects_an_unplayable_path(path: str) -> None:
754    """A path that is neither a URL nor an existing file fails loudly instead of degrading."""
755    renderer = _tts_renderer(path)
756
757    with pytest.raises(InvalidDataError, match="unusable stream path"):
758        await renderer._render_tts_media("hello world")
759
760
761async def test_render_tts_media_gives_up_on_a_stalled_engine(
762    monkeypatch: pytest.MonkeyPatch,
763) -> None:
764    """A stalled TTS engine fails the clip instead of pinning the render path."""
765    monkeypatch.setattr("music_assistant.helpers.tts.TTS_QUERY_TIMEOUT_SECONDS", 0.01)
766
767    async def _answers_too_late(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
768        await asyncio.sleep(5)
769        return SimpleNamespace(
770            path="http://example.test/late.mp3",
771            audio_format=AudioFormat(content_type=ContentType.MP3),
772        )
773
774    renderer = _tts_renderer("http://example.test/late.mp3")
775    engine = cast("Any", renderer)._get_tts_engine.return_value
776    engine.provider.get_tts_message = AsyncMock(side_effect=_answers_too_late)
777
778    with pytest.raises(MusicAssistantError, match="did not respond within"):
779        await renderer._render_tts_media("hello world")
780
781
782async def test_render_tts_media_reports_an_engine_side_timeout_as_is() -> None:
783    """A timeout raised by the TTS engine itself is not reported as our own cap."""
784    renderer = _tts_renderer("http://example.test/late.mp3")
785    engine = cast("Any", renderer)._get_tts_engine.return_value
786    engine.provider.get_tts_message = AsyncMock(side_effect=TimeoutError)
787
788    with pytest.raises(TimeoutError) as error:
789        await renderer._render_tts_media("hello world")
790    assert "did not respond within" not in str(error.value)
791
792
793async def test_local_file_clip_yields_local_file_streamdetails(tmp_path: Path) -> None:
794    """A disk-rendered clip reaches the core as LOCAL_FILE streamdetails."""
795    clip = tmp_path / "section.mp3"
796    clip.write_bytes(b"")
797    renderer = _tts_renderer(str(clip))
798    _attach_queue(renderer, [_clip_item("sess_001")])
799
800    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
801
802    assert streamdetails.stream_type == StreamType.LOCAL_FILE
803    assert streamdetails.path == str(clip)
804