/
/
/
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_resolve_deferred_placeholders_timestamp_spells_out_weekday() -> None:
768 """The <timestamp> value names the weekday so the LLM never has to derive it."""
769 renderer = DummyRenderer()
770
771 values = await renderer._resolve_deferred_placeholders("Just plain text, no tokens.")
772
773 assert "Thursday" in values["<timestamp>"]
774 assert "July" in values["<timestamp>"]
775
776
777async def test_mint_clip_media_resolves_host_tts_engine() -> None:
778 """A clip whose host declares a tts_engine reaches _get_tts_engine with that override."""
779 renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
780 renderer._hosts = {"rick": {"id": "rick", "tts_engine": "tts.rick_voice"}}
781 cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
782 _attach_normalization(renderer)
783 item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
784
785 await renderer._mint_clip_media(item, "hello world", "clip_1")
786
787 cast("Any", renderer)._get_tts_engine.assert_awaited_once_with("tts.rick_voice")
788
789
790async def test_mint_clip_media_forwards_the_hosts_options() -> None:
791 """A host's configured TTS options are forwarded into the render call."""
792 renderer = DummyRenderer()
793 renderer._hosts = {
794 "rick": {
795 "id": "rick",
796 "tts_engine": "",
797 "options": {"voice": "en_US-lessac-medium", "length_scale": 1.2},
798 }
799 }
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 == [{"voice": "en_US-lessac-medium", "length_scale": 1.2}]
807
808
809async def test_mint_clip_media_sends_no_options_for_a_host_without_any() -> None:
810 """A host with no configured options forwards an empty dict, not None."""
811 renderer = DummyRenderer()
812 renderer._hosts = {"rick": {"id": "rick", "tts_engine": ""}}
813 cast("Any", renderer).mass = SimpleNamespace(metadata=SimpleNamespace(locale="en_US"))
814 _attach_normalization(renderer)
815 item = _clip_item("sess_001", **{ATTR_HOST_ID: "rick"})
816
817 await renderer._mint_clip_media(item, "hello world", "clip_1")
818
819 assert renderer.tts_options == [{}]
820
821
822async def test_render_tts_media_forwards_the_hosts_tts_options() -> None:
823 """A host's configured TTS options reach the engine's get_tts_message call."""
824 renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
825 renderer._hosts = {
826 "rick": {
827 "id": "rick",
828 "tts_engine": "",
829 "options": {"voice": "en_US-lessac-medium", "length_scale": 1.2},
830 }
831 }
832 _attach_queue(renderer, [_clip_item("sess_001", **{ATTR_HOST_ID: "rick"})])
833
834 await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
835
836 engine = cast("Any", renderer)._get_tts_engine.return_value
837 engine.provider.get_tts_message.assert_awaited_once_with(
838 "Good evening, it is warm out.",
839 language="en-US",
840 engine_id="tts.cloud",
841 options={"voice": "en_US-lessac-medium", "length_scale": 1.2},
842 )
843
844
845async def test_render_tts_media_streams_a_url_over_http() -> None:
846 """A TTS engine returning a proxy URL yields an HTTP stream with the MP3 default format."""
847 renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
848
849 path, stream_type, audio_format = await renderer._render_tts_media("hello world")
850
851 assert path == "http://example.test/api/tts_proxy/abc123.mp3"
852 assert stream_type == StreamType.HTTP
853 assert audio_format.content_type == ContentType.MP3
854 engine = cast("Any", renderer)._get_tts_engine.return_value
855 # the provider-scoped engine.id, never engine.uid, and never omitted
856 engine.provider.get_tts_message.assert_awaited_once_with(
857 "hello world", language=None, engine_id="tts.cloud", options=None
858 )
859
860
861async def test_render_tts_media_streams_a_local_file_from_disk(tmp_path: Path) -> None:
862 """A TTS engine that renders to disk is played as a local file rather than fetched."""
863 clip = tmp_path / "section.mp3"
864 clip.write_bytes(b"")
865 renderer = _tts_renderer(str(clip))
866 _attach_queue(renderer, [_clip_item("sess_001")])
867
868 path, stream_type, _ = await renderer._render_tts_media("hello world")
869
870 assert path == str(clip)
871 assert stream_type == StreamType.LOCAL_FILE
872
873
874async def test_render_tts_media_keeps_a_declared_audio_format(tmp_path: Path) -> None:
875 """A TTS engine that declares its own format has it carried into the clip streamdetails."""
876 clip = tmp_path / "section.wav"
877 clip.write_bytes(b"")
878 renderer = _tts_renderer(str(clip), AudioFormat(content_type=ContentType.WAV))
879
880 _, _, audio_format = await renderer._render_tts_media("hello world")
881
882 assert audio_format.content_type == ContentType.WAV
883
884
885@pytest.mark.parametrize("path", ["", "section.mp3", "/does/not/exist.mp3"])
886async def test_render_tts_media_rejects_an_unplayable_path(path: str) -> None:
887 """A path that is neither a URL nor an existing file fails loudly instead of degrading."""
888 renderer = _tts_renderer(path)
889
890 with pytest.raises(InvalidDataError, match="unusable stream path"):
891 await renderer._render_tts_media("hello world")
892
893
894async def test_render_tts_media_gives_up_on_a_stalled_engine(
895 monkeypatch: pytest.MonkeyPatch,
896) -> None:
897 """A stalled TTS engine fails the clip instead of pinning the render path."""
898 monkeypatch.setattr("music_assistant.helpers.tts.TTS_QUERY_TIMEOUT_SECONDS", 0.01)
899
900 async def _answers_too_late(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
901 await asyncio.sleep(5)
902 return SimpleNamespace(
903 path="http://example.test/late.mp3",
904 audio_format=AudioFormat(content_type=ContentType.MP3),
905 )
906
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=_answers_too_late)
910
911 with pytest.raises(MusicAssistantError, match="did not respond within"):
912 await renderer._render_tts_media("hello world")
913
914
915async def test_render_tts_media_reports_an_engine_side_timeout_as_is() -> None:
916 """A timeout raised by the TTS engine itself is not reported as our own cap."""
917 renderer = _tts_renderer("http://example.test/late.mp3")
918 engine = cast("Any", renderer)._get_tts_engine.return_value
919 engine.provider.get_tts_message = AsyncMock(side_effect=TimeoutError)
920
921 with pytest.raises(TimeoutError) as error:
922 await renderer._render_tts_media("hello world")
923 assert "did not respond within" not in str(error.value)
924
925
926async def test_local_file_clip_yields_local_file_streamdetails(tmp_path: Path) -> None:
927 """A disk-rendered clip reaches the core as LOCAL_FILE streamdetails."""
928 clip = tmp_path / "section.mp3"
929 clip.write_bytes(b"")
930 renderer = _tts_renderer(str(clip))
931 _attach_queue(renderer, [_clip_item("sess_001")])
932
933 streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
934
935 assert streamdetails.stream_type == StreamType.LOCAL_FILE
936 assert streamdetails.path == str(clip)
937
938
939async def test_a_measured_clip_is_lifted_to_the_target_plus_the_boost() -> None:
940 """A clip quieter than the levelled music is served through the provider's own chain."""
941 renderer = DummyRenderer()
942 renderer.measured_loudness = -18.0
943 _attach_queue(renderer, [_clip_item("sess_001")])
944 _attach_normalization(renderer, target=-14, boost=3)
945
946 streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
947
948 assert streamdetails.stream_type == StreamType.CUSTOM
949 assert streamdetails.decoded_audio_format == TTS_CLIP_PCM_FORMAT
950 assert streamdetails.audio_format.content_type == ContentType.MP3
951 assert streamdetails.data.gain_db == pytest.approx(7.0)
952
953
954async def test_the_clip_is_evened_out_and_lifted_before_it_is_limited(
955 monkeypatch: pytest.MonkeyPatch,
956) -> None:
957 """The filter chain reaches ffmpeg as speechnorm, then gain, then the peak limiter."""
958 captured: dict[str, Any] = {}
959
960 async def fake_ffmpeg_stream(**kwargs: Any) -> AsyncGenerator[bytes]:
961 captured.update(kwargs)
962 yield b"pcm"
963
964 monkeypatch.setattr(
965 "music_assistant.providers.ai_radio.rendering.get_ffmpeg_stream", fake_ffmpeg_stream
966 )
967 renderer = DummyRenderer()
968 renderer.measured_loudness = -18.0
969 _attach_queue(renderer, [_clip_item("sess_001")])
970 _attach_normalization(renderer, target=-14, boost=3)
971 streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
972
973 chunks = [chunk async for chunk in renderer.get_audio_stream(streamdetails)]
974
975 assert chunks == [b"pcm"]
976 assert captured["audio_input"] == streamdetails.path
977 assert captured["input_format"].content_type == ContentType.MP3
978 assert captured["output_format"] == TTS_CLIP_PCM_FORMAT
979 assert captured["filter_params"] == [
980 TTS_SPEECHNORM_FILTER,
981 "volume=7.0dB",
982 f"alimiter=limit={TTS_PEAK_CEILING_DB}dB:level=false:latency=true",
983 ]
984
985
986async def test_the_engine_reference_is_measured_once_and_reused() -> None:
987 """A second clip from the same voice levels itself against the stored measurement."""
988 renderer = DummyRenderer()
989 renderer.measured_loudness = -18.0
990 _attach_queue(renderer, [_clip_item("sess_001"), _clip_item("sess_002")])
991 _attach_normalization(renderer)
992
993 first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
994 second = await renderer.get_stream_details("sess_002", MediaType.SOUND_EFFECT)
995
996 assert len(renderer.measure_calls) == 1
997 assert first.data.gain_db == second.data.gain_db
998
999
1000async def test_a_short_clip_levels_itself_but_never_becomes_the_reference() -> None:
1001 """A clip of a few words is too thin a sample to speak for the rest of the engine."""
1002
1003 class BriefRenderer(DummyRenderer):
1004 async def _probe_duration(self, path: str) -> int | None:
1005 return MIN_LOUDNESS_REFERENCE_SECONDS - 1
1006
1007 renderer = BriefRenderer()
1008 renderer.measured_loudness = -18.0
1009 _attach_queue(renderer, [_clip_item("sess_001"), _clip_item("sess_002")])
1010 _attach_normalization(renderer)
1011
1012 first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1013 await renderer.get_stream_details("sess_002", MediaType.SOUND_EFFECT)
1014
1015 assert first.stream_type == StreamType.CUSTOM
1016 assert len(renderer.measure_calls) == 2
1017 assert cast("Any", renderer)._engine_loudness == {}
1018
1019
1020async def test_clip_plays_untouched_when_the_queue_does_not_normalize() -> None:
1021 """With the music unlevelled there is nothing to match, so the clip airs as rendered."""
1022 renderer = DummyRenderer()
1023 renderer.measured_loudness = -18.0
1024 _attach_queue(renderer, [_clip_item("sess_001")])
1025 _attach_normalization(renderer, enabled=False)
1026
1027 streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1028
1029 assert streamdetails.stream_type == StreamType.HTTP
1030 assert streamdetails.decoded_audio_format is None
1031 assert streamdetails.data is None
1032
1033
1034async def test_clip_plays_untouched_when_it_could_not_be_measured() -> None:
1035 """A failed measurement leaves the clip playable at its own level."""
1036 renderer = DummyRenderer()
1037 _attach_queue(renderer, [_clip_item("sess_001")])
1038 _attach_normalization(renderer)
1039
1040 streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1041
1042 assert renderer.measure_calls
1043 assert streamdetails.stream_type == StreamType.HTTP
1044 assert streamdetails.data is None
1045
1046
1047async def test_a_clip_above_the_wanted_level_is_trimmed_back_down() -> None:
1048 """The trim runs in either direction, so a loud voice is brought down to the target."""
1049 renderer = DummyRenderer()
1050 renderer.measured_loudness = -8.0
1051 _attach_queue(renderer, [_clip_item("sess_001")])
1052 _attach_normalization(renderer, target=-14, boost=3)
1053
1054 streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1055
1056 assert streamdetails.stream_type == StreamType.CUSTOM
1057 assert streamdetails.data.gain_db == pytest.approx(-3.0)
1058
1059
1060async def test_the_levelled_clip_does_not_hand_out_the_shared_pcm_format() -> None:
1061 """Core writes what ffmpeg reports onto this format, so it may not be the shared one."""
1062 renderer = DummyRenderer()
1063 renderer.measured_loudness = -18.0
1064 _attach_queue(renderer, [_clip_item("sess_001")])
1065 _attach_normalization(renderer)
1066
1067 streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1068
1069 assert streamdetails.decoded_audio_format == TTS_CLIP_PCM_FORMAT
1070 assert streamdetails.decoded_audio_format is not TTS_CLIP_PCM_FORMAT
1071
1072
1073# verbatim ffmpeg 7.1 output, so the parsing this depends on is covered for real
1074FFMPEG_LOUDNORM_OUTPUT = b"""[Parsed_loudnorm_0 @ 0x93b41d440] \n{
1075\t"input_i" : "-18.37",
1076\t"input_tp" : "-1.89",
1077\t"input_lra" : "0.40",
1078\t"input_thresh" : "-27.86",
1079\t"normalization_type" : "dynamic"
1080}
1081"""
1082
1083
1084async def test_the_measurement_reads_the_level_out_of_ffmpegs_report(
1085 monkeypatch: pytest.MonkeyPatch,
1086) -> None:
1087 """The engine reference comes from loudnorm's report on the rendered clip."""
1088 captured: dict[str, Any] = {}
1089
1090 async def fake_check_output(*args: str, **_kwargs: Any) -> tuple[int, bytes]:
1091 captured["args"] = args
1092 return 0, FFMPEG_LOUDNORM_OUTPUT
1093
1094 monkeypatch.setattr(
1095 "music_assistant.providers.ai_radio.rendering.check_output", fake_check_output
1096 )
1097 renderer = DummyRenderer()
1098
1099 loudness = await AIRadioRenderMixin._measure_loudness(renderer, "http://ha.invalid/clip.mp3")
1100
1101 assert loudness == -18.37
1102 assert "http://ha.invalid/clip.mp3" in captured["args"]
1103 # the reading has to come from behind speechnorm, or the gain corrects for a level
1104 # that never reaches it
1105 assert f"{TTS_SPEECHNORM_FILTER},loudnorm=print_format=json" in captured["args"]
1106
1107
1108async def test_a_failed_measurement_leaves_the_level_unknown(
1109 monkeypatch: pytest.MonkeyPatch,
1110) -> None:
1111 """An ffmpeg run that did not succeed yields no reference rather than a wrong one."""
1112
1113 async def fake_check_output(*_args: str, **_kwargs: Any) -> tuple[int, bytes]:
1114 return 1, b"ffmpeg: Invalid data found when processing input"
1115
1116 monkeypatch.setattr(
1117 "music_assistant.providers.ai_radio.rendering.check_output", fake_check_output
1118 )
1119 renderer = DummyRenderer()
1120
1121 assert (
1122 await AIRadioRenderMixin._measure_loudness(renderer, "http://ha.invalid/clip.mp3") is None
1123 )
1124
1125
1126async def test_clip_plays_untouched_when_tracks_are_not_normalized() -> None:
1127 """The queue switch alone does not mean the music around the clip is levelled."""
1128 renderer = DummyRenderer()
1129 renderer.measured_loudness = -18.0
1130 _attach_queue(renderer, [_clip_item("sess_001")])
1131 _attach_normalization(renderer, tracks_mode=VolumeNormalizationMode.DISABLED.value)
1132
1133 streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1134
1135 assert streamdetails.stream_type == StreamType.HTTP
1136 assert streamdetails.data is None
1137
1138
1139async def test_no_measurement_is_taken_when_the_reading_has_nowhere_to_go() -> None:
1140 """Measuring costs a fetch and a decode, so a queue that will not use it is not charged."""
1141 renderer = DummyRenderer()
1142 renderer.measured_loudness = -18.0
1143 _attach_queue(renderer, [_clip_item("sess_001")])
1144 _attach_normalization(renderer, enabled=False)
1145
1146 await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
1147
1148 assert renderer.measure_calls == []
1149