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 collections.abc import AsyncGenerator
8from pathlib import Path
9from types import SimpleNamespace
10from typing import Any, cast
11from unittest.mock import AsyncMock, MagicMock
12
13import pytest
14from music_assistant_models.enums import (
15    ContentType,
16    MediaType,
17    StreamType,
18    VolumeNormalizationMode,
19)
20from music_assistant_models.errors import (
21    InvalidDataError,
22    MediaNotFoundError,
23    MusicAssistantError,
24)
25from music_assistant_models.media_items import AudioFormat, ProviderMapping, SoundEffect
26from music_assistant_models.queue_item import QueueItem
27
28from music_assistant.constants import (
29    CONF_VALUE_DISABLED,
30    CONF_VALUE_ENABLED,
31    CONF_VOLUME_NORMALIZATION,
32    CONF_VOLUME_NORMALIZATION_TARGET,
33    CONF_VOLUME_NORMALIZATION_TRACKS,
34)
35from music_assistant.helpers.tags import AudioTags
36from music_assistant.models.plugin import PluginProvider, TTSEngine
37from music_assistant.providers.ai_radio.constants import (
38    ATTR_HOST_ID,
39    ATTR_MAX_CHARS,
40    ATTR_PROMPT,
41    ATTR_RENDERED_TEXT,
42    ATTR_SESSION_ID,
43    ATTR_STATION_ID,
44    ATTR_WEB_SEARCH_MODE,
45    CLIP_STREAMDETAILS_EXPIRATION,
46    CONF_TTS_LOUDNESS_BOOST,
47    DEFAULT_LLM_INSTRUCTIONS,
48    MIN_LOUDNESS_REFERENCE_SECONDS,
49    TTS_CLIP_PCM_FORMAT,
50    TTS_PEAK_CEILING_DB,
51    TTS_SPEECHNORM_FILTER,
52)
53from music_assistant.providers.ai_radio.models import SessionState
54from music_assistant.providers.ai_radio.rendering import AIRadioRenderMixin
55
56
57class DummyRenderer(AIRadioRenderMixin):
58    """Minimal harness exposing the render path."""
59
60    domain = "ai_radio"
61    instance_id = "ai_radio--test"
62
63    def __init__(self) -> None:
64        """Initialize the harness with recording stubs."""
65        self.logger = logging.getLogger("tests.ai_radio.rendering")
66        self._sessions: dict[str, Any] = {}
67        self._hosts: dict[str, dict[str, Any]] = {}
68        self.llm_prompts: list[str] = []
69        self.tts_texts: list[str] = []
70        self.tts_options: list[dict[str, Any] | None] = []
71        self.weather_calls = 0
72        self.fail_generation = False
73        self.measure_calls: list[str] = []
74        self.measured_loudness: float | None = None
75
76    def _configured_now(self) -> Any:
77        return __import__("datetime").datetime(2026, 7, 30, 18, 30)
78
79    async def _generate_text(
80        self, instructions: str, prompt: str, web_mode: str, language: str | None = None
81    ) -> str:
82        # a real suspension point so concurrent callers actually interleave under
83        # asyncio.gather, otherwise the lock in get_stream_details is never exercised
84        await asyncio.sleep(0)
85        if self.fail_generation:
86            raise RuntimeError("llm down")
87        self.llm_prompts.append(prompt)
88        return "Good evening, it is warm out."
89
90    async def _prepare_weather_tokens(self) -> dict[str, str]:
91        self.weather_calls += 1
92        return {"<weather_hourly>": f"fresh weather {self.weather_calls}"}
93
94    async def _render_tts_media(
95        self,
96        text: str,
97        engine_uid: str | None = None,
98        language: str | None = None,
99        options: dict[str, Any] | None = None,
100    ) -> tuple[str, StreamType, AudioFormat]:
101        self.tts_texts.append(text)
102        self.tts_options.append(options)
103        return (
104            f"http://ha.invalid/api/tts_proxy/{len(self.tts_texts)}.mp3",
105            StreamType.HTTP,
106            AudioFormat(content_type=ContentType.MP3),
107        )
108
109    async def _probe_duration(self, path: str) -> int | None:
110        return 9
111
112    async def _measure_loudness(self, path: str) -> float | None:
113        self.measure_calls.append(path)
114        return self.measured_loudness
115
116
117class RealTtsRenderer(DummyRenderer):
118    """Harness that exercises the mixin's real TTS path instead of the DummyRenderer stub."""
119
120    _render_tts_media = AIRadioRenderMixin._render_tts_media
121
122
123def _tts_renderer(path: str, audio_format: AudioFormat | None = None) -> RealTtsRenderer:
124    """Build a renderer whose TTS engine returns StreamDetails carrying the given path."""
125    renderer = RealTtsRenderer()
126    plugin = MagicMock(spec=PluginProvider)
127    plugin.instance_id = "hass_1"
128    plugin.get_tts_message = AsyncMock(
129        return_value=SimpleNamespace(
130            path=path, audio_format=audio_format or AudioFormat(content_type=ContentType.UNKNOWN)
131        )
132    )
133    engine = TTSEngine(id="tts.cloud", name="Cloud", provider=plugin)
134    cast("Any", renderer)._get_tts_engine = AsyncMock(return_value=engine)
135    return renderer
136
137
138def _clip_item(clip_id: str, queue_id: str = "player_a", **overrides: Any) -> QueueItem:
139    """Build a queue item for a pending AI Radio clip."""
140    attributes: dict[str, Any] = {
141        ATTR_SESSION_ID: "sess",
142        ATTR_STATION_ID: "st",
143        ATTR_PROMPT: "It is <timestamp>. Weather: <weather_hourly>.",
144        ATTR_MAX_CHARS: 300,
145        ATTR_WEB_SEARCH_MODE: "disabled",
146    }
147    attributes.update(overrides)
148    media_item = SoundEffect(
149        item_id=clip_id,
150        provider="ai_radio--test",
151        name="Weather",
152        provider_mappings={
153            ProviderMapping(
154                item_id=clip_id,
155                provider_domain="ai_radio",
156                provider_instance="ai_radio--test",
157            )
158        },
159    )
160    return QueueItem(
161        queue_id=queue_id,
162        queue_item_id=f"qi_{clip_id}",
163        name="Weather",
164        duration=None,
165        media_item=media_item,
166        extra_attributes=attributes,
167    )
168
169
170def _attach_queues(renderer: DummyRenderer, queues: dict[str, list[QueueItem]]) -> list[bool]:
171    """Wire a minimal player_queues stub and return the signal_update call log."""
172    signals: list[bool] = []
173    cast("Any", renderer).mass = SimpleNamespace(
174        player_queues=SimpleNamespace(
175            all=lambda: tuple(SimpleNamespace(queue_id=queue_id) for queue_id in queues),
176            items=lambda queue_id, limit=500, offset=0: queues.get(queue_id, [])[
177                offset : offset + limit
178            ],
179            signal_update=lambda _queue_id, items_changed=False: signals.append(items_changed),
180        ),
181        metadata=SimpleNamespace(locale="en_US"),
182    )
183    _attach_normalization(renderer, queue_ids=tuple(queues))
184    return signals
185
186
187def _attach_queue(renderer: DummyRenderer, items: list[QueueItem]) -> list[bool]:
188    """Wire a single-queue player_queues stub and return the signal_update call log."""
189    return _attach_queues(renderer, {"player_a": items})
190
191
192def _attach_normalization(
193    renderer: DummyRenderer,
194    *,
195    enabled: bool = True,
196    target: int = -14,
197    boost: int = 3,
198    tracks_mode: str = VolumeNormalizationMode.FALLBACK_DYNAMIC.value,
199    queue_ids: tuple[str, ...] = ("player_a",),
200) -> None:
201    """Wire the queue, streams and provider config that decide the clip's loudness gain."""
202
203    def queue_setting(queue_id: str, key: str, default: str) -> str:
204        assert queue_id in queue_ids
205        assert key == CONF_VOLUME_NORMALIZATION
206        assert default == CONF_VALUE_ENABLED
207        return CONF_VALUE_ENABLED if enabled else CONF_VALUE_DISABLED
208
209    def streams_setting(key: str, **_kwargs: Any) -> str | int:
210        if key == CONF_VOLUME_NORMALIZATION_TRACKS:
211            return tracks_mode
212        assert key == CONF_VOLUME_NORMALIZATION_TARGET
213        return target
214
215    def provider_setting(key: str) -> int:
216        assert key == CONF_TTS_LOUDNESS_BOOST
217        return boost
218
219    mass = cast("Any", renderer).mass
220    mass.config = SimpleNamespace(get_effective_player_queue_config_value=queue_setting)
221    mass.streams = SimpleNamespace(get_config_value=streams_setting)
222    cast("Any", renderer).config = SimpleNamespace(get_value=provider_setting)
223
224
225async def test_render_resolves_deferred_placeholders_at_render_time() -> None:
226    """The prompt sent to the LLM carries render-time weather, not plan-time."""
227    renderer = DummyRenderer()
228    item = _clip_item("sess_001")
229    _attach_queue(renderer, [item])
230
231    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
232
233    assert renderer.weather_calls == 1
234    assert "fresh weather 1" in renderer.llm_prompts[0]
235    assert "<timestamp>" not in renderer.llm_prompts[0]
236    assert streamdetails.media_type == MediaType.SOUND_EFFECT
237    assert streamdetails.stream_type == StreamType.HTTP
238    assert streamdetails.duration == 9
239    assert streamdetails.expiration == 60
240    assert streamdetails.can_seek is False
241    assert streamdetails.allow_seek is False
242
243
244async def test_render_caches_the_script_and_the_minted_media() -> None:
245    """A second render within the cache window reuses both the stored script and media."""
246    renderer = DummyRenderer()
247    item = _clip_item("sess_001")
248    signals = _attach_queue(renderer, [item])
249
250    first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
251    second = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
252
253    assert len(renderer.llm_prompts) == 1
254    assert renderer.tts_texts == ["Good evening, it is warm out."]
255    assert first.path == second.path
256    assert item.extra_attributes[ATTR_RENDERED_TEXT] == "Good evening, it is warm out."
257    assert signals == [True]
258
259
260async def test_concurrent_renders_call_the_llm_once() -> None:
261    """Two simultaneous requests for one clip render a single script."""
262    renderer = DummyRenderer()
263    _attach_queue(renderer, [_clip_item("sess_001")])
264
265    await asyncio.gather(
266        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
267        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
268    )
269
270    assert len(renderer.llm_prompts) == 1
271
272
273async def test_concurrent_renders_mint_the_clip_only_once() -> None:
274    """Three simultaneous requests for one clip share a single minted TTS render."""
275    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
276    _attach_queue(renderer, [_clip_item("sess_001")])
277
278    results = await asyncio.gather(
279        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
280        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
281        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
282    )
283
284    engine = cast("Any", renderer)._get_tts_engine.return_value
285    engine.provider.get_tts_message.assert_awaited_once()
286    assert len({result.path for result in results}) == 1
287
288
289async def test_cached_media_remints_once_it_expires() -> None:
290    """A render requested after the cache window elapses mints a fresh clip."""
291    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
292    _attach_queue(renderer, [_clip_item("sess_001")])
293
294    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
295    cached = cast("Any", renderer)._media_cache["sess_001"]
296    cached.minted_at -= CLIP_STREAMDETAILS_EXPIRATION + 1
297
298    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
299
300    engine = cast("Any", renderer)._get_tts_engine.return_value
301    assert engine.provider.get_tts_message.await_count == 2
302
303
304async def test_a_late_cache_hit_expires_with_the_url_it_serves() -> None:
305    """A hit late in the window hands out the url's remaining life, not a fresh full window."""
306    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
307    _attach_queue(renderer, [_clip_item("sess_001")])
308
309    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
310    cached = cast("Any", renderer)._media_cache["sess_001"]
311    cached.minted_at -= 45
312
313    late = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
314
315    engine = cast("Any", renderer)._get_tts_engine.return_value
316    assert engine.provider.get_tts_message.await_count == 1
317    assert 14 <= late.expiration <= 15
318
319
320async def test_a_cache_hit_with_no_useful_life_left_remints() -> None:
321    """A hit in the last seconds of the window mints again instead of serving a dying url."""
322    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
323    _attach_queue(renderer, [_clip_item("sess_001")])
324
325    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
326    cached = cast("Any", renderer)._media_cache["sess_001"]
327    cached.minted_at -= CLIP_STREAMDETAILS_EXPIRATION - 2
328
329    fresh = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
330
331    engine = cast("Any", renderer)._get_tts_engine.return_value
332    assert engine.provider.get_tts_message.await_count == 2
333    assert fresh.expiration == CLIP_STREAMDETAILS_EXPIRATION
334
335
336async def test_expired_cache_entries_are_pruned_on_the_next_mint() -> None:
337    """Minting a clip drops the entries whose urls died, so the cache cannot grow forever."""
338    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
339    _attach_queue(renderer, [_clip_item("sess_001"), _clip_item("sess_002")])
340
341    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
342    media_cache = cast("Any", renderer)._media_cache
343    media_cache["sess_001"].minted_at -= CLIP_STREAMDETAILS_EXPIRATION + 1
344    await renderer.get_stream_details("sess_002", MediaType.SOUND_EFFECT)
345
346    assert set(media_cache) == {"sess_002"}
347
348
349async def test_render_tts_media_passes_the_locale_as_language() -> None:
350    """The DJ script's locale reaches the TTS engine as a hyphenated language code."""
351    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
352    _attach_queue(renderer, [_clip_item("sess_001")])
353
354    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
355
356    engine = cast("Any", renderer)._get_tts_engine.return_value
357    engine.provider.get_tts_message.assert_awaited_once_with(
358        "Good evening, it is warm out.", language="en-US", engine_id="tts.cloud", options={}
359    )
360
361
362async def test_render_tts_media_falls_back_without_language_on_rejection() -> None:
363    """An engine that rejects the requested language is retried once without it."""
364    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
365    _attach_queue(renderer, [_clip_item("sess_001")])
366    engine = cast("Any", renderer)._get_tts_engine.return_value
367    engine.provider.get_tts_message = AsyncMock(
368        side_effect=[
369            Exception("unsupported language"),
370            SimpleNamespace(
371                path="http://example.test/api/tts_proxy/abc123.mp3",
372                audio_format=AudioFormat(content_type=ContentType.MP3),
373            ),
374        ]
375    )
376
377    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
378
379    assert streamdetails.path == "http://example.test/api/tts_proxy/abc123.mp3"
380    assert engine.provider.get_tts_message.await_count == 2
381    first_call, second_call = engine.provider.get_tts_message.await_args_list
382    assert first_call.kwargs["language"] == "en-US"
383    assert second_call.kwargs["language"] is None
384
385
386async def test_render_tts_media_does_not_retry_after_a_timeout_style_failure() -> None:
387    """A structured MusicAssistantError is not a language rejection, so it skips the retry."""
388    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
389    _attach_queue(renderer, [_clip_item("sess_001")])
390    engine = cast("Any", renderer)._get_tts_engine.return_value
391    engine.provider.get_tts_message = AsyncMock(
392        side_effect=MusicAssistantError("engine did not respond within 5s")
393    )
394
395    with pytest.raises(MediaNotFoundError):
396        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
397
398    engine.provider.get_tts_message.assert_awaited_once()
399
400
401async def test_clip_is_found_in_the_owning_sessions_queue() -> None:
402    """The session registry points the lookup at the queue that holds the clip."""
403    renderer = DummyRenderer()
404    session = SessionState(session_id="sess", station_id="st", queue_id="player_b")
405    renderer._sessions = {"sess": session}
406    _attach_queues(
407        renderer,
408        {"player_a": [], "player_b": [_clip_item("sess_001", queue_id="player_b")]},
409    )
410
411    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
412
413    assert streamdetails.item_id == "sess_001"
414
415
416async def test_clip_is_found_by_scanning_every_queue_after_a_restart() -> None:
417    """With the session registry gone, the clip is still located in its persisted queue."""
418    renderer = DummyRenderer()
419    _attach_queues(
420        renderer,
421        {"player_a": [], "player_b": [_clip_item("sess_001", queue_id="player_b")]},
422    )
423
424    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
425
426    assert streamdetails.item_id == "sess_001"
427
428
429async def test_unknown_clip_raises_media_not_found() -> None:
430    """A clip id that is not in any queue is reported as missing media."""
431    renderer = DummyRenderer()
432    _attach_queue(renderer, [_clip_item("sess_001")])
433
434    with pytest.raises(MediaNotFoundError):
435        await renderer.get_stream_details("sess_999", MediaType.SOUND_EFFECT)
436
437
438async def test_clip_without_prompt_raises_media_not_found() -> None:
439    """A clip whose attributes were lost is reported as missing media."""
440    renderer = DummyRenderer()
441    item = _clip_item("sess_001")
442    item.extra_attributes.pop(ATTR_PROMPT)
443    _attach_queue(renderer, [item])
444
445    with pytest.raises(MediaNotFoundError):
446        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
447
448
449async def test_llm_failure_raises_media_not_found() -> None:
450    """An LLM failure surfaces as missing media so the core skips the clip."""
451    renderer = DummyRenderer()
452    renderer.fail_generation = True
453    _attach_queue(renderer, [_clip_item("sess_001")])
454
455    with pytest.raises(MediaNotFoundError):
456        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
457
458
459async def test_render_failure_increments_the_session_skip_counter() -> None:
460    """A skipped clip is recorded on its session without failing the run."""
461    renderer = DummyRenderer()
462    session = SessionState(session_id="sess", station_id="st")
463    renderer._sessions = {"sess": session}
464    _attach_queue(renderer, [_clip_item("sess_001")])
465    renderer.fail_generation = True
466
467    with pytest.raises(MediaNotFoundError):
468        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
469
470    assert session.skipped_sections == 1
471    assert session.last_render_error
472
473
474async def test_tts_failure_raises_media_not_found_and_records_skip() -> None:
475    """A TTS failure surfaces as missing media and is recorded on the owning session."""
476
477    class UnspeakableRenderer(DummyRenderer):
478        async def _render_tts_media(
479            self,
480            text: str,
481            engine_uid: str | None = None,
482            language: str | None = None,
483            options: dict[str, Any] | None = None,
484        ) -> tuple[str, StreamType, AudioFormat]:
485            raise RuntimeError("tts down")
486
487    renderer = UnspeakableRenderer()
488    session = SessionState(session_id="sess", station_id="st")
489    renderer._sessions = {"sess": session}
490    _attach_queue(renderer, [_clip_item("sess_001")])
491
492    with pytest.raises(MediaNotFoundError):
493        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
494
495    assert session.skipped_sections == 1
496    assert session.last_render_error
497
498
499async def test_probe_failure_is_not_fatal() -> None:
500    """A failed duration probe yields streamdetails without a duration."""
501
502    class UnprobableRenderer(DummyRenderer):
503        async def _probe_duration(self, path: str) -> int | None:
504            return None
505
506    renderer = UnprobableRenderer()
507    _attach_queue(renderer, [_clip_item("sess_001")])
508
509    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
510
511    assert streamdetails.duration is None
512
513
514class RealProbeRenderer(DummyRenderer):
515    """Harness that exercises the mixin's real duration probe instead of the stub."""
516
517    _probe_duration = AIRadioRenderMixin._probe_duration
518
519
520def _failing_probe(message: str, monkeypatch: pytest.MonkeyPatch) -> RealProbeRenderer:
521    """Build a renderer whose duration probe fails with the given ffprobe message."""
522
523    async def _raise(*_args: Any, **_kwargs: Any) -> AudioTags:
524        raise InvalidDataError(message)
525
526    monkeypatch.setattr("music_assistant.providers.ai_radio.rendering.async_parse_tags", _raise)
527    renderer = RealProbeRenderer()
528    renderer._sessions = {"sess": SessionState(session_id="sess", station_id="st")}
529    _attach_queue(renderer, [_clip_item("sess_001")])
530    return renderer
531
532
533@pytest.mark.parametrize(
534    "server_error",
535    ["Server returned 5XX Server Error reply", "HTTP error 500 Internal Server Error"],
536)
537async def test_tts_server_error_fails_the_clip_with_an_actionable_message(
538    server_error: str, monkeypatch: pytest.MonkeyPatch
539) -> None:
540    """An engine that hands out a URL it cannot render fails the clip, not the playback."""
541    renderer = _failing_probe(
542        f"Unable to retrieve info for http://ha.invalid/api/tts_proxy/1.mp3 ({server_error})",
543        monkeypatch,
544    )
545
546    with pytest.raises(MediaNotFoundError):
547        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
548
549    session = renderer._sessions["sess"]
550    assert session.skipped_sections == 1
551    assert "enough credit" in session.last_render_error
552    # the hint is a guess, so the whole probe message travels with it - the url included,
553    # since that is what tells a failing engine apart from a failing tts server behind it
554    assert "http://ha.invalid/api/tts_proxy/1.mp3" in session.last_render_error
555    assert server_error in session.last_render_error
556
557
558async def test_unmeasurable_clip_still_plays(monkeypatch: pytest.MonkeyPatch) -> None:
559    """A probe that only fails to measure the audio leaves the clip playable."""
560    renderer = _failing_probe(
561        "Unable to retrieve info for http://ha.invalid/api/tts_proxy/1.mp3 "
562        "(Invalid or unsupported media file)",
563        monkeypatch,
564    )
565
566    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
567
568    assert streamdetails.duration is None
569    assert renderer._sessions["sess"].skipped_sections == 0
570
571
572async def test_generate_script_uses_host_instructions() -> None:
573    """The prompt sent to the LLM carries the resolved host's persona instructions."""
574    renderer = DummyRenderer()
575    renderer._hosts = {"rick": {"id": "rick", "instructions": "Persona text.", "tts_engine": ""}}
576    captured: dict[str, str] = {}
577
578    async def fake_generate_text(
579        instructions: str,
580        prompt: str,  # noqa: ARG001
581        web_mode: str,  # noqa: ARG001
582        language: str | None = None,  # noqa: ARG001
583    ) -> str:
584        captured["instructions"] = instructions
585        return "script"
586
587    cast("Any", renderer)._generate_text = fake_generate_text
588    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
589
590    text = await renderer._generate_script(item, "p", "clip_1")
591
592    assert text == "script"
593    assert captured["instructions"] == "Persona text."
594
595
596async def test_generate_script_falls_back_to_default_instructions() -> None:
597    """A clip whose host is gone by render time still generates, using the default persona."""
598    renderer = DummyRenderer()
599    renderer._hosts = {}
600    captured: dict[str, str] = {}
601
602    async def fake_generate_text(
603        instructions: str,
604        prompt: str,  # noqa: ARG001
605        web_mode: str,  # noqa: ARG001
606        language: str | None = None,  # noqa: ARG001
607    ) -> str:
608        # mirrors the empty-to-default fallback the real _generate_text applies (runtime.py)
609        captured["instructions"] = instructions.strip() or DEFAULT_LLM_INSTRUCTIONS
610        return "script"
611
612    cast("Any", renderer)._generate_text = fake_generate_text
613    item = _clip_item("sess_001", **{ATTR_HOST_ID: "gone", ATTR_PROMPT: "p"})
614
615    await renderer._generate_script(item, "p", "clip_1")
616
617    assert captured["instructions"] == DEFAULT_LLM_INSTRUCTIONS
618
619
620async def test_generate_script_forwards_the_hosts_language() -> None:
621    """The host's configured language reaches _generate_text, ready to override the locale."""
622    renderer = DummyRenderer()
623    renderer._hosts = {
624        "rick": {
625            "id": "rick",
626            "instructions": "Persona text.",
627            "tts_engine": "",
628            "language": "fr_FR",
629        }
630    }
631    captured: dict[str, str | None] = {}
632
633    async def fake_generate_text(
634        instructions: str,  # noqa: ARG001
635        prompt: str,  # noqa: ARG001
636        web_mode: str,  # noqa: ARG001
637        language: str | None = None,
638    ) -> str:
639        captured["language"] = language
640        return "script"
641
642    cast("Any", renderer)._generate_text = fake_generate_text
643    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
644
645    await renderer._generate_script(item, "p", "clip_1")
646
647    assert captured["language"] == "fr_FR"
648
649
650async def test_generate_script_forwards_empty_language_when_host_has_none() -> None:
651    """A host with no configured language forwards an empty string, not None."""
652    renderer = DummyRenderer()
653    renderer._hosts = {"rick": {"id": "rick", "instructions": "Persona text.", "tts_engine": ""}}
654    captured: dict[str, str | None] = {}
655
656    async def fake_generate_text(
657        instructions: str,  # noqa: ARG001
658        prompt: str,  # noqa: ARG001
659        web_mode: str,  # noqa: ARG001
660        language: str | None = None,
661    ) -> str:
662        captured["language"] = language
663        return "script"
664
665    cast("Any", renderer)._generate_text = fake_generate_text
666    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
667
668    await renderer._generate_script(item, "p", "clip_1")
669
670    assert captured["language"] == ""
671
672
673async def test_render_tts_media_prefers_the_hosts_language_over_the_locale() -> None:
674    """A host's configured language reaches the TTS engine, overriding the server locale."""
675    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
676    renderer._hosts = {"rick": {"id": "rick", "tts_engine": "", "language": "fr_FR"}}
677    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
678
679    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
680
681    engine = cast("Any", renderer)._get_tts_engine.return_value
682    engine.provider.get_tts_message.assert_awaited_once_with(
683        "Good evening, it is warm out.", language="fr-FR", engine_id="tts.cloud", options={}
684    )
685
686
687async def test_render_tts_media_falls_back_to_locale_when_host_language_is_empty() -> None:
688    """A host with no configured language falls back to the server locale for the TTS call."""
689    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
690    renderer._hosts = {"rick": {"id": "rick", "tts_engine": ""}}
691    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
692
693    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
694
695    engine = cast("Any", renderer)._get_tts_engine.return_value
696    engine.provider.get_tts_message.assert_awaited_once_with(
697        "Good evening, it is warm out.", language="en-US", engine_id="tts.cloud", options={}
698    )
699
700
701async def test_resolve_deferred_placeholders_skips_weather_without_token() -> None:
702    """A prompt with no weather placeholder never triggers a weather fetch."""
703    renderer = DummyRenderer()
704
705    values = await renderer._resolve_deferred_placeholders("Just plain text, no tokens.")
706
707    assert renderer.weather_calls == 0
708    assert values["<weather_hourly>"] == ""
709    assert values["<weather_daily>"] == ""
710
711
712async def test_mint_clip_media_resolves_host_tts_engine() -> None:
713    """A clip whose host declares a tts_engine reaches _get_tts_engine with that override."""
714    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
715    renderer._hosts = {"rick": {"id": "rick", "tts_engine": "tts.rick_voice"}}
716    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
717    _attach_normalization(renderer)
718    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
719
720    await renderer._mint_clip_media(item, "hello world", "clip_1")
721
722    cast("Any", renderer)._get_tts_engine.assert_awaited_once_with("tts.rick_voice")
723
724
725async def test_mint_clip_media_forwards_the_hosts_options() -> None:
726    """A host's configured TTS options are forwarded into the render call."""
727    renderer = DummyRenderer()
728    renderer._hosts = {
729        "rick": {
730            "id": "rick",
731            "tts_engine": "",
732            "options": {"voice": "en_US-lessac-medium", "length_scale": 1.2},
733        }
734    }
735    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
736    _attach_normalization(renderer)
737    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
738
739    await renderer._mint_clip_media(item, "hello world", "clip_1")
740
741    assert renderer.tts_options == [{"voice": "en_US-lessac-medium", "length_scale": 1.2}]
742
743
744async def test_mint_clip_media_sends_no_options_for_a_host_without_any() -> None:
745    """A host with no configured options forwards an empty dict, not None."""
746    renderer = DummyRenderer()
747    renderer._hosts = {"rick": {"id": "rick", "tts_engine": ""}}
748    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
749    _attach_normalization(renderer)
750    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
751
752    await renderer._mint_clip_media(item, "hello world", "clip_1")
753
754    assert renderer.tts_options == [{}]
755
756
757async def test_render_tts_media_forwards_the_hosts_tts_options() -> None:
758    """A host's configured TTS options reach the engine's get_tts_message call."""
759    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
760    renderer._hosts = {
761        "rick": {
762            "id": "rick",
763            "tts_engine": "",
764            "options": {"voice": "en_US-lessac-medium", "length_scale": 1.2},
765        }
766    }
767    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
768
769    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
770
771    engine = cast("Any", renderer)._get_tts_engine.return_value
772    engine.provider.get_tts_message.assert_awaited_once_with(
773        "Good evening, it is warm out.",
774        language="en-US",
775        engine_id="tts.cloud",
776        options={"voice": "en_US-lessac-medium", "length_scale": 1.2},
777    )
778
779
780async def test_render_tts_media_streams_a_url_over_http() -> None:
781    """A TTS engine returning a proxy URL yields an HTTP stream with the MP3 default format."""
782    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
783
784    path, stream_type, audio_format = await renderer._render_tts_media("hello world")
785
786    assert path == "http://example.test/api/tts_proxy/abc123.mp3"
787    assert stream_type == StreamType.HTTP
788    assert audio_format.content_type == ContentType.MP3
789    engine = cast("Any", renderer)._get_tts_engine.return_value
790    # the provider-scoped engine.id, never engine.uid, and never omitted
791    engine.provider.get_tts_message.assert_awaited_once_with(
792        "hello world", language=None, engine_id="tts.cloud", options=None
793    )
794
795
796async def test_render_tts_media_streams_a_local_file_from_disk(tmp_path: Path) -> None:
797    """A TTS engine that renders to disk is played as a local file rather than fetched."""
798    clip = tmp_path / "section.mp3"
799    clip.write_bytes(b"")
800    renderer = _tts_renderer(str(clip))
801    _attach_queue(renderer, [_clip_item("sess_001")])
802
803    path, stream_type, _ = await renderer._render_tts_media("hello world")
804
805    assert path == str(clip)
806    assert stream_type == StreamType.LOCAL_FILE
807
808
809async def test_render_tts_media_keeps_a_declared_audio_format(tmp_path: Path) -> None:
810    """A TTS engine that declares its own format has it carried into the clip streamdetails."""
811    clip = tmp_path / "section.wav"
812    clip.write_bytes(b"")
813    renderer = _tts_renderer(str(clip), AudioFormat(content_type=ContentType.WAV))
814
815    _, _, audio_format = await renderer._render_tts_media("hello world")
816
817    assert audio_format.content_type == ContentType.WAV
818
819
820@pytest.mark.parametrize("path", ["", "section.mp3", "/does/not/exist.mp3"])
821async def test_render_tts_media_rejects_an_unplayable_path(path: str) -> None:
822    """A path that is neither a URL nor an existing file fails loudly instead of degrading."""
823    renderer = _tts_renderer(path)
824
825    with pytest.raises(InvalidDataError, match="unusable stream path"):
826        await renderer._render_tts_media("hello world")
827
828
829async def test_render_tts_media_gives_up_on_a_stalled_engine(
830    monkeypatch: pytest.MonkeyPatch,
831) -> None:
832    """A stalled TTS engine fails the clip instead of pinning the render path."""
833    monkeypatch.setattr("music_assistant.helpers.tts.TTS_QUERY_TIMEOUT_SECONDS", 0.01)
834
835    async def _answers_too_late(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
836        await asyncio.sleep(5)
837        return SimpleNamespace(
838            path="http://example.test/late.mp3",
839            audio_format=AudioFormat(content_type=ContentType.MP3),
840        )
841
842    renderer = _tts_renderer("http://example.test/late.mp3")
843    engine = cast("Any", renderer)._get_tts_engine.return_value
844    engine.provider.get_tts_message = AsyncMock(side_effect=_answers_too_late)
845
846    with pytest.raises(MusicAssistantError, match="did not respond within"):
847        await renderer._render_tts_media("hello world")
848
849
850async def test_render_tts_media_reports_an_engine_side_timeout_as_is() -> None:
851    """A timeout raised by the TTS engine itself is not reported as our own cap."""
852    renderer = _tts_renderer("http://example.test/late.mp3")
853    engine = cast("Any", renderer)._get_tts_engine.return_value
854    engine.provider.get_tts_message = AsyncMock(side_effect=TimeoutError)
855
856    with pytest.raises(TimeoutError) as error:
857        await renderer._render_tts_media("hello world")
858    assert "did not respond within" not in str(error.value)
859
860
861async def test_local_file_clip_yields_local_file_streamdetails(tmp_path: Path) -> None:
862    """A disk-rendered clip reaches the core as LOCAL_FILE streamdetails."""
863    clip = tmp_path / "section.mp3"
864    clip.write_bytes(b"")
865    renderer = _tts_renderer(str(clip))
866    _attach_queue(renderer, [_clip_item("sess_001")])
867
868    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
869
870    assert streamdetails.stream_type == StreamType.LOCAL_FILE
871    assert streamdetails.path == str(clip)
872
873
874async def test_a_measured_clip_is_lifted_to_the_target_plus_the_boost() -> None:
875    """A clip quieter than the levelled music is served through the provider's own chain."""
876    renderer = DummyRenderer()
877    renderer.measured_loudness = -18.0
878    _attach_queue(renderer, [_clip_item("sess_001")])
879    _attach_normalization(renderer, target=-14, boost=3)
880
881    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
882
883    assert streamdetails.stream_type == StreamType.CUSTOM
884    assert streamdetails.decoded_audio_format == TTS_CLIP_PCM_FORMAT
885    assert streamdetails.audio_format.content_type == ContentType.MP3
886    assert streamdetails.data.gain_db == pytest.approx(7.0)
887
888
889async def test_the_clip_is_evened_out_and_lifted_before_it_is_limited(
890    monkeypatch: pytest.MonkeyPatch,
891) -> None:
892    """The filter chain reaches ffmpeg as speechnorm, then gain, then the peak limiter."""
893    captured: dict[str, Any] = {}
894
895    async def fake_ffmpeg_stream(**kwargs: Any) -> AsyncGenerator[bytes]:
896        captured.update(kwargs)
897        yield b"pcm"
898
899    monkeypatch.setattr(
900        "music_assistant.providers.ai_radio.rendering.get_ffmpeg_stream", fake_ffmpeg_stream
901    )
902    renderer = DummyRenderer()
903    renderer.measured_loudness = -18.0
904    _attach_queue(renderer, [_clip_item("sess_001")])
905    _attach_normalization(renderer, target=-14, boost=3)
906    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
907
908    chunks = [chunk async for chunk in renderer.get_audio_stream(streamdetails)]
909
910    assert chunks == [b"pcm"]
911    assert captured["audio_input"] == streamdetails.path
912    assert captured["input_format"].content_type == ContentType.MP3
913    assert captured["output_format"] == TTS_CLIP_PCM_FORMAT
914    assert captured["filter_params"] == [
915        TTS_SPEECHNORM_FILTER,
916        "volume=7.0dB",
917        f"alimiter=limit={TTS_PEAK_CEILING_DB}dB:level=false:latency=true",
918    ]
919
920
921async def test_the_engine_reference_is_measured_once_and_reused() -> None:
922    """A second clip from the same voice levels itself against the stored measurement."""
923    renderer = DummyRenderer()
924    renderer.measured_loudness = -18.0
925    _attach_queue(renderer, [_clip_item("sess_001"), _clip_item("sess_002")])
926    _attach_normalization(renderer)
927
928    first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
929    second = await renderer.get_stream_details("sess_002", MediaType.SOUND_EFFECT)
930
931    assert len(renderer.measure_calls) == 1
932    assert first.data.gain_db == second.data.gain_db
933
934
935async def test_a_short_clip_levels_itself_but_never_becomes_the_reference() -> None:
936    """A clip of a few words is too thin a sample to speak for the rest of the engine."""
937
938    class BriefRenderer(DummyRenderer):
939        async def _probe_duration(self, path: str) -> int | None:
940            return MIN_LOUDNESS_REFERENCE_SECONDS - 1
941
942    renderer = BriefRenderer()
943    renderer.measured_loudness = -18.0
944    _attach_queue(renderer, [_clip_item("sess_001"), _clip_item("sess_002")])
945    _attach_normalization(renderer)
946
947    first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
948    await renderer.get_stream_details("sess_002", MediaType.SOUND_EFFECT)
949
950    assert first.stream_type == StreamType.CUSTOM
951    assert len(renderer.measure_calls) == 2
952    assert cast("Any", renderer)._engine_loudness == {}
953
954
955async def test_clip_plays_untouched_when_the_queue_does_not_normalize() -> None:
956    """With the music unlevelled there is nothing to match, so the clip airs as rendered."""
957    renderer = DummyRenderer()
958    renderer.measured_loudness = -18.0
959    _attach_queue(renderer, [_clip_item("sess_001")])
960    _attach_normalization(renderer, enabled=False)
961
962    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
963
964    assert streamdetails.stream_type == StreamType.HTTP
965    assert streamdetails.decoded_audio_format is None
966    assert streamdetails.data is None
967
968
969async def test_clip_plays_untouched_when_it_could_not_be_measured() -> None:
970    """A failed measurement leaves the clip playable at its own level."""
971    renderer = DummyRenderer()
972    _attach_queue(renderer, [_clip_item("sess_001")])
973    _attach_normalization(renderer)
974
975    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
976
977    assert renderer.measure_calls
978    assert streamdetails.stream_type == StreamType.HTTP
979    assert streamdetails.data is None
980
981
982async def test_a_clip_above_the_wanted_level_is_trimmed_back_down() -> None:
983    """The trim runs in either direction, so a loud voice is brought down to the target."""
984    renderer = DummyRenderer()
985    renderer.measured_loudness = -8.0
986    _attach_queue(renderer, [_clip_item("sess_001")])
987    _attach_normalization(renderer, target=-14, boost=3)
988
989    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
990
991    assert streamdetails.stream_type == StreamType.CUSTOM
992    assert streamdetails.data.gain_db == pytest.approx(-3.0)
993
994
995async def test_the_levelled_clip_does_not_hand_out_the_shared_pcm_format() -> None:
996    """Core writes what ffmpeg reports onto this format, so it may not be the shared one."""
997    renderer = DummyRenderer()
998    renderer.measured_loudness = -18.0
999    _attach_queue(renderer, [_clip_item("sess_001")])
1000    _attach_normalization(renderer)
1001
1002    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1003
1004    assert streamdetails.decoded_audio_format == TTS_CLIP_PCM_FORMAT
1005    assert streamdetails.decoded_audio_format is not TTS_CLIP_PCM_FORMAT
1006
1007
1008# verbatim ffmpeg 7.1 output, so the parsing this depends on is covered for real
1009FFMPEG_LOUDNORM_OUTPUT = b"""[Parsed_loudnorm_0 @ 0x93b41d440] \n{
1010\t"input_i" : "-18.37",
1011\t"input_tp" : "-1.89",
1012\t"input_lra" : "0.40",
1013\t"input_thresh" : "-27.86",
1014\t"normalization_type" : "dynamic"
1015}
1016"""
1017
1018
1019async def test_the_measurement_reads_the_level_out_of_ffmpegs_report(
1020    monkeypatch: pytest.MonkeyPatch,
1021) -> None:
1022    """The engine reference comes from loudnorm's report on the rendered clip."""
1023    captured: dict[str, Any] = {}
1024
1025    async def fake_check_output(*args: str, **_kwargs: Any) -> tuple[int, bytes]:
1026        captured["args"] = args
1027        return 0, FFMPEG_LOUDNORM_OUTPUT
1028
1029    monkeypatch.setattr(
1030        "music_assistant.providers.ai_radio.rendering.check_output", fake_check_output
1031    )
1032    renderer = DummyRenderer()
1033
1034    loudness = await AIRadioRenderMixin._measure_loudness(renderer, "http://ha.invalid/clip.mp3")
1035
1036    assert loudness == -18.37
1037    assert "http://ha.invalid/clip.mp3" in captured["args"]
1038    # the reading has to come from behind speechnorm, or the gain corrects for a level
1039    # that never reaches it
1040    assert f"{TTS_SPEECHNORM_FILTER},loudnorm=print_format=json" in captured["args"]
1041
1042
1043async def test_a_failed_measurement_leaves_the_level_unknown(
1044    monkeypatch: pytest.MonkeyPatch,
1045) -> None:
1046    """An ffmpeg run that did not succeed yields no reference rather than a wrong one."""
1047
1048    async def fake_check_output(*_args: str, **_kwargs: Any) -> tuple[int, bytes]:
1049        return 1, b"ffmpeg: Invalid data found when processing input"
1050
1051    monkeypatch.setattr(
1052        "music_assistant.providers.ai_radio.rendering.check_output", fake_check_output
1053    )
1054    renderer = DummyRenderer()
1055
1056    assert (
1057        await AIRadioRenderMixin._measure_loudness(renderer, "http://ha.invalid/clip.mp3") is None
1058    )
1059
1060
1061async def test_clip_plays_untouched_when_tracks_are_not_normalized() -> None:
1062    """The queue switch alone does not mean the music around the clip is levelled."""
1063    renderer = DummyRenderer()
1064    renderer.measured_loudness = -18.0
1065    _attach_queue(renderer, [_clip_item("sess_001")])
1066    _attach_normalization(renderer, tracks_mode=VolumeNormalizationMode.DISABLED.value)
1067
1068    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1069
1070    assert streamdetails.stream_type == StreamType.HTTP
1071    assert streamdetails.data is None
1072
1073
1074async def test_no_measurement_is_taken_when_the_reading_has_nowhere_to_go() -> None:
1075    """Measuring costs a fetch and a decode, so a queue that will not use it is not charged."""
1076    renderer = DummyRenderer()
1077    renderer.measured_loudness = -18.0
1078    _attach_queue(renderer, [_clip_item("sess_001")])
1079    _attach_normalization(renderer, enabled=False)
1080
1081    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1082
1083    assert renderer.measure_calls == []
1084