music-assistant-server

1"""Tests for the Ambient Sounds provider."""
2
3import pathlib
4from typing import Any, cast
5from unittest.mock import AsyncMock, MagicMock, patch
6
7import pytest
8from music_assistant_models.enums import ContentType, MediaType, ProviderFeature, StreamType
9from music_assistant_models.errors import AudioError, InvalidDataError, MediaNotFoundError
10
11from music_assistant.providers import ambient_sounds
12from music_assistant.providers.ambient_sounds import (
13    CONF_KEY_CUSTOM_SOUNDS,
14    PRESETS,
15    AmbientSoundsProvider,
16)
17
18
19def _create_provider(cache_path: str) -> AmbientSoundsProvider:
20    """
21    Create an AmbientSoundsProvider with mocked dependencies.
22
23    :param cache_path: The cache directory the provider renders loop files into.
24    """
25    with patch.object(AmbientSoundsProvider, "__init__", lambda *_a, **_kw: None):
26        provider = AmbientSoundsProvider.__new__(AmbientSoundsProvider)
27
28    provider.config = MagicMock()
29    provider.config.instance_id = "ambient_sounds"
30    provider.config.setup_data = {}
31    # no declared config entries: get_setup_value's fallback resolves to the default
32    provider.config.values = {}
33    provider.config.get_value = lambda _key, default=None: default
34    provider.manifest = MagicMock()
35    provider.manifest.domain = "ambient_sounds"
36    provider.mass = MagicMock()
37    provider.mass.cache_path = cache_path
38    provider.logger = MagicMock()
39    # in-memory, path-aware stand-in for persistent config storage: the provider
40    # stores its custom sounds in its own setup_data via the base class helpers
41    config_store: dict[str, Any] = {"providers": {"ambient_sounds": {"setup_data": {}}}}
42
43    def _config_get(key: str, default: Any = None) -> Any:
44        parent: Any = config_store
45        for part in key.split("/"):
46            if not isinstance(parent, dict) or part not in parent:
47                return default
48            parent = parent[part]
49        return parent
50
51    def _config_set(key: str, value: Any, immediate: bool = False) -> None:  # noqa: ARG001
52        parts = key.split("/")
53        parent = config_store
54        for part in parts[:-1]:
55            parent = parent.setdefault(part, {})
56        parent[parts[-1]] = value
57
58    provider.mass.config.get = _config_get
59    provider.mass.config.set = _config_set
60    provider.mass.config.encrypt_string = lambda value: value
61    provider.mass.config.decrypt_string = lambda value: value
62    provider.mass.cache.get = AsyncMock(return_value=None)
63    provider.mass.cache.set = AsyncMock()
64    provider.mass.cache.delete = AsyncMock()
65    provider._unregister_handles = []
66
67    return provider
68
69
70def _mock_media_info(duration: float | None = 3600.0, icyname: str | None = None) -> MagicMock:
71    """Create a mocked AudioTags result for a probed custom sound url."""
72    media_info = MagicMock()
73    media_info.duration = duration
74    media_info.format = "mp3"
75    media_info.sample_rate = 44100
76    media_info.bits_per_sample = 16
77    media_info.channels = 2
78    media_info.raw = {}
79    media_info.get = lambda key, default=None: {"icyname": icyname}.get(key, default)
80    return media_info
81
82
83async def test_sound_effects_enumeration(tmp_path: pathlib.Path) -> None:
84    """All presets are enumerated as sound effect items with correct metadata."""
85    provider = _create_provider(str(tmp_path))
86    items = [item async for item in provider.get_sound_effects()]
87    assert len(items) == len(PRESETS)
88    assert {item.item_id for item in items} == set(PRESETS)
89    for item in items:
90        assert item.media_type == MediaType.SOUND_EFFECT
91        assert item.name
92        assert item.translation_key == item.item_id
93        assert item.metadata.description
94        assert item.duration == ambient_sounds.LOOP_DURATION
95
96
97async def test_get_sound_effect(tmp_path: pathlib.Path) -> None:
98    """A single sound effect resolves by id; unknown ids raise MediaNotFoundError."""
99    provider = _create_provider(str(tmp_path))
100    item = await provider.get_sound_effect("white_noise")
101    assert item.name == "White noise"
102    assert item.provider_mappings
103    with pytest.raises(MediaNotFoundError):
104        await provider.get_sound_effect("unknown_preset")
105
106
107async def test_supported_features() -> None:
108    """The provider advertises sound effects and browse, no library features."""
109    assert ProviderFeature.SOUND_EFFECTS in ambient_sounds.SUPPORTED_FEATURES
110    assert ProviderFeature.BROWSE in ambient_sounds.SUPPORTED_FEATURES
111    assert not any(x for x in ambient_sounds.SUPPORTED_FEATURES if x.value.startswith("library_"))
112
113
114async def test_stream_details_renders_loop(
115    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
116) -> None:
117    """Streamdetails render the loop file once and reuse it on subsequent calls."""
118    # shrink the loop so the test renders quickly
119    monkeypatch.setattr(ambient_sounds, "LOOP_DURATION", 2)
120    monkeypatch.setattr(ambient_sounds, "CROSSFADE_DURATION", 1)
121    provider = _create_provider(str(tmp_path))
122    await provider.handle_async_init()
123
124    stream_details = await provider.get_stream_details("ocean_waves")
125    assert stream_details.stream_type == StreamType.LOCAL_FILE
126    assert stream_details.media_type == MediaType.SOUND_EFFECT
127    assert stream_details.duration == 2
128    assert isinstance(stream_details.path, str)
129    loop_file = pathlib.Path(stream_details.path)
130    assert loop_file.is_file()
131    assert loop_file.read_bytes()[:4] == b"fLaC"
132    # no leftover temp file from the render
133    assert not loop_file.with_name(f"{loop_file.name}.tmp").exists()
134
135    # a second call must reuse the rendered file instead of rendering again
136    mtime = loop_file.stat().st_mtime
137    stream_details_2 = await provider.get_stream_details("ocean_waves")
138    assert stream_details_2.path == stream_details.path
139    assert loop_file.stat().st_mtime == mtime
140
141
142async def test_stream_details_unknown_preset(tmp_path: pathlib.Path) -> None:
143    """Streamdetails for an unknown preset raise MediaNotFoundError."""
144    provider = _create_provider(str(tmp_path))
145    await provider.handle_async_init()
146    with pytest.raises(MediaNotFoundError):
147        await provider.get_stream_details("unknown_preset")
148
149
150async def test_add_custom_sound(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
151    """A custom sound is probed, stored and enumerated alongside the presets."""
152    monkeypatch.setattr(
153        ambient_sounds, "async_parse_tags", AsyncMock(return_value=_mock_media_info())
154    )
155    provider = _create_provider(str(tmp_path))
156    url = "https://example.com/sounds/rain.mp3"
157
158    item = await provider.add_sound(url, "Rain")
159    assert item.item_id == url
160    assert item.name == "Rain"
161    assert item.duration == 3600
162    assert item.media_type == MediaType.SOUND_EFFECT
163
164    items = [x async for x in provider.get_sound_effects()]
165    assert len(items) == len(PRESETS) + 1
166    assert (await provider.get_sound_effect(url)).name == "Rain"
167
168    # adding the same url again replaces the stored entry instead of duplicating it
169    await provider.add_sound(url, "Rain 2")
170    stored = provider._stored_sounds()
171    assert len(stored) == 1
172    assert stored[0]["name"] == "Rain 2"
173
174
175async def test_add_custom_sound_invalid_url(
176    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
177) -> None:
178    """An url that fails to probe is rejected and not stored."""
179    # async_parse_tags raises InvalidDataError (not AudioError) on ffprobe failures;
180    # the frontend relies on that exact error code to show its friendly message
181    monkeypatch.setattr(
182        ambient_sounds, "async_parse_tags", AsyncMock(side_effect=InvalidDataError("not audio"))
183    )
184    provider = _create_provider(str(tmp_path))
185    with pytest.raises(InvalidDataError):
186        await provider.add_sound("https://example.com/not_audio", "Broken")
187    assert not provider.get_setup_value(CONF_KEY_CUSTOM_SOUNDS)
188
189
190async def test_add_custom_sound_rejects_non_http_urls(
191    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
192) -> None:
193    """Only absolute http(s) urls reach ffprobe; anything else is rejected upfront."""
194    parse_tags_mock = AsyncMock(return_value=_mock_media_info())
195    monkeypatch.setattr(ambient_sounds, "async_parse_tags", parse_tags_mock)
196    provider = _create_provider(str(tmp_path))
197    for bad_url in (
198        "file:///etc/passwd",
199        "/etc/passwd",
200        "concat:file1|file2",
201        "ftp://example.com/sound.mp3",
202        "https://",
203    ):
204        with pytest.raises(InvalidDataError):
205            await provider.add_sound(bad_url, "Evil")
206    parse_tags_mock.assert_not_awaited()
207    assert not provider.get_setup_value(CONF_KEY_CUSTOM_SOUNDS)
208
209
210async def test_remove_custom_sound(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
211    """A removed custom sound is no longer enumerated or resolvable."""
212    monkeypatch.setattr(
213        ambient_sounds, "async_parse_tags", AsyncMock(return_value=_mock_media_info())
214    )
215    provider = _create_provider(str(tmp_path))
216    url = "https://example.com/sounds/rain.mp3"
217    await provider.add_sound(url, "Rain")
218
219    await provider.remove_sound(url)
220    items = [x async for x in provider.get_sound_effects()]
221    assert len(items) == len(PRESETS)
222    with pytest.raises(MediaNotFoundError):
223        await provider.get_sound_effect(url)
224
225
226async def test_custom_sound_stream_details(
227    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
228) -> None:
229    """Streamdetails for a custom sound point at its url with probed format info."""
230    monkeypatch.setattr(
231        ambient_sounds, "async_parse_tags", AsyncMock(return_value=_mock_media_info())
232    )
233    provider = _create_provider(str(tmp_path))
234    url = "https://example.com/sounds/rain.mp3"
235    await provider.add_sound(url, "Rain")
236
237    stream_details = await provider.get_stream_details(url)
238    assert stream_details.stream_type == StreamType.HTTP
239    assert stream_details.media_type == MediaType.SOUND_EFFECT
240    assert stream_details.path == url
241    assert stream_details.duration == 3600
242    assert stream_details.audio_format.content_type == ContentType.MP3
243    assert stream_details.can_seek
244    assert stream_details.allow_seek
245
246
247async def test_custom_sound_live_stream_not_seekable(
248    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
249) -> None:
250    """An endless (radio-style) stream without duration is marked as not seekable."""
251    monkeypatch.setattr(
252        ambient_sounds,
253        "async_parse_tags",
254        AsyncMock(return_value=_mock_media_info(duration=None, icyname="Some Radio")),
255    )
256    provider = _create_provider(str(tmp_path))
257    url = "https://example.com/streams/radio"
258    await provider.add_sound(url, "Radio")
259
260    stream_details = await provider.get_stream_details(url)
261    assert stream_details.duration is None
262    assert not stream_details.can_seek
263    assert not stream_details.allow_seek
264
265
266async def test_remove_custom_sound_clears_cached_media_info(
267    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
268) -> None:
269    """Removing a custom sound also drops its cached media info."""
270    monkeypatch.setattr(
271        ambient_sounds, "async_parse_tags", AsyncMock(return_value=_mock_media_info())
272    )
273    provider = _create_provider(str(tmp_path))
274    url = "https://example.com/sounds/rain.mp3"
275    await provider.add_sound(url, "Rain")
276
277    await provider.remove_sound(url)
278    cache_delete = cast("AsyncMock", provider.mass.cache.delete)
279    cache_delete.assert_awaited_once_with(
280        url, provider=provider.instance_id, category=ambient_sounds.CACHE_CATEGORY_MEDIA_INFO
281    )
282
283
284async def test_unload_unregisters_api_commands(tmp_path: pathlib.Path) -> None:
285    """Unloading the provider unregisters its API commands so a reload can re-register."""
286    provider = _create_provider(str(tmp_path))
287    handles = [MagicMock(), MagicMock()]
288    provider._unregister_handles.extend(handles)
289
290    await provider.unload()
291    for handle in handles:
292        handle.assert_called_once()
293    assert not provider._unregister_handles
294
295
296async def test_failed_render_leaves_no_temp_file(
297    tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
298) -> None:
299    """A failed render raises AudioError and cleans up its partial temp file."""
300
301    async def fake_check_output(*args: str) -> tuple[int, bytes]:
302        # simulate ffmpeg dying halfway: partial output written, non-zero exit
303        pathlib.Path(args[-1]).write_bytes(b"partial")
304        return 1, b"boom"
305
306    monkeypatch.setattr(ambient_sounds, "check_output", fake_check_output)
307    provider = _create_provider(str(tmp_path))
308    await provider.handle_async_init()
309    with pytest.raises(AudioError):
310        await provider.get_stream_details("white_noise")
311    assert not list(tmp_path.rglob("*.tmp"))
312