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