/
/
/
1"""Tests for filesystem provider sync configuration behavior."""
2
3from unittest.mock import AsyncMock, MagicMock, patch
4
5import pytest
6from music_assistant_models.enums import MediaType, ProviderFeature
7from music_assistant_models.media_items import Podcast
8
9from music_assistant.providers.filesystem_local import LocalFileSystemProvider
10from music_assistant.providers.filesystem_local.constants import (
11 AUDIOBOOK_EXTENSIONS,
12 CONF_ENTRY_CONTENT_TYPE,
13 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS,
14 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS,
15 CONF_ENTRY_LIBRARY_SYNC_PODCASTS,
16 CONF_ENTRY_LIBRARY_SYNC_TRACKS,
17 PODCAST_EPISODE_EXTENSIONS,
18 TRACK_EXTENSIONS,
19)
20from music_assistant.providers.filesystem_local.helpers import FileSystemItem
21
22
23def _create_provider(
24 content_type: str = "music",
25 sync_tracks: bool = True,
26 sync_playlists: bool = True,
27 sync_audiobooks: bool = True,
28 sync_podcasts: bool = True,
29) -> LocalFileSystemProvider:
30 """
31 Create a LocalFileSystemProvider with mocked dependencies.
32
33 :param content_type: The media content type ("music", "audiobooks", "podcasts").
34 :param sync_tracks: Whether the tracks sync checkbox is enabled.
35 :param sync_playlists: Whether the playlists sync checkbox is enabled.
36 :param sync_audiobooks: Whether the audiobooks sync checkbox is enabled.
37 :param sync_podcasts: Whether the podcasts sync checkbox is enabled.
38 """
39 config_values = {
40 CONF_ENTRY_CONTENT_TYPE.key: content_type,
41 CONF_ENTRY_LIBRARY_SYNC_TRACKS.key: sync_tracks,
42 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS.key: sync_playlists,
43 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS.key: sync_audiobooks,
44 CONF_ENTRY_LIBRARY_SYNC_PODCASTS.key: sync_podcasts,
45 }
46
47 mock_config = MagicMock()
48 mock_config.get_value = MagicMock(side_effect=lambda key: config_values.get(key))
49
50 with patch.object(LocalFileSystemProvider, "__init__", lambda *_a, **_kw: None):
51 provider = LocalFileSystemProvider.__new__(LocalFileSystemProvider)
52
53 provider.config = mock_config
54 provider.media_content_type = content_type
55 provider.write_access = False
56 provider._sync_tracks = sync_tracks
57 provider._sync_playlists = sync_playlists
58 provider.sync_running = False
59 provider.mass = MagicMock()
60 provider.logger = MagicMock()
61
62 return provider
63
64
65class TestSupportedFeatures:
66 """Test that supported_features reflects content type, not sync preferences."""
67
68 def test_music_content_type(self) -> None:
69 """Music content type advertises all music library features."""
70 provider = _create_provider(content_type="music")
71 features = provider.supported_features
72 assert ProviderFeature.LIBRARY_TRACKS in features
73 assert ProviderFeature.LIBRARY_PLAYLISTS in features
74 assert ProviderFeature.LIBRARY_ALBUMS in features
75 assert ProviderFeature.LIBRARY_ARTISTS in features
76
77 def test_audiobooks_content_type(self) -> None:
78 """Audiobooks content type advertises audiobook library feature."""
79 provider = _create_provider(content_type="audiobooks")
80 assert ProviderFeature.LIBRARY_AUDIOBOOKS in provider.supported_features
81
82 def test_podcasts_content_type(self) -> None:
83 """Podcasts content type advertises podcast library feature."""
84 provider = _create_provider(content_type="podcasts")
85 assert ProviderFeature.LIBRARY_PODCASTS in provider.supported_features
86
87
88class TestSyncLibraryEarlyReturn:
89 """Test that sync_library returns early when all sync options are disabled."""
90
91 @pytest.mark.asyncio
92 async def test_music_returns_early_when_all_disabled(self) -> None:
93 """Music provider returns early when both tracks and playlists sync are disabled."""
94 provider = _create_provider(sync_tracks=False, sync_playlists=False)
95 await provider.sync_library(MediaType.TRACK)
96 provider.mass.music.database.get_rows_from_query.assert_not_called() # type: ignore[attr-defined]
97
98 @pytest.mark.asyncio
99 async def test_audiobooks_returns_early_when_disabled(self) -> None:
100 """Audiobooks provider returns early when audiobook sync is disabled."""
101 provider = _create_provider(content_type="audiobooks", sync_audiobooks=False)
102 await provider.sync_library(MediaType.AUDIOBOOK)
103 provider.mass.music.database.get_rows_from_query.assert_not_called() # type: ignore[attr-defined]
104
105 @pytest.mark.asyncio
106 async def test_podcasts_returns_early_when_disabled(self) -> None:
107 """Podcasts provider returns early when podcast sync is disabled."""
108 provider = _create_provider(content_type="podcasts", sync_podcasts=False)
109 await provider.sync_library(MediaType.PODCAST)
110 provider.mass.music.database.get_rows_from_query.assert_not_called() # type: ignore[attr-defined]
111
112
113class TestProcessItemRespectsConfig:
114 """Test that _process_item_async skips items when sync is disabled."""
115
116 @pytest.mark.asyncio
117 async def test_tracks_skipped_when_sync_disabled(self) -> None:
118 """Track files are not imported when track sync is disabled."""
119 provider = _create_provider(sync_tracks=False)
120 item = MagicMock()
121 item.ext = next(iter(TRACK_EXTENSIONS))
122 item.relative_path = "Artist/Album/track.mp3"
123
124 result = await provider._process_item_async(item, None)
125
126 assert result is False
127 provider.mass.music.tracks.add_item_to_library.assert_not_called() # type: ignore[attr-defined]
128
129 @pytest.mark.asyncio
130 async def test_tracks_imported_when_sync_enabled(self) -> None:
131 """Track files are imported when track sync is enabled."""
132 provider = _create_provider(sync_tracks=True)
133 item = MagicMock()
134 item.ext = next(iter(TRACK_EXTENSIONS))
135 item.absolute_path = "/media/Artist/Album/track.mp3"
136 item.relative_path = "Artist/Album/track.mp3"
137 item.file_size = 1000
138
139 mock_track = MagicMock()
140 provider._parse_track = AsyncMock(return_value=mock_track) # type: ignore[method-assign]
141 provider.mass.music.tracks.add_item_to_library = AsyncMock() # type: ignore[method-assign,misc]
142
143 with patch(
144 "music_assistant.providers.filesystem_local.async_parse_tags",
145 new_callable=AsyncMock,
146 ):
147 result = await provider._process_item_async(item, None)
148
149 assert result is True
150 provider.mass.music.tracks.add_item_to_library.assert_called_once()
151
152 @pytest.mark.asyncio
153 async def test_playlists_skipped_when_sync_disabled(self) -> None:
154 """Playlist files are not imported when playlist sync is disabled."""
155 provider = _create_provider(sync_playlists=False)
156 item = MagicMock()
157 item.ext = "m3u"
158 item.relative_path = "playlists/favorites.m3u"
159
160 result = await provider._process_item_async(item, None)
161
162 assert result is False
163 provider.mass.music.playlists.add_item_to_library.assert_not_called() # type: ignore[attr-defined]
164
165 @pytest.mark.asyncio
166 async def test_playlists_imported_when_sync_enabled(self) -> None:
167 """Playlist files are imported when playlist sync is enabled."""
168 provider = _create_provider(sync_playlists=True)
169 item = MagicMock()
170 item.ext = "m3u"
171 item.absolute_path = "/media/playlists/favorites.m3u"
172 item.relative_path = "playlists/favorites.m3u"
173
174 mock_playlist = MagicMock()
175 provider.get_playlist = AsyncMock(return_value=mock_playlist) # type: ignore[method-assign]
176 provider.mass.music.playlists.add_item_to_library = AsyncMock() # type: ignore[method-assign,misc]
177
178 result = await provider._process_item_async(item, None)
179
180 assert result is True
181 provider.mass.music.playlists.add_item_to_library.assert_called_once()
182
183 @pytest.mark.asyncio
184 async def test_audiobooks_imported_when_sync_enabled(self) -> None:
185 """Audiobook files are imported when audiobook sync is enabled."""
186 provider = _create_provider(content_type="audiobooks", sync_audiobooks=True)
187 item = MagicMock()
188 item.ext = next(iter(AUDIOBOOK_EXTENSIONS))
189 item.absolute_path = "/media/Author/Book/chapter01.m4b"
190 item.relative_path = "Author/Book/chapter01.m4b"
191 item.file_size = 5000
192
193 mock_audiobook = MagicMock()
194 provider._parse_audiobook = AsyncMock(return_value=mock_audiobook) # type: ignore[method-assign]
195 provider.mass.music.audiobooks.add_item_to_library = AsyncMock() # type: ignore[method-assign,misc]
196
197 with patch(
198 "music_assistant.providers.filesystem_local.async_parse_tags",
199 new_callable=AsyncMock,
200 ):
201 result = await provider._process_item_async(item, None)
202
203 assert result is True
204 provider.mass.music.audiobooks.add_item_to_library.assert_called_once()
205
206 @pytest.mark.asyncio
207 async def test_podcasts_imported_when_sync_enabled(self) -> None:
208 """Podcast files are imported when podcast sync is enabled."""
209 provider = _create_provider(content_type="podcasts", sync_podcasts=True)
210 item = MagicMock()
211 item.ext = next(iter(PODCAST_EPISODE_EXTENSIONS))
212 item.absolute_path = "/media/Podcast/episode01.mp3"
213 item.relative_path = "Podcast/episode01.mp3"
214 item.file_size = 3000
215
216 mock_episode = MagicMock()
217 mock_episode.podcast = MagicMock(spec=Podcast)
218 provider._parse_podcast_episode = AsyncMock(return_value=mock_episode) # type: ignore[method-assign]
219 provider.mass.music.podcasts.add_item_to_library = AsyncMock() # type: ignore[method-assign,misc]
220
221 with patch(
222 "music_assistant.providers.filesystem_local.async_parse_tags",
223 new_callable=AsyncMock,
224 ):
225 result = await provider._process_item_async(item, None)
226
227 assert result is True
228 provider.mass.music.podcasts.add_item_to_library.assert_called_once()
229
230
231def _classify(
232 provider: LocalFileSystemProvider,
233 relative_path: str,
234 file_checksums: dict[str, str] | None = None,
235) -> tuple[list[tuple[FileSystemItem, str | None]], set[str]]:
236 """Run a single file through _classify_scan_item and return its scan buckets."""
237 item = FileSystemItem(
238 filename=relative_path.rsplit("/", 1)[-1],
239 relative_path=relative_path,
240 absolute_path=f"/media/{relative_path}",
241 is_dir=False,
242 checksum="1",
243 )
244 items_to_process: list[tuple[FileSystemItem, str | None]] = []
245 cur_filenames: set[str] = set()
246 provider._classify_scan_item(
247 item,
248 file_checksums=file_checksums or {},
249 cue_file_checksums={},
250 cur_filenames=cur_filenames,
251 items_to_process=items_to_process,
252 unchanged_cue_items=[],
253 cue_stems=set(),
254 ignore_album_playlists=False,
255 metadata_files=[],
256 )
257 return items_to_process, cur_filenames
258
259
260class TestClassifyScanItemSkipsUnimportedFiles:
261 """
262 The scan must only queue files this provider actually imports.
263
264 The walker looks for the union of every content type's extensions, so a file
265 that this content type never imports gets no provider mapping and would be
266 queued again on every single sync.
267 """
268
269 def test_music_skips_audiobook_only_extension(self) -> None:
270 """An .m4b in a music library is never imported, so it must not be queued."""
271 provider = _create_provider(content_type="music")
272 items, _ = _classify(provider, "Boeken/luisterboek.m4b")
273 assert items == []
274
275 def test_skipped_file_still_counts_as_present(self) -> None:
276 """A skipped file is still on disk, so the deletion pass must not claim it."""
277 provider = _create_provider(content_type="music")
278 _, cur_filenames = _classify(provider, "Boeken/luisterboek.m4b")
279 assert cur_filenames == {"Boeken/luisterboek.m4b"}
280
281 def test_indexed_track_survives_disabling_track_sync(self) -> None:
282 """
283 Turning off track sync must not delete the tracks already in the library.
284
285 The deletion pass removes everything the scan did not report as present.
286 """
287 provider = _create_provider(content_type="music", sync_tracks=False)
288 path = "Artist/Album/track.mp3"
289 items, cur_filenames = _classify(provider, path, file_checksums={path: "1"})
290 assert items == []
291 assert cur_filenames == {path}
292
293 def test_indexed_playlist_survives_disabling_playlist_sync(self) -> None:
294 """Turning off playlist sync must not delete the playlists already in the library."""
295 provider = _create_provider(content_type="music", sync_playlists=False)
296 path = "playlists/favorites.m3u"
297 items, cur_filenames = _classify(provider, path, file_checksums={path: "1"})
298 assert items == []
299 assert cur_filenames == {path}
300
301 def test_music_queues_track(self) -> None:
302 """A regular track file is still queued."""
303 provider = _create_provider(content_type="music")
304 items, _ = _classify(provider, "Artist/Album/track.mp3")
305 assert [item.relative_path for item, _ in items] == ["Artist/Album/track.mp3"]
306
307 def test_music_queues_cue(self) -> None:
308 """A CUE sheet is still queued."""
309 provider = _create_provider(content_type="music")
310 items, _ = _classify(provider, "Artist/Album/album.cue")
311 assert [item.relative_path for item, _ in items] == ["Artist/Album/album.cue"]
312
313 def test_music_skips_tracks_when_sync_disabled(self) -> None:
314 """Track files are not queued when track sync is off."""
315 provider = _create_provider(content_type="music", sync_tracks=False)
316 items, _ = _classify(provider, "Artist/Album/track.mp3")
317 assert items == []
318
319 def test_music_skips_playlists_when_sync_disabled(self) -> None:
320 """Playlist files are not queued when playlist sync is off."""
321 provider = _create_provider(content_type="music", sync_playlists=False)
322 items, _ = _classify(provider, "playlists/favorites.m3u")
323 assert items == []
324
325 def test_audiobooks_skips_track_only_extension(self) -> None:
326 """A .wav in an audiobooks library is never imported, so it must not be queued."""
327 provider = _create_provider(content_type="audiobooks")
328 items, _ = _classify(provider, "Author/Book/chapter.wav")
329 assert items == []
330
331 def test_audiobooks_skips_playlist(self) -> None:
332 """A playlist in an audiobooks library is never imported, so it must not be queued."""
333 provider = _create_provider(content_type="audiobooks")
334 items, _ = _classify(provider, "Author/Book/book.m3u")
335 assert items == []
336
337 def test_podcasts_skips_cue(self) -> None:
338 """A CUE sheet in a podcasts library is never imported, so it must not be queued."""
339 provider = _create_provider(content_type="podcasts")
340 items, _ = _classify(provider, "Podcast/episode.cue")
341 assert items == []
342
343 def test_podcasts_queues_episode(self) -> None:
344 """A regular episode file is still queued."""
345 provider = _create_provider(content_type="podcasts")
346 items, _ = _classify(provider, "Podcast/episode01.mp3")
347 assert [item.relative_path for item, _ in items] == ["Podcast/episode01.mp3"]
348