/
/
/
1"""Tests for filesystem provider sync behavior when the storage misbehaves."""
2
3import errno
4from typing import Any
5from unittest.mock import AsyncMock, MagicMock, patch
6
7from music_assistant_models.enums import MediaType
8
9from music_assistant.providers.filesystem_local import LocalFileSystemProvider
10from music_assistant.providers.filesystem_local.constants import (
11 CONF_ENTRY_CONTENT_TYPE,
12 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
13 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
14)
15from music_assistant.providers.filesystem_local.cue import make_cue_track_id
16from music_assistant.providers.filesystem_local.helpers import ScanErrors
17
18# two previously indexed tracks; the scans below only find the first one, so the
19# second is what a deletion pass would remove from the library
20FOUND_FILE = "Artist/Album/track1.mp3"
21MISSING_FILE = "Artist/Album/track2.mp3"
22
23
24def _create_provider() -> LocalFileSystemProvider:
25 """Create a music LocalFileSystemProvider with mocked dependencies."""
26 config_values = {
27 CONF_ENTRY_CONTENT_TYPE.key: "music",
28 CONF_ENTRY_LIBRARY_SYNC_TRACKS.key: True,
29 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS.key: True,
30 }
31 mock_config = MagicMock()
32 mock_config.get_value = MagicMock(side_effect=lambda key: config_values.get(key))
33
34 with patch.object(LocalFileSystemProvider, "__init__", lambda *_a, **_kw: None):
35 provider = LocalFileSystemProvider.__new__(LocalFileSystemProvider)
36
37 provider.config = mock_config
38 provider.media_content_type = "music"
39 provider.sync_running = False
40 provider.logger = MagicMock()
41 provider.available = True
42 provider.mass = MagicMock()
43 provider.mass.music.database.get_rows_from_query = AsyncMock(
44 return_value=[
45 {"provider_item_id": FOUND_FILE, "details": "1"},
46 {"provider_item_id": MISSING_FILE, "details": "1"},
47 ]
48 )
49 provider._process_deletions = AsyncMock() # type: ignore[method-assign]
50 provider._process_orphaned_albums_and_artists = AsyncMock() # type: ignore[method-assign]
51 provider._set_available = MagicMock() # type: ignore[method-assign]
52 return provider
53
54
55def _enumerate_result(
56 *,
57 failed_dirs: int = 0,
58 failed_entries: int = 0,
59 fatal: bool = False,
60 found_files: set[str] | None = None,
61) -> Any:
62 """Build an _enumerate_files_for_sync stub with the given scan outcome."""
63
64 async def _enumerate(**kwargs: Any) -> None:
65 scan_errors: ScanErrors = kwargs["scan_errors"]
66 scan_errors.failed_dirs = failed_dirs
67 scan_errors.failed_entries = failed_entries
68 if fatal:
69 scan_errors.fatal = OSError("storage gone")
70 kwargs["cur_filenames"].update(found_files or set())
71
72 return AsyncMock(side_effect=_enumerate)
73
74
75async def test_deletions_run_on_clean_scan() -> None:
76 """A scan without errors processes deletions as usual."""
77 provider = _create_provider()
78 provider._enumerate_files_for_sync = _enumerate_result( # type: ignore[method-assign]
79 found_files={FOUND_FILE}
80 )
81
82 await provider.sync_library(MediaType.TRACK)
83
84 provider._process_deletions.assert_awaited_once_with({MISSING_FILE}) # type: ignore[attr-defined]
85 provider._process_orphaned_albums_and_artists.assert_awaited_once() # type: ignore[attr-defined]
86
87
88async def test_deletions_skipped_when_directories_failed() -> None:
89 """A scan that could not read some directories must not delete their content."""
90 provider = _create_provider()
91 provider._enumerate_files_for_sync = _enumerate_result( # type: ignore[method-assign]
92 failed_dirs=3, found_files={FOUND_FILE}
93 )
94
95 await provider.sync_library(MediaType.TRACK)
96
97 provider._process_deletions.assert_not_called() # type: ignore[attr-defined]
98 provider._process_orphaned_albums_and_artists.assert_not_called() # type: ignore[attr-defined]
99 # the storage itself is reachable, so the provider stays available
100 provider._set_available.assert_called_once_with(True) # type: ignore[attr-defined]
101
102
103async def test_deletions_skipped_when_files_failed() -> None:
104 """A scan that could not read some files must not delete them either."""
105 provider = _create_provider()
106 provider._enumerate_files_for_sync = _enumerate_result( # type: ignore[method-assign]
107 failed_entries=2, found_files={FOUND_FILE}
108 )
109
110 await provider.sync_library(MediaType.TRACK)
111
112 provider._process_deletions.assert_not_called() # type: ignore[attr-defined]
113 provider._process_orphaned_albums_and_artists.assert_not_called() # type: ignore[attr-defined]
114 provider._set_available.assert_called_once_with(True) # type: ignore[attr-defined]
115
116
117async def test_failed_item_is_kept_in_the_scan_result() -> None:
118 """A file that cannot be processed stays in the scan result so it is not deleted."""
119 provider = _create_provider()
120 provider._sync_tracks = True
121 cur_filenames: set[str] = set()
122 item = MagicMock()
123 item.ext = "mp3"
124 item.relative_path = MISSING_FILE
125 item.absolute_path = f"/media/{MISSING_FILE}"
126
127 with patch(
128 "music_assistant.providers.filesystem_local.async_parse_tags",
129 AsyncMock(side_effect=OSError(errno.EIO, "i/o error")),
130 ):
131 result = await provider._process_item_async(item, None, cur_filenames)
132
133 assert result is False
134 # the file is still on disk, so the deletion step must not treat it as removed
135 assert cur_filenames == {MISSING_FILE}
136
137
138async def test_failed_cue_keeps_its_previous_tracks() -> None:
139 """A CUE sheet that cannot be parsed keeps the track ids of the previous scan."""
140 provider = _create_provider()
141 cue_path = "Artist/Album/album.cue"
142 cue_tracks = {make_cue_track_id(cue_path, 1), make_cue_track_id(cue_path, 2)}
143 cur_filenames: set[str] = set()
144 item = MagicMock()
145 item.ext = "cue"
146 item.relative_path = cue_path
147 item.absolute_path = f"/media/{cue_path}"
148 provider._cue = MagicMock()
149 provider._cue.parse_tracks = AsyncMock(side_effect=OSError(errno.EIO, "i/o error"))
150
151 result = await provider._process_item_async(
152 item, None, cur_filenames, prev_filenames={*cue_tracks, MISSING_FILE}
153 )
154
155 assert result is False
156 # without the track ids the deletion step would drop every track of the album
157 assert cur_filenames == {cue_path, *cue_tracks}
158
159
160async def test_sync_aborts_on_fatal_scan_error() -> None:
161 """A scan aborted by the circuit breaker skips deletions and flags the provider down."""
162 provider = _create_provider()
163 provider._enumerate_files_for_sync = _enumerate_result( # type: ignore[method-assign]
164 failed_dirs=20, fatal=True
165 )
166
167 await provider.sync_library(MediaType.TRACK)
168
169 provider._process_deletions.assert_not_called() # type: ignore[attr-defined]
170 provider._process_orphaned_albums_and_artists.assert_not_called() # type: ignore[attr-defined]
171 provider._set_available.assert_called_once_with(False) # type: ignore[attr-defined]
172