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 pathlib import Path
8from types import SimpleNamespace
9from typing import Any, cast
10from unittest.mock import AsyncMock, MagicMock
11
12import pytest
13from music_assistant_models.enums import ContentType, MediaType, StreamType
14from music_assistant_models.errors import (
15    InvalidDataError,
16    MediaNotFoundError,
17    MusicAssistantError,
18)
19from music_assistant_models.media_items import AudioFormat, ProviderMapping, SoundEffect
20from music_assistant_models.queue_item import QueueItem
21
22from music_assistant.helpers.tags import AudioTags
23from music_assistant.models.plugin import PluginProvider, TTSEngine
24from music_assistant.providers.ai_radio.constants import (
25    ATTR_MAX_CHARS,
26    ATTR_PROMPT,
27    ATTR_RENDERED_TEXT,
28    ATTR_SESSION_ID,
29    ATTR_STATION_ID,
30    ATTR_WEB_SEARCH_MODE,
31)
32from music_assistant.providers.ai_radio.models import SessionState
33from music_assistant.providers.ai_radio.rendering import AIRadioRenderMixin
34
35
36class DummyRenderer(AIRadioRenderMixin):
37    """Minimal harness exposing the render path."""
38
39    domain = "ai_radio"
40    instance_id = "ai_radio--test"
41
42    def __init__(self) -> None:
43        """Initialize the harness with recording stubs."""
44        self.logger = logging.getLogger("tests.ai_radio.rendering")
45        self._sessions: dict[str, Any] = {}
46        self._stations: dict[str, dict[str, Any]] = {
47            "st": {"id": "st", "general": {"instructions": "be warm"}}
48        }
49        self.llm_prompts: list[str] = []
50        self.tts_texts: list[str] = []
51        self.weather_calls = 0
52        self.fail_generation = False
53
54    def _configured_now(self) -> Any:
55        return __import__("datetime").datetime(2026, 7, 30, 18, 30)
56
57    async def _generate_text(self, station: dict[str, Any], prompt: str, web_mode: str) -> str:
58        # a real suspension point so concurrent callers actually interleave under
59        # asyncio.gather, otherwise the lock in get_stream_details is never exercised
60        await asyncio.sleep(0)
61        if self.fail_generation:
62            raise RuntimeError("llm down")
63        self.llm_prompts.append(prompt)
64        return "Good evening, it is warm out."
65
66    async def _prepare_runtime_tokens(self, station: dict[str, Any]) -> dict[str, str]:
67        self.weather_calls += 1
68        return {"<weather_hourly>": f"fresh weather {self.weather_calls}"}
69
70    async def _render_tts_media(self, text: str) -> tuple[str, StreamType, AudioFormat]:
71        self.tts_texts.append(text)
72        return (
73            f"http://ha.invalid/api/tts_proxy/{len(self.tts_texts)}.mp3",
74            StreamType.HTTP,
75            AudioFormat(content_type=ContentType.MP3),
76        )
77
78    async def _probe_duration(self, path: str) -> int | None:
79        return 9
80
81
82class RealTtsRenderer(DummyRenderer):
83    """Harness that exercises the mixin's real TTS path instead of the DummyRenderer stub."""
84
85    _render_tts_media = AIRadioRenderMixin._render_tts_media
86
87
88def _tts_renderer(path: str, audio_format: AudioFormat | None = None) -> RealTtsRenderer:
89    """Build a renderer whose TTS engine returns StreamDetails carrying the given path."""
90    renderer = RealTtsRenderer()
91    plugin = MagicMock(spec=PluginProvider)
92    plugin.instance_id = "hass_1"
93    plugin.get_tts_message = AsyncMock(
94        return_value=SimpleNamespace(
95            path=path, audio_format=audio_format or AudioFormat(content_type=ContentType.UNKNOWN)
96        )
97    )
98    engine = TTSEngine(id="tts.cloud", name="Cloud", provider=plugin)
99    cast("Any", renderer)._get_tts_engine = AsyncMock(return_value=engine)
100    return renderer
101
102
103def _clip_item(clip_id: str, queue_id: str = "player_a", **overrides: Any) -> QueueItem:
104    """Build a queue item for a pending AI Radio clip."""
105    attributes: dict[str, Any] = {
106        ATTR_SESSION_ID: "sess",
107        ATTR_STATION_ID: "st",
108        ATTR_PROMPT: "It is <timestamp>. Weather: <weather_hourly>.",
109        ATTR_MAX_CHARS: 300,
110        ATTR_WEB_SEARCH_MODE: "disabled",
111    }
112    attributes.update(overrides)
113    media_item = SoundEffect(
114        item_id=clip_id,
115        provider="ai_radio--test",
116        name="Weather",
117        provider_mappings={
118            ProviderMapping(
119                item_id=clip_id,
120                provider_domain="ai_radio",
121                provider_instance="ai_radio--test",
122            )
123        },
124    )
125    return QueueItem(
126        queue_id=queue_id,
127        queue_item_id=f"qi_{clip_id}",
128        name="Weather",
129        duration=None,
130        media_item=media_item,
131        extra_attributes=attributes,
132    )
133
134
135def _attach_queues(renderer: DummyRenderer, queues: dict[str, list[QueueItem]]) -> list[bool]:
136    """Wire a minimal player_queues stub and return the signal_update call log."""
137    signals: list[bool] = []
138    cast("Any", renderer).mass = SimpleNamespace(
139        player_queues=SimpleNamespace(
140            all=lambda: tuple(SimpleNamespace(queue_id=queue_id) for queue_id in queues),
141            items=lambda queue_id, limit=500, offset=0: queues.get(queue_id, [])[
142                offset : offset + limit
143            ],
144            signal_update=lambda _queue_id, items_changed=False: signals.append(items_changed),
145        ),
146        metadata=SimpleNamespace(locale="en_US"),
147    )
148    return signals
149
150
151def _attach_queue(renderer: DummyRenderer, items: list[QueueItem]) -> list[bool]:
152    """Wire a single-queue player_queues stub and return the signal_update call log."""
153    return _attach_queues(renderer, {"player_a": items})
154
155
156async def test_render_resolves_deferred_placeholders_at_render_time() -> None:
157    """The prompt sent to the LLM carries render-time weather, not plan-time."""
158    renderer = DummyRenderer()
159    item = _clip_item("sess_001")
160    _attach_queue(renderer, [item])
161
162    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
163
164    assert renderer.weather_calls == 1
165    assert "fresh weather 1" in renderer.llm_prompts[0]
166    assert "<timestamp>" not in renderer.llm_prompts[0]
167    assert streamdetails.media_type == MediaType.SOUND_EFFECT
168    assert streamdetails.stream_type == StreamType.HTTP
169    assert streamdetails.duration == 9
170    assert streamdetails.expiration == 60
171    assert streamdetails.can_seek is False
172    assert streamdetails.allow_seek is False
173
174
175async def test_render_caches_the_script_and_remints_the_url() -> None:
176    """A second render reuses the stored script but mints a fresh URL."""
177    renderer = DummyRenderer()
178    item = _clip_item("sess_001")
179    signals = _attach_queue(renderer, [item])
180
181    first = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
182    second = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
183
184    assert len(renderer.llm_prompts) == 1
185    assert renderer.tts_texts == [
186        "Good evening, it is warm out.",
187        "Good evening, it is warm out.",
188    ]
189    assert first.path != second.path
190    assert item.extra_attributes[ATTR_RENDERED_TEXT] == "Good evening, it is warm out."
191    assert signals == [True]
192
193
194async def test_concurrent_renders_call_the_llm_once() -> None:
195    """Two simultaneous requests for one clip render a single script."""
196    renderer = DummyRenderer()
197    _attach_queue(renderer, [_clip_item("sess_001")])
198
199    await asyncio.gather(
200        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
201        renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT),
202    )
203
204    assert len(renderer.llm_prompts) == 1
205
206
207async def test_clip_is_found_in_the_owning_sessions_queue() -> None:
208    """The session registry points the lookup at the queue that holds the clip."""
209    renderer = DummyRenderer()
210    session = SessionState(session_id="sess", station_id="st", queue_id="player_b")
211    renderer._sessions = {"sess": session}
212    _attach_queues(
213        renderer,
214        {"player_a": [], "player_b": [_clip_item("sess_001", queue_id="player_b")]},
215    )
216
217    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
218
219    assert streamdetails.item_id == "sess_001"
220
221
222async def test_clip_is_found_by_scanning_every_queue_after_a_restart() -> None:
223    """With the session registry gone, the clip is still located in its persisted queue."""
224    renderer = DummyRenderer()
225    _attach_queues(
226        renderer,
227        {"player_a": [], "player_b": [_clip_item("sess_001", queue_id="player_b")]},
228    )
229
230    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
231
232    assert streamdetails.item_id == "sess_001"
233
234
235async def test_unknown_clip_raises_media_not_found() -> None:
236    """A clip id that is not in any queue is reported as missing media."""
237    renderer = DummyRenderer()
238    _attach_queue(renderer, [_clip_item("sess_001")])
239
240    with pytest.raises(MediaNotFoundError):
241        await renderer.get_stream_details("sess_999", MediaType.SOUND_EFFECT)
242
243
244async def test_clip_without_prompt_raises_media_not_found() -> None:
245    """A clip whose attributes were lost is reported as missing media."""
246    renderer = DummyRenderer()
247    item = _clip_item("sess_001")
248    item.extra_attributes.pop(ATTR_PROMPT)
249    _attach_queue(renderer, [item])
250
251    with pytest.raises(MediaNotFoundError):
252        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
253
254
255async def test_llm_failure_raises_media_not_found() -> None:
256    """An LLM failure surfaces as missing media so the core skips the clip."""
257    renderer = DummyRenderer()
258    renderer.fail_generation = True
259    _attach_queue(renderer, [_clip_item("sess_001")])
260
261    with pytest.raises(MediaNotFoundError):
262        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
263
264
265async def test_render_failure_increments_the_session_skip_counter() -> None:
266    """A skipped clip is recorded on its session without failing the run."""
267    renderer = DummyRenderer()
268    session = SessionState(session_id="sess", station_id="st")
269    renderer._sessions = {"sess": session}
270    _attach_queue(renderer, [_clip_item("sess_001")])
271    renderer.fail_generation = True
272
273    with pytest.raises(MediaNotFoundError):
274        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
275
276    assert session.skipped_sections == 1
277    assert session.last_render_error
278
279
280async def test_tts_failure_raises_media_not_found_and_records_skip() -> None:
281    """A TTS failure surfaces as missing media and is recorded on the owning session."""
282
283    class UnspeakableRenderer(DummyRenderer):
284        async def _render_tts_media(self, text: str) -> tuple[str, StreamType, AudioFormat]:
285            raise RuntimeError("tts down")
286
287    renderer = UnspeakableRenderer()
288    session = SessionState(session_id="sess", station_id="st")
289    renderer._sessions = {"sess": session}
290    _attach_queue(renderer, [_clip_item("sess_001")])
291
292    with pytest.raises(MediaNotFoundError):
293        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
294
295    assert session.skipped_sections == 1
296    assert session.last_render_error
297
298
299async def test_probe_failure_is_not_fatal() -> None:
300    """A failed duration probe yields streamdetails without a duration."""
301
302    class UnprobableRenderer(DummyRenderer):
303        async def _probe_duration(self, path: str) -> int | None:
304            return None
305
306    renderer = UnprobableRenderer()
307    _attach_queue(renderer, [_clip_item("sess_001")])
308
309    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
310
311    assert streamdetails.duration is None
312
313
314class RealProbeRenderer(DummyRenderer):
315    """Harness that exercises the mixin's real duration probe instead of the stub."""
316
317    _probe_duration = AIRadioRenderMixin._probe_duration
318
319
320def _failing_probe(message: str, monkeypatch: pytest.MonkeyPatch) -> RealProbeRenderer:
321    """Build a renderer whose duration probe fails with the given ffprobe message."""
322
323    async def _raise(*_args: Any, **_kwargs: Any) -> AudioTags:
324        raise InvalidDataError(message)
325
326    monkeypatch.setattr("music_assistant.providers.ai_radio.rendering.async_parse_tags", _raise)
327    renderer = RealProbeRenderer()
328    renderer._sessions = {"sess": SessionState(session_id="sess", station_id="st")}
329    _attach_queue(renderer, [_clip_item("sess_001")])
330    return renderer
331
332
333async def test_tts_server_error_fails_the_clip_with_an_actionable_message(
334    monkeypatch: pytest.MonkeyPatch,
335) -> None:
336    """An engine that hands out a URL it cannot render fails the clip, not the playback."""
337    renderer = _failing_probe(
338        "Unable to retrieve info for http://ha.invalid/api/tts_proxy/1.mp3 "
339        "(Server returned 5XX Server Error reply)",
340        monkeypatch,
341    )
342
343    with pytest.raises(MediaNotFoundError):
344        await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
345
346    session = renderer._sessions["sess"]
347    assert session.skipped_sections == 1
348    assert "enough credit" in session.last_render_error
349
350
351async def test_unmeasurable_clip_still_plays(monkeypatch: pytest.MonkeyPatch) -> None:
352    """A probe that only fails to measure the audio leaves the clip playable."""
353    renderer = _failing_probe(
354        "Unable to retrieve info for http://ha.invalid/api/tts_proxy/1.mp3 "
355        "(Invalid or unsupported media file)",
356        monkeypatch,
357    )
358
359    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
360
361    assert streamdetails.duration is None
362    assert renderer._sessions["sess"].skipped_sections == 0
363
364
365async def test_render_tts_media_streams_a_url_over_http() -> None:
366    """A TTS engine returning a proxy URL yields an HTTP stream with the MP3 default format."""
367    renderer = _tts_renderer("http://example.test/api/tts_proxy/abc123.mp3")
368
369    path, stream_type, audio_format = await renderer._render_tts_media("hello world")
370
371    assert path == "http://example.test/api/tts_proxy/abc123.mp3"
372    assert stream_type == StreamType.HTTP
373    assert audio_format.content_type == ContentType.MP3
374    engine = cast("Any", renderer)._get_tts_engine.return_value
375    # the provider-scoped engine.id, never engine.uid, and never omitted
376    engine.provider.get_tts_message.assert_awaited_once_with("hello world", engine_id="tts.cloud")
377
378
379async def test_render_tts_media_streams_a_local_file_from_disk(tmp_path: Path) -> None:
380    """A TTS engine that renders to disk is played as a local file rather than fetched."""
381    clip = tmp_path / "section.mp3"
382    clip.write_bytes(b"")
383    renderer = _tts_renderer(str(clip))
384    _attach_queue(renderer, [_clip_item("sess_001")])
385
386    path, stream_type, _ = await renderer._render_tts_media("hello world")
387
388    assert path == str(clip)
389    assert stream_type == StreamType.LOCAL_FILE
390
391
392async def test_render_tts_media_keeps_a_declared_audio_format(tmp_path: Path) -> None:
393    """A TTS engine that declares its own format has it carried into the clip streamdetails."""
394    clip = tmp_path / "section.wav"
395    clip.write_bytes(b"")
396    renderer = _tts_renderer(str(clip), AudioFormat(content_type=ContentType.WAV))
397
398    _, _, audio_format = await renderer._render_tts_media("hello world")
399
400    assert audio_format.content_type == ContentType.WAV
401
402
403@pytest.mark.parametrize("path", ["", "section.mp3", "/does/not/exist.mp3"])
404async def test_render_tts_media_rejects_an_unplayable_path(path: str) -> None:
405    """A path that is neither a URL nor an existing file fails loudly instead of degrading."""
406    renderer = _tts_renderer(path)
407
408    with pytest.raises(InvalidDataError, match="unusable stream path"):
409        await renderer._render_tts_media("hello world")
410
411
412async def test_render_tts_media_gives_up_on_a_stalled_engine(
413    monkeypatch: pytest.MonkeyPatch,
414) -> None:
415    """A stalled TTS engine fails the clip instead of pinning the render path."""
416    monkeypatch.setattr(
417        "music_assistant.providers.ai_radio.rendering.TTS_QUERY_TIMEOUT_SECONDS", 0.01
418    )
419
420    async def _answers_too_late(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
421        await asyncio.sleep(5)
422        return SimpleNamespace(
423            path="http://example.test/late.mp3",
424            audio_format=AudioFormat(content_type=ContentType.MP3),
425        )
426
427    renderer = _tts_renderer("http://example.test/late.mp3")
428    engine = cast("Any", renderer)._get_tts_engine.return_value
429    engine.provider.get_tts_message = AsyncMock(side_effect=_answers_too_late)
430
431    with pytest.raises(MusicAssistantError, match="did not respond within"):
432        await renderer._render_tts_media("hello world")
433
434
435async def test_render_tts_media_reports_an_engine_side_timeout_as_is() -> None:
436    """A timeout raised by the TTS engine itself is not reported as our own cap."""
437    renderer = _tts_renderer("http://example.test/late.mp3")
438    engine = cast("Any", renderer)._get_tts_engine.return_value
439    engine.provider.get_tts_message = AsyncMock(side_effect=TimeoutError)
440
441    with pytest.raises(TimeoutError) as error:
442        await renderer._render_tts_media("hello world")
443    assert "did not respond within" not in str(error.value)
444
445
446async def test_local_file_clip_yields_local_file_streamdetails(tmp_path: Path) -> None:
447    """A disk-rendered clip reaches the core as LOCAL_FILE streamdetails."""
448    clip = tmp_path / "section.mp3"
449    clip.write_bytes(b"")
450    renderer = _tts_renderer(str(clip))
451    _attach_queue(renderer, [_clip_item("sess_001")])
452
453    streamdetails = await renderer.get_stream_details("sess_001", MediaType.SOUND_EFFECT)
454
455    assert streamdetails.stream_type == StreamType.LOCAL_FILE
456    assert streamdetails.path == str(clip)
457