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