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