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