music-assistant-server

1"""Tests for the CloudFileSystemProvider base class."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from unittest.mock import AsyncMock, MagicMock
7from urllib.parse import quote
8
9import pytest
10from aiohttp import web
11from aiohttp.test_utils import make_mocked_request
12from music_assistant_models.errors import MediaNotFoundError, ProviderUnavailableError
13
14from music_assistant.providers.filesystem_cloud.base import CloudFileSystemProvider, RawItem
15from music_assistant.providers.filesystem_local.helpers import FileSystemItem, ScanErrors
16
17if TYPE_CHECKING:
18    from aiohttp import ClientResponse
19
20BASE_URL = "http://ma.local:8097"
21INSTANCE_ID = "cloud--test"
22ROOT_ID = "root-id"
23
24# a small two-level tree: /Artist/track.mp3 + /cover.jpg
25TREE: dict[str, list[RawItem]] = {
26    ROOT_ID: [
27        ("artist-id", "Artist", True, "2026-01-01", None),
28        ("cover-id", "cover.jpg", False, "2026-01-01", 4),
29        ("notes-id", "notes.txt", False, "2026-01-01", 2),
30    ],
31    "artist-id": [
32        ("track-id", "track.mp3", False, "2026-01-02", 1000),
33    ],
34}
35FILE_DATA = {"cover-id": b"image-bytes", "track-id": b"audio-bytes", "notes-id": b"n"}
36
37
38class _StubCloudProvider(CloudFileSystemProvider):
39    """CloudFileSystemProvider wired to an in-memory folder tree."""
40
41    tree: dict[str, list[RawItem]]
42    file_data: dict[str, bytes]
43    fail_list: set[str]
44    fail_download: set[str]
45    list_calls: list[str]
46    download_headers: dict[str, str] | None
47
48    async def _api_list_children(self, folder_id: str) -> list[RawItem]:
49        self.list_calls.append(folder_id)
50        if folder_id in self.fail_list:
51            raise ProviderUnavailableError(f"list failed: {folder_id}")
52        return self.tree.get(folder_id, [])
53
54    async def _api_download_bytes(self, file_id: str) -> bytes:
55        if file_id in self.fail_download:
56            raise ProviderUnavailableError(f"download failed: {file_id}")
57        return self.file_data[file_id]
58
59    async def _api_download_response(self, file_id: str, headers: dict[str, str]) -> ClientResponse:
60        self.download_headers = headers
61        raise ProviderUnavailableError(f"download failed: {file_id}")
62
63
64def _make_provider(tree: dict[str, list[RawItem]] | None = None) -> _StubCloudProvider:
65    provider = _StubCloudProvider.__new__(_StubCloudProvider)
66    provider.root_folder_id = ROOT_ID
67    provider._dir_cache = {}
68    provider._dir_cache_expiry = {}
69    provider._unregister_stream_route = None
70    provider.logger = MagicMock()
71    provider.media_content_type = "music"
72    provider.config = MagicMock()
73    provider.config.instance_id = INSTANCE_ID
74    provider.config.get_value = MagicMock(return_value=False)
75    provider.mass = MagicMock()
76    provider.mass.streams.base_url = BASE_URL
77    provider.tree = TREE if tree is None else tree
78    provider.file_data = FILE_DATA
79    provider.fail_list = set()
80    provider.fail_download = set()
81    provider.list_calls = []
82    provider.download_headers = None
83    return provider
84
85
86async def _run_enumerate(
87    provider: _StubCloudProvider,
88    *,
89    scan_errors: ScanErrors | None = None,
90) -> None:
91    """Drive _enumerate_files_for_sync with empty sync buckets."""
92    await provider._enumerate_files_for_sync(
93        file_checksums={},
94        cue_file_checksums={},
95        cur_filenames=set(),
96        items_to_process=[],
97        unchanged_cue_items=[],
98        cue_stems=set(),
99        scan_errors=scan_errors if scan_errors is not None else ScanErrors(),
100    )
101
102
103async def test_scandir_converts_raw_items() -> None:
104    """Raw API entries become FileSystemItems; files stream via the dynamic URL."""
105    provider = _make_provider()
106
107    items = {item.filename: item for item in await provider._scandir("")}
108
109    folder = items["Artist"]
110    assert folder.is_dir
111    assert folder.relative_path == "Artist"
112    assert folder.absolute_path == ""
113    image = items["cover.jpg"]
114    assert not image.is_dir
115    assert image.checksum == "2026-01-01"
116    assert image.file_size == 4
117    assert image.absolute_path == f"{BASE_URL}/{INSTANCE_ID}_stream?path={quote('cover.jpg')}"
118    # the listing is cached for path lookups
119    assert "cover.jpg" in provider._dir_cache[""]
120
121
122async def test_scandir_skips_duplicate_names() -> None:
123    """Clouds may allow duplicate names in a folder; only the first wins."""
124    provider = _make_provider(
125        tree={ROOT_ID: [("id1", "a.mp3", False, "1", 1), ("id2", "a.mp3", False, "2", 2)]}
126    )
127    logger = MagicMock()
128    provider.logger = logger
129
130    items = await provider._scandir("")
131
132    assert len(items) == 1
133    assert provider._dir_cache[""]["a.mp3"][0] == "id1"
134    logger.warning.assert_called_once()
135
136
137async def test_scandir_sanitizes_slashes_in_names() -> None:
138    """Slashes in cloud file names would corrupt the path scheme."""
139    provider = _make_provider(tree={ROOT_ID: [("id1", "AC/DC", True, "1", None)]})
140
141    items = await provider._scandir("")
142
143    assert items[0].relative_path == "AC_DC"
144
145
146async def test_scandir_serves_recent_listing_from_cache() -> None:
147    """Re-listing a folder within the TTL must not hit the API again."""
148    provider = _make_provider()
149
150    first = await provider._scandir("")
151    second = await provider._scandir("")
152
153    assert provider.list_calls == [ROOT_ID]
154    assert [item.relative_path for item in second] == [item.relative_path for item in first]
155
156
157async def test_scandir_refetches_after_ttl_expiry() -> None:
158    """An expired listing is fetched fresh from the API."""
159    provider = _make_provider()
160
161    await provider._scandir("")
162    provider._dir_cache_expiry[""] = 0.0
163    await provider._scandir("")
164
165    assert provider.list_calls == [ROOT_ID, ROOT_ID]
166
167
168async def test_sync_walk_bypasses_listing_cache() -> None:
169    """A sync must always fetch fresh listings, even right after browsing."""
170    provider = _make_provider()
171    provider._classify_scan_item = MagicMock()  # type: ignore[method-assign]
172
173    await provider._scandir("")
174    await _run_enumerate(provider)
175
176    # root listed once by browse and again (fresh) by the sync walk
177    assert provider.list_calls.count(ROOT_ID) == 2
178
179
180async def test_lookup_answers_from_cache() -> None:
181    """Resolving the same path twice must not re-list the folder."""
182    provider = _make_provider()
183
184    first = await provider.resolve("Artist/track.mp3")
185    calls_after_first = len(provider.list_calls)
186    second = await provider.resolve("Artist/track.mp3")
187
188    assert first.relative_path == second.relative_path == "Artist/track.mp3"
189    # resolving walked root + Artist once; the second resolve was free
190    assert provider.list_calls == [ROOT_ID, "artist-id"]
191    assert len(provider.list_calls) == calls_after_first
192
193
194async def test_parent_folder_helpers_usable_on_cloud_items() -> None:
195    """
196    Parent-folder helpers must work although absolute_path is a stream URL.
197
198    The audiobook/podcast logic in LocalFileSystemProvider scans an item's
199    parent folder and uses its name; both must be based on the relative path.
200    """
201    provider = _make_provider()
202
203    track = await provider.resolve("Artist/track.mp3")
204
205    assert track.relative_parent_path == "Artist"
206    assert track.parent_name == "Artist"
207    items = await provider._scandir(track.relative_parent_path)
208    assert [item.filename for item in items] == ["track.mp3"]
209
210
211async def test_resolve_id_of_root_and_unknown_path() -> None:
212    """The empty path is the root folder; unknown paths raise MediaNotFoundError."""
213    provider = _make_provider()
214
215    assert await provider._resolve_id("") == ROOT_ID
216    with pytest.raises(MediaNotFoundError):
217        await provider._resolve_id("Artist/nope.mp3")
218
219
220async def test_resolve_missing_raises() -> None:
221    """Resolving a missing path raises MediaNotFoundError."""
222    provider = _make_provider()
223
224    with pytest.raises(MediaNotFoundError):
225        await provider.resolve("missing.mp3")
226
227
228async def test_exists() -> None:
229    """exists() is True for present paths, False for absent/empty ones."""
230    provider = _make_provider()
231
232    assert await provider.exists("Artist/track.mp3")
233    assert not await provider.exists("Artist/nope.mp3")
234    assert not await provider.exists("")
235
236
237async def test_exists_returns_false_on_api_failure() -> None:
238    """An API failure while probing must read as 'does not exist', not crash."""
239    provider = _make_provider()
240    provider.fail_list = {ROOT_ID}
241
242    assert not await provider.exists("Artist/track.mp3")
243
244
245def test_normalize_path() -> None:
246    """Paths are stripped and ./.. segments collapsed."""
247    provider = _make_provider()
248
249    assert provider._normalize_path("/Artist/track.mp3/") == "Artist/track.mp3"
250    assert provider._normalize_path("Artist/./track.mp3") == "Artist/track.mp3"
251    assert provider._normalize_path("Artist/../cover.jpg") == "cover.jpg"
252    assert provider._normalize_path(".") == ""
253
254
255async def test_read_file() -> None:
256    """_read_file returns the file bytes for a relative path."""
257    provider = _make_provider()
258
259    assert await provider._read_file("cover.jpg") == b"image-bytes"
260
261
262async def test_read_file_failure_maps_to_media_not_found() -> None:
263    """A failed download surfaces as MediaNotFoundError to the inherited callers."""
264    provider = _make_provider()
265    provider.fail_download = {"cover-id"}
266
267    with pytest.raises(MediaNotFoundError):
268        await provider._read_file("cover.jpg")
269
270
271async def test_resolve_image_strips_cache_suffix() -> None:
272    """The ?cs= suffix added for embedded images must be dropped before lookup."""
273    provider = _make_provider()
274
275    assert await provider.resolve_image("cover.jpg?cs=12345") == b"image-bytes"
276
277
278async def test_resolve_image_extracts_embedded_art(monkeypatch: pytest.MonkeyPatch) -> None:
279    """An audio file path means embedded art, extracted over the stream URL."""
280    provider = _make_provider()
281    extract = AsyncMock(return_value=b"art-bytes")
282    monkeypatch.setattr(
283        "music_assistant.providers.filesystem_cloud.base.get_embedded_image", extract
284    )
285
286    result = await provider.resolve_image("Artist/track.mp3?cs=999")
287
288    assert result == b"art-bytes"
289    extract.assert_awaited_once_with(provider._stream_url("Artist/track.mp3"))
290
291
292async def test_resolve_image_embedded_art_missing(monkeypatch: pytest.MonkeyPatch) -> None:
293    """No embedded art in the audio file raises MediaNotFoundError."""
294    provider = _make_provider()
295    monkeypatch.setattr(
296        "music_assistant.providers.filesystem_cloud.base.get_embedded_image",
297        AsyncMock(return_value=None),
298    )
299
300    with pytest.raises(MediaNotFoundError):
301        await provider.resolve_image("Artist/track.mp3")
302
303
304async def test_enumerate_classifies_supported_files_only() -> None:
305    """The sync walk visits all folders and classifies only supported extensions."""
306    provider = _make_provider()
307    provider._classify_scan_item = MagicMock()  # type: ignore[method-assign]
308
309    await _run_enumerate(provider)
310
311    # cover.jpg and notes.txt are not sync candidates; folder art is handled separately
312    classified = [
313        call.args[0].relative_path for call in provider._classify_scan_item.call_args_list
314    ]
315    assert classified == ["Artist/track.mp3"]
316
317
318async def test_enumerate_subfolder_error_is_skipped() -> None:
319    """A failing subfolder is logged and skipped; the rest of the sync continues."""
320    provider = _make_provider(
321        tree={
322            ROOT_ID: [
323                ("bad-id", "Bad", True, "1", None),
324                ("artist-id", "Artist", True, "1", None),
325            ],
326            "artist-id": [("track-id", "track.mp3", False, "1", 1)],
327        }
328    )
329    provider.fail_list = {"bad-id"}
330    provider._classify_scan_item = MagicMock()  # type: ignore[method-assign]
331    logger = MagicMock()
332    provider.logger = logger
333    scan_errors = ScanErrors()
334
335    await _run_enumerate(provider, scan_errors=scan_errors)
336
337    assert not scan_errors.fatal
338    assert scan_errors.failed_dirs == 1
339    logger.warning.assert_called_once()
340    assert provider._classify_scan_item.call_count == 1
341
342
343async def test_enumerate_root_error_aborts() -> None:
344    """A root-level listing failure is reported so the sync aborts."""
345    provider = _make_provider()
346    provider.fail_list = {ROOT_ID}
347    scan_errors = ScanErrors()
348
349    await _run_enumerate(provider, scan_errors=scan_errors)
350
351    assert scan_errors.fatal is not None
352
353
354async def test_is_reachable_asks_the_api_for_the_root() -> None:
355    """Cloud storage has no local path to stat, so reachability is a live root listing."""
356    provider = _make_provider()
357
358    assert await provider._is_reachable() is True
359
360    provider.fail_list = {ROOT_ID}
361    with pytest.raises(ProviderUnavailableError):
362        await provider._is_reachable()
363
364
365async def test_enumerate_stops_on_directory_cycle() -> None:
366    """A path re-appearing during the walk must be visited only once."""
367    provider = _make_provider()
368    listing = {
369        "": [FileSystemItem("A", "A", "", is_dir=True)],
370        "A": [FileSystemItem("B", "A/B", "", is_dir=True)],
371        "A/B": [FileSystemItem("A", "A", "", is_dir=True)],
372    }
373    scanned: list[str] = []
374
375    async def fake_scandir(path: str, use_cache: bool = True) -> list[FileSystemItem]:
376        assert use_cache is False
377        scanned.append(path)
378        return listing[path]
379
380    provider._scandir = AsyncMock(side_effect=fake_scandir)  # type: ignore[method-assign]
381
382    await _run_enumerate(provider)
383
384    assert sorted(scanned) == ["", "A", "A/B"]
385
386
387async def test_stream_request_without_path_is_bad_request() -> None:
388    """The stream route rejects requests without a path parameter."""
389    provider = _make_provider()
390    request = make_mocked_request("GET", f"/{INSTANCE_ID}_stream")
391
392    with pytest.raises(web.HTTPBadRequest):
393        await provider._handle_stream_request(request)
394
395
396async def test_stream_request_unknown_path_is_not_found() -> None:
397    """The stream route returns 404 for paths that don't resolve."""
398    provider = _make_provider()
399    request = make_mocked_request("GET", f"/{INSTANCE_ID}_stream?path=nope.mp3")
400
401    with pytest.raises(web.HTTPNotFound):
402        await provider._handle_stream_request(request)
403
404
405@pytest.mark.parametrize("path", ["cover.jpg", "notes.txt", "playlist.m3u"])
406async def test_stream_request_non_audio_extension_is_not_found(path: str) -> None:
407    """The stream route refuses non-audio files so it can't leak arbitrary cloud files."""
408    provider = _make_provider()
409    request = make_mocked_request("GET", f"/{INSTANCE_ID}_stream?path={quote(path)}")
410
411    with pytest.raises(web.HTTPNotFound):
412        await provider._handle_stream_request(request)
413
414    # rejected before any cloud API lookup
415    assert provider.list_calls == []
416
417
418async def test_stream_request_forwards_range_and_maps_api_failure() -> None:
419    """The Range header is forwarded to the download; API failure becomes a 502."""
420    provider = _make_provider()
421    request = make_mocked_request(
422        "GET",
423        f"/{INSTANCE_ID}_stream?path={quote('Artist/track.mp3')}",
424        headers={"Range": "bytes=100-"},
425    )
426
427    with pytest.raises(web.HTTPBadGateway):
428        await provider._handle_stream_request(request)
429
430    assert provider.download_headers == {"Range": "bytes=100-"}
431
432
433async def test_post_init_registers_stream_route_and_unload_removes_it() -> None:
434    """_post_init registers the dynamic stream route; unload unregisters it."""
435    provider = _make_provider()
436    unregister = MagicMock()
437    provider.mass.streams.register_dynamic_route = MagicMock(return_value=unregister)
438
439    await provider._post_init()
440
441    provider.mass.streams.register_dynamic_route.assert_called_once()
442    assert provider.mass.streams.register_dynamic_route.call_args.args[0] == (
443        f"/{INSTANCE_ID}_stream"
444    )
445
446    await provider.unload()
447    unregister.assert_called_once()
448
449
450async def test_unload_without_registered_route() -> None:
451    """Unload before _post_init (e.g. failed setup) must not raise."""
452    provider = _make_provider()
453
454    await provider.unload()
455
456
457# tree with a playlist folder next to the music: /Playlists/list.m3u + /Artist/track.mp3
458PLAYLIST_TREE: dict[str, list[RawItem]] = {
459    ROOT_ID: [
460        ("playlists-id", "Playlists", True, "2026-01-01", None),
461        ("artist-id", "Artist", True, "2026-01-01", None),
462    ],
463    "playlists-id": [
464        ("list-id", "list.m3u", False, "2026-01-01", 10),
465    ],
466    "artist-id": [
467        ("track-id", "track.mp3", False, "2026-01-02", 1000),
468    ],
469}
470
471
472def _make_playlist_provider(
473    monkeypatch: pytest.MonkeyPatch,
474) -> tuple[_StubCloudProvider, MagicMock]:
475    """Return a provider on PLAYLIST_TREE with tag parsing and track building stubbed."""
476    provider = _make_provider(tree=PLAYLIST_TREE)
477    # the inherited implementation lives in (and imports from) filesystem_local
478    monkeypatch.setattr(
479        "music_assistant.providers.filesystem_local.async_parse_tags",
480        AsyncMock(return_value=MagicMock()),
481    )
482    track = MagicMock(name="track")
483    provider._parse_track = AsyncMock(return_value=track)  # type: ignore[method-assign]
484    return provider, track
485
486
487async def test_parse_playlist_line_relative_to_playlist_folder(
488    monkeypatch: pytest.MonkeyPatch,
489) -> None:
490    """A line relative to the playlist's own folder resolves through the cloud lookup."""
491    provider, track = _make_playlist_provider(monkeypatch)
492
493    result = await provider._parse_playlist_line("../Artist/track.mp3", "Playlists")
494
495    assert result is track
496    file_item = provider._parse_track.await_args.args[0]  # type: ignore[attr-defined]
497    assert file_item.relative_path == "Artist/track.mp3"
498
499
500async def test_parse_playlist_line_sibling_of_playlist(monkeypatch: pytest.MonkeyPatch) -> None:
501    """A bare filename next to the playlist file resolves relative to its folder."""
502    provider, track = _make_playlist_provider(monkeypatch)
503
504    assert await provider._parse_playlist_line("track.mp3", "Artist") is track
505
506
507async def test_parse_playlist_line_relative_to_root(monkeypatch: pytest.MonkeyPatch) -> None:
508    """A root-relative line still resolves after the playlist-folder candidate misses."""
509    provider, track = _make_playlist_provider(monkeypatch)
510
511    assert await provider._parse_playlist_line("Artist/track.mp3", "Playlists") is track
512
513
514async def test_parse_playlist_line_missing_file_returns_none(
515    monkeypatch: pytest.MonkeyPatch,
516) -> None:
517    """A line pointing at a nonexistent file logs a warning instead of raising."""
518    provider, _ = _make_playlist_provider(monkeypatch)
519
520    assert await provider._parse_playlist_line("Missing/nope.mp3", "Playlists") is None
521    provider.logger.warning.assert_called_once()  # type: ignore[attr-defined]
522