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
533async def test_tts_server_error_fails_the_clip_with_an_actionable_message(
534    monkeypatch: pytest.MonkeyPatch,
535) -> None:
536    """An engine that hands out a URL it cannot render fails the clip, not the playback."""
537    renderer = _failing_probe(
538        "Unable to retrieve info for http://ha.invalid/api/tts_proxy/1.mp3 "
539        "(Server returned 5XX Server Error reply)",
540        monkeypatch,
541    )
542
543    with pytest.raises(MediaNotFoundError):
544        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
545
546    session = renderer._sessions["sess"]
547    assert session.skipped_sections == 1
548    assert "enough credit" in session.last_render_error
549
550
551async def test_unmeasurable_clip_still_plays(monkeypatch: pytest.MonkeyPatch) -> None:
552    """A probe that only fails to measure the audio leaves the clip playable."""
553    renderer = _failing_probe(
554        "Unable to retrieve info for http://ha.invalid/api/tts_proxy/1.mp3 "
555        "(Invalid or unsupported media file)",
556        monkeypatch,
557    )
558
559    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
560
561    assert streamdetails.duration is None
562    assert renderer._sessions["sess"].skipped_sections == 0
563
564
565async def test_generate_script_uses_host_instructions() -> None:
566    """The prompt sent to the LLM carries the resolved host's persona instructions."""
567    renderer = DummyRenderer()
568    renderer._hosts = {"rick": {"id": "rick", "instructions": "Persona text.", "tts_engine": ""}}
569    captured: dict[str, str] = {}
570
571    async def fake_generate_text(
572        instructions: str,
573        prompt: str,  # noqa: ARG001
574        web_mode: str,  # noqa: ARG001
575        language: str | None = None,  # noqa: ARG001
576    ) -> str:
577        captured["instructions"] = instructions
578        return "script"
579
580    cast("Any", renderer)._generate_text = fake_generate_text
581    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
582
583    text = await renderer._generate_script(item, "p", "clip_1")
584
585    assert text == "script"
586    assert captured["instructions"] == "Persona text."
587
588
589async def test_generate_script_falls_back_to_default_instructions() -> None:
590    """A clip whose host is gone by render time still generates, using the default persona."""
591    renderer = DummyRenderer()
592    renderer._hosts = {}
593    captured: dict[str, str] = {}
594
595    async def fake_generate_text(
596        instructions: str,
597        prompt: str,  # noqa: ARG001
598        web_mode: str,  # noqa: ARG001
599        language: str | None = None,  # noqa: ARG001
600    ) -> str:
601        # mirrors the empty-to-default fallback the real _generate_text applies (runtime.py)
602        captured["instructions"] = instructions.strip() or DEFAULT_LLM_INSTRUCTIONS
603        return "script"
604
605    cast("Any", renderer)._generate_text = fake_generate_text
606    item = _clip_item("sess_001", **{ATTR_HOST_ID: "gone", ATTR_PROMPT: "p"})
607
608    await renderer._generate_script(item, "p", "clip_1")
609
610    assert captured["instructions"] == DEFAULT_LLM_INSTRUCTIONS
611
612
613async def test_generate_script_forwards_the_hosts_language() -> None:
614    """The host's configured language reaches _generate_text, ready to override the locale."""
615    renderer = DummyRenderer()
616    renderer._hosts = {
617        "rick": {
618            "id": "rick",
619            "instructions": "Persona text.",
620            "tts_engine": "",
621            "language": "fr_FR",
622        }
623    }
624    captured: dict[str, str | None] = {}
625
626    async def fake_generate_text(
627        instructions: str,  # noqa: ARG001
628        prompt: str,  # noqa: ARG001
629        web_mode: str,  # noqa: ARG001
630        language: str | None = None,
631    ) -> str:
632        captured["language"] = language
633        return "script"
634
635    cast("Any", renderer)._generate_text = fake_generate_text
636    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
637
638    await renderer._generate_script(item, "p", "clip_1")
639
640    assert captured["language"] == "fr_FR"
641
642
643async def test_generate_script_forwards_empty_language_when_host_has_none() -> None:
644    """A host with no configured language forwards an empty string, not None."""
645    renderer = DummyRenderer()
646    renderer._hosts = {"rick": {"id": "rick", "instructions": "Persona text.", "tts_engine": ""}}
647    captured: dict[str, str | None] = {}
648
649    async def fake_generate_text(
650        instructions: str,  # noqa: ARG001
651        prompt: str,  # noqa: ARG001
652        web_mode: str,  # noqa: ARG001
653        language: str | None = None,
654    ) -> str:
655        captured["language"] = language
656        return "script"
657
658    cast("Any", renderer)._generate_text = fake_generate_text
659    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick", ATTR_PROMPT: "p"})
660
661    await renderer._generate_script(item, "p", "clip_1")
662
663    assert captured["language"] == ""
664
665
666async def test_render_tts_media_prefers_the_hosts_language_over_the_locale() -> None:
667    """A host's configured language reaches the TTS engine, overriding the server locale."""
668    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
669    renderer._hosts = {"rick": {"id": "rick", "tts_engine": "", "language": "fr_FR"}}
670    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
671
672    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
673
674    engine = cast("Any", renderer)._get_tts_engine.return_value
675    engine.provider.get_tts_message.assert_awaited_once_with(
676        "Good evening, it is warm out.", language="fr-FR", engine_id="tts.cloud", options={}
677    )
678
679
680async def test_render_tts_media_falls_back_to_locale_when_host_language_is_empty() -> None:
681    """A host with no configured language falls back to the server locale for the TTS call."""
682    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
683    renderer._hosts = {"rick": {"id": "rick", "tts_engine": ""}}
684    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
685
686    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
687
688    engine = cast("Any", renderer)._get_tts_engine.return_value
689    engine.provider.get_tts_message.assert_awaited_once_with(
690        "Good evening, it is warm out.", language="en-US", engine_id="tts.cloud", options={}
691    )
692
693
694async def test_resolve_deferred_placeholders_skips_weather_without_token() -> None:
695    """A prompt with no weather placeholder never triggers a weather fetch."""
696    renderer = DummyRenderer()
697
698    values = await renderer._resolve_deferred_placeholders("Just plain text, no tokens.")
699
700    assert renderer.weather_calls == 0
701    assert values["<weather_hourly>"] == ""
702    assert values["<weather_daily>"] == ""
703
704
705async def test_mint_clip_media_resolves_host_tts_engine() -> None:
706    """A clip whose host declares a tts_engine reaches _get_tts_engine with that override."""
707    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
708    renderer._hosts = {"rick": {"id": "rick", "tts_engine": "tts.rick_voice"}}
709    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
710    _attach_normalization(renderer)
711    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
712
713    await renderer._mint_clip_media(item, "hello world", "clip_1")
714
715    cast("Any", renderer)._get_tts_engine.assert_awaited_once_with("tts.rick_voice")
716
717
718async def test_mint_clip_media_forwards_the_hosts_options() -> None:
719    """A host's configured TTS options are forwarded into the render call."""
720    renderer = DummyRenderer()
721    renderer._hosts = {
722        "rick": {
723            "id": "rick",
724            "tts_engine": "",
725            "options": {"voice": "en_US-lessac-medium", "length_scale": 1.2},
726        }
727    }
728    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
729    _attach_normalization(renderer)
730    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
731
732    await renderer._mint_clip_media(item, "hello world", "clip_1")
733
734    assert renderer.tts_options == [{"voice": "en_US-lessac-medium", "length_scale": 1.2}]
735
736
737async def test_mint_clip_media_sends_no_options_for_a_host_without_any() -> None:
738    """A host with no configured options forwards an empty dict, not None."""
739    renderer = DummyRenderer()
740    renderer._hosts = {"rick": {"id": "rick", "tts_engine": ""}}
741    cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
742    _attach_normalization(renderer)
743    item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
744
745    await renderer._mint_clip_media(item, "hello world", "clip_1")
746
747    assert renderer.tts_options == [{}]
748
749
750async def test_render_tts_media_forwards_the_hosts_tts_options() -> None:
751    """A host's configured TTS options reach the engine's get_tts_message call."""
752    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
753    renderer._hosts = {
754        "rick": {
755            "id": "rick",
756            "tts_engine": "",
757            "options": {"voice": "en_US-lessac-medium", "length_scale": 1.2},
758        }
759    }
760    _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
761
762    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
763
764    engine = cast("Any", renderer)._get_tts_engine.return_value
765    engine.provider.get_tts_message.assert_awaited_once_with(
766        "Good evening, it is warm out.",
767        language="en-US",
768        engine_id="tts.cloud",
769        options={"voice": "en_US-lessac-medium", "length_scale": 1.2},
770    )
771
772
773async def test_render_tts_media_streams_a_url_over_http() -> None:
774    """A TTS engine returning a proxy URL yields an HTTP stream with the MP3 default format."""
775    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
776
777    path, stream_type, audio_format = await renderer._render_tts_media("hello world")
778
779    assert path == "http://example.test/api/tts_proxy/abc123.mp3"
780    assert stream_type == StreamType.HTTP
781    assert audio_format.content_type == ContentType.MP3
782    engine = cast("Any", renderer)._get_tts_engine.return_value
783    # the provider-scoped engine.id, never engine.uid, and never omitted
784    engine.provider.get_tts_message.assert_awaited_once_with(
785        "hello world", language=None, engine_id="tts.cloud", options=None
786    )
787
788
789async def test_render_tts_media_streams_a_local_file_from_disk(tmp_path: Path) -> None:
790    """A TTS engine that renders to disk is played as a local file rather than fetched."""
791    clip = tmp_path / "section.mp3"
792    clip.write_bytes(b"")
793    renderer = _tts_renderer(str(clip))
794    _attach_queue(renderer, [_clip_item("sess_001")])
795
796    path, stream_type, _ = await renderer._render_tts_media("hello world")
797
798    assert path == str(clip)
799    assert stream_type == StreamType.LOCAL_FILE
800
801
802async def test_render_tts_media_keeps_a_declared_audio_format(tmp_path: Path) -> None:
803    """A TTS engine that declares its own format has it carried into the clip streamdetails."""
804    clip = tmp_path / "section.wav"
805    clip.write_bytes(b"")
806    renderer = _tts_renderer(str(clip), AudioFormat(content_type=ContentType.WAV))
807
808    _, _, audio_format = await renderer._render_tts_media("hello world")
809
810    assert audio_format.content_type == ContentType.WAV
811
812
813@pytest.mark.parametrize("path", ["", "section.mp3", "/does/not/exist.mp3"])
814async def test_render_tts_media_rejects_an_unplayable_path(path: str) -> None:
815    """A path that is neither a URL nor an existing file fails loudly instead of degrading."""
816    renderer = _tts_renderer(path)
817
818    with pytest.raises(InvalidDataError, match="unusable stream path"):
819        await renderer._render_tts_media("hello world")
820
821
822async def test_render_tts_media_gives_up_on_a_stalled_engine(
823    monkeypatch: pytest.MonkeyPatch,
824) -> None:
825    """A stalled TTS engine fails the clip instead of pinning the render path."""
826    monkeypatch.setattr("music_assistant.helpers.tts.TTS_QUERY_TIMEOUT_SECONDS", 0.01)
827
828    async def _answers_too_late(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
829        await asyncio.sleep(5)
830        return SimpleNamespace(
831            path="http://example.test/late.mp3",
832            audio_format=AudioFormat(content_type=ContentType.MP3),
833        )
834
835    renderer = _tts_renderer("http://example.test/late.mp3")
836    engine = cast("Any", renderer)._get_tts_engine.return_value
837    engine.provider.get_tts_message = AsyncMock(side_effect=_answers_too_late)
838
839    with pytest.raises(MusicAssistantError, match="did not respond within"):
840        await renderer._render_tts_media("hello world")
841
842
843async def test_render_tts_media_reports_an_engine_side_timeout_as_is() -> None:
844    """A timeout raised by the TTS engine itself is not reported as our own cap."""
845    renderer = _tts_renderer("http://example.test/late.mp3")
846    engine = cast("Any", renderer)._get_tts_engine.return_value
847    engine.provider.get_tts_message = AsyncMock(side_effect=TimeoutError)
848
849    with pytest.raises(TimeoutError) as error:
850        await renderer._render_tts_media("hello world")
851    assert "did not respond within" not in str(error.value)
852
853
854async def test_local_file_clip_yields_local_file_streamdetails(tmp_path: Path) -> None:
855    """A disk-rendered clip reaches the core as LOCAL_FILE streamdetails."""
856    clip = tmp_path / "section.mp3"
857    clip.write_bytes(b"")
858    renderer = _tts_renderer(str(clip))
859    _attach_queue(renderer, [_clip_item("sess_001")])
860
861    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
862
863    assert streamdetails.stream_type == StreamType.LOCAL_FILE
864    assert streamdetails.path == str(clip)
865
866
867async def test_a_measured_clip_is_lifted_to_the_target_plus_the_boost() -> None:
868    """A clip quieter than the levelled music is served through the provider's own chain."""
869    renderer = DummyRenderer()
870    renderer.measured_loudness = -18.0
871    _attach_queue(renderer, [_clip_item("sess_001")])
872    _attach_normalization(renderer, target=-14, boost=3)
873
874    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
875
876    assert streamdetails.stream_type == StreamType.CUSTOM
877    assert streamdetails.decoded_audio_format == TTS_CLIP_PCM_FORMAT
878    assert streamdetails.audio_format.content_type == ContentType.MP3
879    assert streamdetails.data.gain_db == pytest.approx(7.0)
880
881
882async def test_the_clip_is_evened_out_and_lifted_before_it_is_limited(
883    monkeypatch: pytest.MonkeyPatch,
884) -> None:
885    """The filter chain reaches ffmpeg as speechnorm, then gain, then the peak limiter."""
886    captured: dict[str, Any] = {}
887
888    async def fake_ffmpeg_stream(**kwargs: Any) -> AsyncGenerator[bytes]:
889        captured.update(kwargs)
890        yield b"pcm"
891
892    monkeypatch.setattr(
893        "music_assistant.providers.ai_radio.rendering.get_ffmpeg_stream", fake_ffmpeg_stream
894    )
895    renderer = DummyRenderer()
896    renderer.measured_loudness = -18.0
897    _attach_queue(renderer, [_clip_item("sess_001")])
898    _attach_normalization(renderer, target=-14, boost=3)
899    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
900
901    chunks = [chunk async for chunk in renderer.get_audio_stream(streamdetails)]
902
903    assert chunks == [b"pcm"]
904    assert captured["audio_input"] == streamdetails.path
905    assert captured["input_format"].content_type == ContentType.MP3
906    assert captured["output_format"] == TTS_CLIP_PCM_FORMAT
907    assert captured["filter_params"] == [
908        TTS_SPEECHNORM_FILTER,
909        "volume=7.0dB",
910        f"alimiter=limit={TTS_PEAK_CEILING_DB}dB:level=false:latency=true",
911    ]
912
913
914async def test_the_engine_reference_is_measured_once_and_reused() -> None:
915    """A second clip from the same voice levels itself against the stored measurement."""
916    renderer = DummyRenderer()
917    renderer.measured_loudness = -18.0
918    _attach_queue(renderer, [_clip_item("sess_001"), _clip_item("sess_002")])
919    _attach_normalization(renderer)
920
921    first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
922    second = await renderer.get_stream_details("sess_002", MediaType.SOUND_EFFECT)
923
924    assert len(renderer.measure_calls) == 1
925    assert first.data.gain_db == second.data.gain_db
926
927
928async def test_a_short_clip_levels_itself_but_never_becomes_the_reference() -> None:
929    """A clip of a few words is too thin a sample to speak for the rest of the engine."""
930
931    class BriefRenderer(DummyRenderer):
932        async def _probe_duration(self, path: str) -> int | None:
933            return MIN_LOUDNESS_REFERENCE_SECONDS - 1
934
935    renderer = BriefRenderer()
936    renderer.measured_loudness = -18.0
937    _attach_queue(renderer, [_clip_item("sess_001"), _clip_item("sess_002")])
938    _attach_normalization(renderer)
939
940    first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
941    await renderer.get_stream_details("sess_002", MediaType.SOUND_EFFECT)
942
943    assert first.stream_type == StreamType.CUSTOM
944    assert len(renderer.measure_calls) == 2
945    assert cast("Any", renderer)._engine_loudness == {}
946
947
948async def test_clip_plays_untouched_when_the_queue_does_not_normalize() -> None:
949    """With the music unlevelled there is nothing to match, so the clip airs as rendered."""
950    renderer = DummyRenderer()
951    renderer.measured_loudness = -18.0
952    _attach_queue(renderer, [_clip_item("sess_001")])
953    _attach_normalization(renderer, enabled=False)
954
955    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
956
957    assert streamdetails.stream_type == StreamType.HTTP
958    assert streamdetails.decoded_audio_format is None
959    assert streamdetails.data is None
960
961
962async def test_clip_plays_untouched_when_it_could_not_be_measured() -> None:
963    """A failed measurement leaves the clip playable at its own level."""
964    renderer = DummyRenderer()
965    _attach_queue(renderer, [_clip_item("sess_001")])
966    _attach_normalization(renderer)
967
968    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
969
970    assert renderer.measure_calls
971    assert streamdetails.stream_type == StreamType.HTTP
972    assert streamdetails.data is None
973
974
975async def test_a_clip_above_the_wanted_level_is_trimmed_back_down() -> None:
976    """The trim runs in either direction, so a loud voice is brought down to the target."""
977    renderer = DummyRenderer()
978    renderer.measured_loudness = -8.0
979    _attach_queue(renderer, [_clip_item("sess_001")])
980    _attach_normalization(renderer, target=-14, boost=3)
981
982    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
983
984    assert streamdetails.stream_type == StreamType.CUSTOM
985    assert streamdetails.data.gain_db == pytest.approx(-3.0)
986
987
988async def test_the_levelled_clip_does_not_hand_out_the_shared_pcm_format() -> None:
989    """Core writes what ffmpeg reports onto this format, so it may not be the shared one."""
990    renderer = DummyRenderer()
991    renderer.measured_loudness = -18.0
992    _attach_queue(renderer, [_clip_item("sess_001")])
993    _attach_normalization(renderer)
994
995    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
996
997    assert streamdetails.decoded_audio_format == TTS_CLIP_PCM_FORMAT
998    assert streamdetails.decoded_audio_format is not TTS_CLIP_PCM_FORMAT
999
1000
1001# verbatim ffmpeg 7.1 output, so the parsing this depends on is covered for real
1002FFMPEG_LOUDNORM_OUTPUT = b"""[Parsed_loudnorm_0 @ 0x93b41d440] \n{
1003\t"input_i" : "-18.37",
1004\t"input_tp" : "-1.89",
1005\t"input_lra" : "0.40",
1006\t"input_thresh" : "-27.86",
1007\t"normalization_type" : "dynamic"
1008}
1009"""
1010
1011
1012async def test_the_measurement_reads_the_level_out_of_ffmpegs_report(
1013    monkeypatch: pytest.MonkeyPatch,
1014) -> None:
1015    """The engine reference comes from loudnorm's report on the rendered clip."""
1016    captured: dict[str, Any] = {}
1017
1018    async def fake_check_output(*args: str, **_kwargs: Any) -> tuple[int, bytes]:
1019        captured["args"] = args
1020        return 0, FFMPEG_LOUDNORM_OUTPUT
1021
1022    monkeypatch.setattr(
1023        "music_assistant.providers.ai_radio.rendering.check_output", fake_check_output
1024    )
1025    renderer = DummyRenderer()
1026
1027    loudness = await AIRadioRenderMixin._measure_loudness(renderer, "http://ha.invalid/clip.mp3")
1028
1029    assert loudness == -18.37
1030    assert "http://ha.invalid/clip.mp3" in captured["args"]
1031    # the reading has to come from behind speechnorm, or the gain corrects for a level
1032    # that never reaches it
1033    assert f"{TTS_SPEECHNORM_FILTER},loudnorm=print_format=json" in captured["args"]
1034
1035
1036async def test_a_failed_measurement_leaves_the_level_unknown(
1037    monkeypatch: pytest.MonkeyPatch,
1038) -> None:
1039    """An ffmpeg run that did not succeed yields no reference rather than a wrong one."""
1040
1041    async def fake_check_output(*_args: str, **_kwargs: Any) -> tuple[int, bytes]:
1042        return 1, b"ffmpeg: Invalid data found when processing input"
1043
1044    monkeypatch.setattr(
1045        "music_assistant.providers.ai_radio.rendering.check_output", fake_check_output
1046    )
1047    renderer = DummyRenderer()
1048
1049    assert (
1050        await AIRadioRenderMixin._measure_loudness(renderer, "http://ha.invalid/clip.mp3") is None
1051    )
1052
1053
1054async def test_clip_plays_untouched_when_tracks_are_not_normalized() -> None:
1055    """The queue switch alone does not mean the music around the clip is levelled."""
1056    renderer = DummyRenderer()
1057    renderer.measured_loudness = -18.0
1058    _attach_queue(renderer, [_clip_item("sess_001")])
1059    _attach_normalization(renderer, tracks_mode=VolumeNormalizationMode.DISABLED.value)
1060
1061    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1062
1063    assert streamdetails.stream_type == StreamType.HTTP
1064    assert streamdetails.data is None
1065
1066
1067async def test_no_measurement_is_taken_when_the_reading_has_nowhere_to_go() -> None:
1068    """Measuring costs a fetch and a decode, so a queue that will not use it is not charged."""
1069    renderer = DummyRenderer()
1070    renderer.measured_loudness = -18.0
1071    _attach_queue(renderer, [_clip_item("sess_001")])
1072    _attach_normalization(renderer, enabled=False)
1073
1074    await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1075
1076    assert renderer.measure_calls == []
1077