music-assistant-server

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