/
/
/
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)
20
21
22def _create_provider(
23 content_type: str = "music",
24 sync_tracks: bool = True,
25 sync_playlists: bool = True,
26 sync_audiobooks: bool = True,
27 sync_podcasts: bool = True,
28) -> LocalFileSystemProvider:
29 """
30 Create a LocalFileSystemProvider with mocked dependencies.
31
32 :param content_type: The media content type ("music", "audiobooks", "podcasts").
33 :param sync_tracks: Whether the tracks sync checkbox is enabled.
34 :param sync_playlists: Whether the playlists sync checkbox is enabled.
35 :param sync_audiobooks: Whether the audiobooks sync checkbox is enabled.
36 :param sync_podcasts: Whether the podcasts sync checkbox is enabled.
37 """
38 config_values = {
39 CONF_ENTRY_CONTENT_TYPE.key: content_type,
40 CONF_ENTRY_LIBRARY_SYNC_TRACKS.key: sync_tracks,
41 CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS.key: sync_playlists,
42 CONF_ENTRY_LIBRARY_SYNC_AUDIOBOOKS.key: sync_audiobooks,
43 CONF_ENTRY_LIBRARY_SYNC_PODCASTS.key: sync_podcasts,
44 }
45
46 mock_config = MagicMock()
47 mock_config.get_value = MagicMock(side_effect=lambda key: config_values.get(key))
48
49 with patch.object(LocalFileSystemProvider, "__init__", lambda *_a, **_kw: None):
50 provider = LocalFileSystemProvider.__new__(LocalFileSystemProvider)
51
52 provider.config = mock_config
53 provider.media_content_type = content_type
54 provider.write_access = False
55 provider._sync_tracks = sync_tracks
56 provider._sync_playlists = sync_playlists
57 provider.sync_running = False
58 provider.mass = MagicMock()
59 provider.logger = MagicMock()
60
61 return provider
62
63
64class TestSupportedFeatures:
65 """Test that supported_features reflects content type, not sync preferences."""
66
67 def test_music_content_type(self) -> None:
68 """Music content type advertises all music library features."""
69 provider = _create_provider(content_type="music")
70 features = provider.supported_features
71 assert ProviderFeature.LIBRARY_TRACKS in features
72 assert ProviderFeature.LIBRARY_PLAYLISTS in features
73 assert ProviderFeature.LIBRARY_ALBUMS in features
74 assert ProviderFeature.LIBRARY_ARTISTS in features
75
76 def test_audiobooks_content_type(self) -> None:
77 """Audiobooks content type advertises audiobook library feature."""
78 provider = _create_provider(content_type="audiobooks")
79 assert ProviderFeature.LIBRARY_AUDIOBOOKS in provider.supported_features
80
81 def test_podcasts_content_type(self) -> None:
82 """Podcasts content type advertises podcast library feature."""
83 provider = _create_provider(content_type="podcasts")
84 assert ProviderFeature.LIBRARY_PODCASTS in provider.supported_features
85
86
87class TestSyncLibraryEarlyReturn:
88 """Test that sync_library returns early when all sync options are disabled."""
89
90 @pytest.mark.asyncio
91 async def test_music_returns_early_when_all_disabled(self) -> None:
92 """Music provider returns early when both tracks and playlists sync are disabled."""
93 provider = _create_provider(sync_tracks=False, sync_playlists=False)
94 await provider.sync_library(MediaType.TRACK)
95 provider.mass.music.database.get_rows_from_query.assert_not_called() # type: ignore[attr-defined]
96
97 @pytest.mark.asyncio
98 async def test_audiobooks_returns_early_when_disabled(self) -> None:
99 """Audiobooks provider returns early when audiobook sync is disabled."""
100 provider = _create_provider(content_type="audiobooks", sync_audiobooks=False)
101 await provider.sync_library(MediaType.AUDIOBOOK)
102 provider.mass.music.database.get_rows_from_query.assert_not_called() # type: ignore[attr-defined]
103
104 @pytest.mark.asyncio
105 async def test_podcasts_returns_early_when_disabled(self) -> None:
106 """Podcasts provider returns early when podcast sync is disabled."""
107 provider = _create_provider(content_type="podcasts", sync_podcasts=False)
108 await provider.sync_library(MediaType.PODCAST)
109 provider.mass.music.database.get_rows_from_query.assert_not_called() # type: ignore[attr-defined]
110
111
112class TestProcessItemRespectsConfig:
113 """Test that _process_item_async skips items when sync is disabled."""
114
115 @pytest.mark.asyncio
116 async def test_tracks_skipped_when_sync_disabled(self) -> None:
117 """Track files are not imported when track sync is disabled."""
118 provider = _create_provider(sync_tracks=False)
119 item = MagicMock()
120 item.ext = next(iter(TRACK_EXTENSIONS))
121 item.relative_path = "Artist/Album/track.mp3"
122
123 result = await provider._process_item_async(item, None)
124
125 assert result is False
126 provider.mass.music.tracks.add_item_to_library.assert_not_called() # type: ignore[attr-defined]
127
128 @pytest.mark.asyncio
129 async def test_tracks_imported_when_sync_enabled(self) -> None:
130 """Track files are imported when track sync is enabled."""
131 provider = _create_provider(sync_tracks=True)
132 item = MagicMock()
133 item.ext = next(iter(TRACK_EXTENSIONS))
134 item.absolute_path = "/media/Artist/Album/track.mp3"
135 item.relative_path = "Artist/Album/track.mp3"
136 item.file_size = 1000
137
138 mock_track = MagicMock()
139 provider._parse_track = AsyncMock(return_value=mock_track) # type: ignore[method-assign]
140 provider.mass.music.tracks.add_item_to_library = AsyncMock() # type: ignore[method-assign,misc]
141
142 with patch(
143 "music_assistant.providers.filesystem_local.async_parse_tags",
144 new_callable=AsyncMock,
145 ):
146 result = await provider._process_item_async(item, None)
147
148 assert result is True
149 provider.mass.music.tracks.add_item_to_library.assert_called_once()
150
151 @pytest.mark.asyncio
152 async def test_playlists_skipped_when_sync_disabled(self) -> None:
153 """Playlist files are not imported when playlist sync is disabled."""
154 provider = _create_provider(sync_playlists=False)
155 item = MagicMock()
156 item.ext = "m3u"
157 item.relative_path = "playlists/favorites.m3u"
158
159 result = await provider._process_item_async(item, None)
160
161 assert result is False
162 provider.mass.music.playlists.add_item_to_library.assert_not_called() # type: ignore[attr-defined]
163
164 @pytest.mark.asyncio
165 async def test_playlists_imported_when_sync_enabled(self) -> None:
166 """Playlist files are imported when playlist sync is enabled."""
167 provider = _create_provider(sync_playlists=True)
168 item = MagicMock()
169 item.ext = "m3u"
170 item.absolute_path = "/media/playlists/favorites.m3u"
171 item.relative_path = "playlists/favorites.m3u"
172
173 mock_playlist = MagicMock()
174 provider.get_playlist = AsyncMock(return_value=mock_playlist) # type: ignore[method-assign]
175 provider.mass.music.playlists.add_item_to_library = AsyncMock() # type: ignore[method-assign,misc]
176
177 result = await provider._process_item_async(item, None)
178
179 assert result is True
180 provider.mass.music.playlists.add_item_to_library.assert_called_once()
181
182 @pytest.mark.asyncio
183 async def test_audiobooks_imported_when_sync_enabled(self) -> None:
184 """Audiobook files are imported when audiobook sync is enabled."""
185 provider = _create_provider(content_type="audiobooks", sync_audiobooks=True)
186 item = MagicMock()
187 item.ext = next(iter(AUDIOBOOK_EXTENSIONS))
188 item.absolute_path = "/media/Author/Book/chapter01.m4b"
189 item.relative_path = "Author/Book/chapter01.m4b"
190 item.file_size = 5000
191
192 mock_audiobook = MagicMock()
193 provider._parse_audiobook = AsyncMock(return_value=mock_audiobook) # type: ignore[method-assign]
194 provider.mass.music.audiobooks.add_item_to_library = AsyncMock() # type: ignore[method-assign,misc]
195
196 with patch(
197 "music_assistant.providers.filesystem_local.async_parse_tags",
198 new_callable=AsyncMock,
199 ):
200 result = await provider._process_item_async(item, None)
201
202 assert result is True
203 provider.mass.music.audiobooks.add_item_to_library.assert_called_once()
204
205 @pytest.mark.asyncio
206 async def test_podcasts_imported_when_sync_enabled(self) -> None:
207 """Podcast files are imported when podcast sync is enabled."""
208 provider = _create_provider(content_type="podcasts", sync_podcasts=True)
209 item = MagicMock()
210 item.ext = next(iter(PODCAST_EPISODE_EXTENSIONS))
211 item.absolute_path = "/media/Podcast/episode01.mp3"
212 item.relative_path = "Podcast/episode01.mp3"
213 item.file_size = 3000
214
215 mock_episode = MagicMock()
216 mock_episode.podcast = MagicMock(spec=Podcast)
217 provider._parse_podcast_episode = AsyncMock(return_value=mock_episode) # type: ignore[method-assign]
218 provider.mass.music.podcasts.add_item_to_library = AsyncMock() # type: ignore[method-assign,misc]
219
220 with patch(
221 "music_assistant.providers.filesystem_local.async_parse_tags",
222 new_callable=AsyncMock,
223 ):
224 result = await provider._process_item_async(item, None)
225
226 assert result is True
227 provider.mass.music.podcasts.add_item_to_library.assert_called_once()
228