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