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