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