/
/
/
1"""Tests for cleanup of library items/images when a provider is removed."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, patch
7
8import pytest
9from music_assistant_models.enums import EventType, ImageType
10from music_assistant_models.errors import MediaNotFoundError, MusicAssistantError
11from music_assistant_models.media_items import (
12 Album,
13 Artist,
14 Audiobook,
15 MediaItemImage,
16 Playlist,
17 Podcast,
18 ProviderMapping,
19 Track,
20 UniqueList,
21)
22
23from music_assistant.constants import (
24 DB_TABLE_ALBUM_ARTISTS,
25 DB_TABLE_ALBUM_TRACKS,
26 DB_TABLE_PROVIDER_MAPPINGS,
27 DB_TABLE_TRACK_ARTISTS,
28 DB_TABLE_TRACKS,
29)
30from music_assistant.controllers.music.media.base import MediaControllerBase
31from music_assistant.mass import MusicAssistant
32
33FS_INSTANCE = "filesystem_local--AbCd"
34STREAM_INSTANCE = "spotify--EfGh"
35
36
37def _artist_with_two_providers(name: str) -> Artist:
38 """Build an artist that exists on both a filesystem and a streaming provider."""
39 artist = Artist(
40 item_id="fs1",
41 provider=FS_INSTANCE,
42 name=name,
43 provider_mappings={
44 ProviderMapping(
45 item_id="fs1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
46 ),
47 ProviderMapping(
48 item_id="sp1", provider_domain="spotify", provider_instance=STREAM_INSTANCE
49 ),
50 },
51 )
52 artist.metadata.images = UniqueList(
53 [
54 # local file path image: unresolvable once the filesystem provider is gone
55 MediaItemImage(
56 type=ImageType.THUMB,
57 path="Artist/folder.jpg",
58 provider=FS_INSTANCE,
59 remotely_accessible=False,
60 ),
61 # streaming image: stays valid, must be kept
62 MediaItemImage(
63 type=ImageType.THUMB,
64 path="https://i.scdn.co/image/abc",
65 provider=STREAM_INSTANCE,
66 remotely_accessible=True,
67 ),
68 ]
69 )
70 return artist
71
72
73async def test_remove_provider_mappings_strips_provider_images(mass: MusicAssistant) -> None:
74 """Removing a provider's mappings keeps the item but drops only that provider's images."""
75 artists = mass.music.artists
76 db_artist = await artists.add_item_to_library(_artist_with_two_providers("Two Provider Artist"))
77 db_id = int(db_artist.item_id)
78 assert db_artist.metadata.images is not None
79 assert len(db_artist.metadata.images) == 2
80
81 await artists.remove_provider_mappings(db_id, FS_INSTANCE)
82
83 # the item survives (it still has the streaming provider mapping)
84 updated = await artists.get_library_item(db_id)
85 assert {pm.provider_instance for pm in updated.provider_mappings} == {STREAM_INSTANCE}
86 # only the streaming image remains; the (now unresolvable) filesystem image is gone
87 assert updated.metadata.images is not None
88 assert [img.provider for img in updated.metadata.images] == [STREAM_INSTANCE]
89
90
91async def test_cleanup_filesystem_provider_does_not_reset_database(mass: MusicAssistant) -> None:
92 """Removing a local (filesystem) provider cleans up its items without wiping the db."""
93 artists = mass.music.artists
94 # an artist that only exists on the filesystem provider (should be removed entirely)
95 fs_only = Artist(
96 item_id="fsonly",
97 provider=FS_INSTANCE,
98 name="Filesystem Only Artist",
99 provider_mappings={
100 ProviderMapping(
101 item_id="fsonly",
102 provider_domain="filesystem_local",
103 provider_instance=FS_INSTANCE,
104 )
105 },
106 )
107 db_fs_only = await artists.add_item_to_library(fs_only)
108 # an artist from an unrelated provider that must survive the cleanup
109 survivor = Artist(
110 item_id="sp2",
111 provider=STREAM_INSTANCE,
112 name="Surviving Artist",
113 provider_mappings={
114 ProviderMapping(
115 item_id="sp2", provider_domain="spotify", provider_instance=STREAM_INSTANCE
116 )
117 },
118 )
119 db_survivor = await artists.add_item_to_library(survivor)
120
121 with patch.object(mass.music, "_reset_database", AsyncMock()) as mock_reset:
122 await mass.music.cleanup_provider(FS_INSTANCE)
123
124 # the whole database must NOT be wiped for a local provider anymore
125 mock_reset.assert_not_called()
126 # the filesystem-only item is gone, the unrelated item survives
127 with pytest.raises(MediaNotFoundError):
128 await artists.get_library_item(int(db_fs_only.item_id))
129 assert await artists.get_library_item(int(db_survivor.item_id))
130
131
132async def test_remove_single_provider_mapping_strips_provider_images(mass: MusicAssistant) -> None:
133 """Removing the last mapping of a provider instance also strips its images."""
134 artists = mass.music.artists
135 db_artist = await artists.add_item_to_library(
136 _artist_with_two_providers("Single Mapping Artist")
137 )
138 db_id = int(db_artist.item_id)
139
140 await artists.remove_provider_mapping(db_id, FS_INSTANCE, "fs1")
141
142 updated = await artists.get_library_item(db_id)
143 assert {pm.provider_instance for pm in updated.provider_mappings} == {STREAM_INSTANCE}
144 assert updated.metadata.images is not None
145 assert [img.provider for img in updated.metadata.images] == [STREAM_INSTANCE]
146
147
148async def test_remove_single_provider_mapping_keeps_images_if_instance_remains(
149 mass: MusicAssistant,
150) -> None:
151 """A provider's images are kept while another mapping of that same instance remains."""
152 artists = mass.music.artists
153 artist = Artist(
154 item_id="fs1",
155 provider=FS_INSTANCE,
156 name="Duplicate Mapping Artist",
157 provider_mappings={
158 ProviderMapping(
159 item_id="fs1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
160 ),
161 ProviderMapping(
162 item_id="fs2", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
163 ),
164 },
165 )
166 artist.metadata.images = UniqueList(
167 [
168 MediaItemImage(
169 type=ImageType.THUMB,
170 path="Artist/folder.jpg",
171 provider=FS_INSTANCE,
172 remotely_accessible=False,
173 )
174 ]
175 )
176 db_artist = await artists.add_item_to_library(artist)
177 db_id = int(db_artist.item_id)
178
179 # remove only one of the two mappings for the same provider instance
180 await artists.remove_provider_mapping(db_id, FS_INSTANCE, "fs1")
181
182 # the instance still maps the item (via fs2), so its image must be kept
183 updated = await artists.get_library_item(db_id)
184 assert {pm.item_id for pm in updated.provider_mappings} == {"fs2"}
185 assert updated.metadata.images is not None
186 assert [img.provider for img in updated.metadata.images] == [FS_INSTANCE]
187
188
189async def test_cleanup_suppresses_media_item_updated_events(mass: MusicAssistant) -> None:
190 """A surviving item must not emit MEDIA_ITEM_UPDATED during bulk provider cleanup."""
191 artists = mass.music.artists
192 db_artist = await artists.add_item_to_library(_artist_with_two_providers("Suppressed Artist"))
193 db_id = int(db_artist.item_id)
194
195 updated_object_ids: list[str | None] = []
196 real_signal_event = mass.signal_event
197
198 def _spy(event: EventType, object_id: str | None = None, data: object = None) -> None:
199 if event == EventType.MEDIA_ITEM_UPDATED:
200 updated_object_ids.append(object_id)
201 real_signal_event(event, object_id, data)
202
203 with patch.object(mass, "signal_event", side_effect=_spy):
204 await mass.music.cleanup_provider(FS_INSTANCE)
205
206 # no per-item update event was emitted during the bulk cleanup
207 assert updated_object_ids == []
208 # but the cleanup work still happened: the filesystem image was stripped, item kept
209 updated = await artists.get_library_item(db_id)
210 assert {pm.provider_instance for pm in updated.provider_mappings} == {STREAM_INSTANCE}
211 assert [img.provider for img in (updated.metadata.images or [])] == [STREAM_INSTANCE]
212
213
214async def test_remove_provider_mappings_emits_event_without_image_changes(
215 mass: MusicAssistant,
216) -> None:
217 """Removing mappings emits an update event even when no images were stripped."""
218 artists = mass.music.artists
219 # artist on two providers but without any images, so nothing gets stripped
220 artist = Artist(
221 item_id="fs1",
222 provider=FS_INSTANCE,
223 name="No Image Artist",
224 provider_mappings={
225 ProviderMapping(
226 item_id="fs1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
227 ),
228 ProviderMapping(
229 item_id="sp1", provider_domain="spotify", provider_instance=STREAM_INSTANCE
230 ),
231 },
232 )
233 db_artist = await artists.add_item_to_library(artist)
234 db_id = int(db_artist.item_id)
235
236 events: list[Artist] = []
237 real_signal_event = mass.signal_event
238
239 def _spy(event: EventType, object_id: str | None = None, data: object = None) -> None:
240 if event == EventType.MEDIA_ITEM_UPDATED and isinstance(data, Artist):
241 events.append(data)
242 real_signal_event(event, object_id, data)
243
244 with patch.object(mass, "signal_event", side_effect=_spy):
245 await artists.remove_provider_mappings(db_id, FS_INSTANCE)
246
247 # the update event fires purely because a provider mapping was removed, and its
248 # payload reflects the removed mapping
249 assert len(events) == 1
250 assert {pm.provider_instance for pm in events[0].provider_mappings} == {STREAM_INSTANCE}
251
252
253async def test_failed_item_removal_keeps_provider_mapping(mass: MusicAssistant) -> None:
254 """A failing library removal must not leave the item behind without any providers."""
255 artists = mass.music.artists
256 fs_only = Artist(
257 item_id="fsonly",
258 provider=FS_INSTANCE,
259 name="Filesystem Only Artist",
260 provider_mappings={
261 ProviderMapping(
262 item_id="fsonly",
263 provider_domain="filesystem_local",
264 provider_instance=FS_INSTANCE,
265 )
266 },
267 )
268 db_artist = await artists.add_item_to_library(fs_only)
269 db_id = int(db_artist.item_id)
270
271 with (
272 patch.object(
273 artists, "remove_item_from_library", AsyncMock(side_effect=MusicAssistantError("boom"))
274 ),
275 pytest.raises(MusicAssistantError),
276 ):
277 await artists.remove_provider_mappings(db_id, FS_INSTANCE)
278
279 # the item survives *with* its mapping, so the removal can be retried
280 updated = await artists.get_library_item(db_id)
281 assert {pm.provider_instance for pm in updated.provider_mappings} == {FS_INSTANCE}
282
283
284async def test_failed_item_removal_keeps_single_provider_mapping(mass: MusicAssistant) -> None:
285 """The same applies when the item's last individual mapping is removed."""
286 artists = mass.music.artists
287 fs_only = Artist(
288 item_id="fsonly",
289 provider=FS_INSTANCE,
290 name="Filesystem Only Artist",
291 provider_mappings={
292 ProviderMapping(
293 item_id="fsonly",
294 provider_domain="filesystem_local",
295 provider_instance=FS_INSTANCE,
296 )
297 },
298 )
299 db_artist = await artists.add_item_to_library(fs_only)
300 db_id = int(db_artist.item_id)
301
302 with (
303 patch.object(
304 artists, "remove_item_from_library", AsyncMock(side_effect=MusicAssistantError("boom"))
305 ),
306 pytest.raises(MusicAssistantError),
307 ):
308 await artists.remove_provider_mapping(db_id, FS_INSTANCE, "fsonly")
309
310 updated = await artists.get_library_item(db_id)
311 assert {pm.provider_instance for pm in updated.provider_mappings} == {FS_INSTANCE}
312
313
314async def _assert_orphan_is_pruned(
315 mass: MusicAssistant, controller: MediaControllerBase[Any], db_id: int
316) -> None:
317 """Strip an item's provider mapping rows and assert the periodic cleanup removes it."""
318 await mass.music.database.delete(
319 DB_TABLE_PROVIDER_MAPPINGS,
320 {"media_type": controller.media_type.value, "item_id": db_id},
321 )
322
323 await mass.music._cleanup_database()
324
325 with pytest.raises(MediaNotFoundError):
326 await controller.get_library_item(db_id)
327
328
329async def test_database_cleanup_removes_orphaned_podcasts(mass: MusicAssistant) -> None:
330 """A podcast without any provider mapping is cleaned up."""
331 podcast = Podcast(
332 item_id="show1",
333 provider=FS_INSTANCE,
334 name="Orphaned Show",
335 provider_mappings={
336 ProviderMapping(
337 item_id="show1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
338 )
339 },
340 )
341 db_item = await mass.music.podcasts.add_item_to_library(podcast)
342 await _assert_orphan_is_pruned(mass, mass.music.podcasts, int(db_item.item_id))
343
344
345async def test_database_cleanup_removes_orphaned_audiobooks(mass: MusicAssistant) -> None:
346 """An audiobook without any provider mapping is cleaned up."""
347 audiobook = Audiobook(
348 item_id="book1",
349 provider=FS_INSTANCE,
350 name="Orphaned Book",
351 provider_mappings={
352 ProviderMapping(
353 item_id="book1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
354 )
355 },
356 )
357 db_item = await mass.music.audiobooks.add_item_to_library(audiobook)
358 await _assert_orphan_is_pruned(mass, mass.music.audiobooks, int(db_item.item_id))
359
360
361async def test_item_without_provider_mappings_raises_media_not_found(
362 mass: MusicAssistant,
363) -> None:
364 """An item that lost all its providers reports a clean 'not found' error."""
365 playlists = mass.music.playlists
366 playlist = Playlist(
367 item_id="pl1",
368 provider=FS_INSTANCE,
369 name="Orphaned Playlist",
370 owner="tester",
371 is_editable=False,
372 provider_mappings={
373 ProviderMapping(
374 item_id="pl1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
375 )
376 },
377 )
378 db_playlist = await playlists.add_item_to_library(playlist)
379 db_id = int(db_playlist.item_id)
380 await mass.music.database.delete(
381 DB_TABLE_PROVIDER_MAPPINGS,
382 {"media_type": playlists.media_type.value, "item_id": db_id},
383 )
384
385 with pytest.raises(MediaNotFoundError):
386 async for _ in playlists.tracks(str(db_id), "library"):
387 pass
388
389
390async def test_database_cleanup_removes_relations_of_deleted_items(mass: MusicAssistant) -> None:
391 """Relation rows left pointing at a deleted item are swept, sparing the ones still in use."""
392 artist = await mass.music.artists.add_item_to_library(
393 Artist(
394 item_id="rel-artist",
395 provider=FS_INSTANCE,
396 name="Relation Artist",
397 provider_mappings={
398 ProviderMapping(
399 item_id="rel-artist",
400 provider_domain="filesystem_local",
401 provider_instance=FS_INSTANCE,
402 )
403 },
404 )
405 )
406 album = await mass.music.albums.add_item_to_library(
407 Album(
408 item_id="rel-album",
409 provider=FS_INSTANCE,
410 name="Relation Album",
411 artists=UniqueList([artist]),
412 provider_mappings={
413 ProviderMapping(
414 item_id="rel-album",
415 provider_domain="filesystem_local",
416 provider_instance=FS_INSTANCE,
417 )
418 },
419 )
420 )
421 track = await mass.music.tracks.add_item_to_library(
422 Track(
423 item_id="rel-track",
424 provider=FS_INSTANCE,
425 name="Relation Track",
426 artists=UniqueList([artist]),
427 album=album,
428 provider_mappings={
429 ProviderMapping(
430 item_id="rel-track",
431 provider_domain="filesystem_local",
432 provider_instance=FS_INSTANCE,
433 )
434 },
435 )
436 )
437 track_id = int(track.item_id)
438 assert await mass.music.database.get_rows(DB_TABLE_ALBUM_TRACKS, {"track_id": track_id})
439 assert await mass.music.database.get_rows(DB_TABLE_TRACK_ARTISTS, {"track_id": track_id})
440
441 # drop the track row itself, the way a removal that only deletes its own side leaves it
442 await mass.music.database.delete(DB_TABLE_TRACKS, {"item_id": track_id})
443
444 await mass.music._cleanup_database()
445
446 assert not await mass.music.database.get_rows(DB_TABLE_ALBUM_TRACKS, {"track_id": track_id})
447 assert not await mass.music.database.get_rows(DB_TABLE_TRACK_ARTISTS, {"track_id": track_id})
448 # the album is still there with its artist, so that relation must survive
449 assert await mass.music.database.get_rows(
450 DB_TABLE_ALBUM_ARTISTS,
451 {"album_id": int(album.item_id), "artist_id": int(artist.item_id)},
452 )
453