music-assistant-server

1"""Tests for cleanup of library items/images when a provider is removed."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, patch
6
7import pytest
8from music_assistant_models.enums import EventType, ImageType
9from music_assistant_models.errors import MediaNotFoundError
10from music_assistant_models.media_items import (
11    Artist,
12    MediaItemImage,
13    ProviderMapping,
14    UniqueList,
15)
16
17from music_assistant.mass import MusicAssistant
18
19FS_INSTANCE = "filesystem_local--AbCd"
20STREAM_INSTANCE = "spotify--EfGh"
21
22
23def _artist_with_two_providers(name: str) -> Artist:
24    """Build an artist that exists on both a filesystem and a streaming provider."""
25    artist = Artist(
26        item_id="fs1",
27        provider=FS_INSTANCE,
28        name=name,
29        provider_mappings={
30            ProviderMapping(
31                item_id="fs1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
32            ),
33            ProviderMapping(
34                item_id="sp1", provider_domain="spotify", provider_instance=STREAM_INSTANCE
35            ),
36        },
37    )
38    artist.metadata.images = UniqueList(
39        [
40            # local file path image: unresolvable once the filesystem provider is gone
41            MediaItemImage(
42                type=ImageType.THUMB,
43                path="Artist/folder.jpg",
44                provider=FS_INSTANCE,
45                remotely_accessible=False,
46            ),
47            # streaming image: stays valid, must be kept
48            MediaItemImage(
49                type=ImageType.THUMB,
50                path="https://i.scdn.co/image/abc",
51                provider=STREAM_INSTANCE,
52                remotely_accessible=True,
53            ),
54        ]
55    )
56    return artist
57
58
59async def test_remove_provider_mappings_strips_provider_images(mass: MusicAssistant) -> None:
60    """Removing a provider's mappings keeps the item but drops only that provider's images."""
61    artists = mass.music.artists
62    db_artist = await artists.add_item_to_library(_artist_with_two_providers("Two Provider Artist"))
63    db_id = int(db_artist.item_id)
64    assert db_artist.metadata.images is not None
65    assert len(db_artist.metadata.images) == 2
66
67    await artists.remove_provider_mappings(db_id, FS_INSTANCE)
68
69    # the item survives (it still has the streaming provider mapping)
70    updated = await artists.get_library_item(db_id)
71    assert {pm.provider_instance for pm in updated.provider_mappings} == {STREAM_INSTANCE}
72    # only the streaming image remains; the (now unresolvable) filesystem image is gone
73    assert updated.metadata.images is not None
74    assert [img.provider for img in updated.metadata.images] == [STREAM_INSTANCE]
75
76
77async def test_cleanup_filesystem_provider_does_not_reset_database(mass: MusicAssistant) -> None:
78    """Removing a local (filesystem) provider cleans up its items without wiping the db."""
79    artists = mass.music.artists
80    # an artist that only exists on the filesystem provider (should be removed entirely)
81    fs_only = Artist(
82        item_id="fsonly",
83        provider=FS_INSTANCE,
84        name="Filesystem Only Artist",
85        provider_mappings={
86            ProviderMapping(
87                item_id="fsonly",
88                provider_domain="filesystem_local",
89                provider_instance=FS_INSTANCE,
90            )
91        },
92    )
93    db_fs_only = await artists.add_item_to_library(fs_only)
94    # an artist from an unrelated provider that must survive the cleanup
95    survivor = Artist(
96        item_id="sp2",
97        provider=STREAM_INSTANCE,
98        name="Surviving Artist",
99        provider_mappings={
100            ProviderMapping(
101                item_id="sp2", provider_domain="spotify", provider_instance=STREAM_INSTANCE
102            )
103        },
104    )
105    db_survivor = await artists.add_item_to_library(survivor)
106
107    with patch.object(mass.music, "_reset_database", AsyncMock()) as mock_reset:
108        await mass.music.cleanup_provider(FS_INSTANCE)
109
110    # the whole database must NOT be wiped for a local provider anymore
111    mock_reset.assert_not_called()
112    # the filesystem-only item is gone, the unrelated item survives
113    with pytest.raises(MediaNotFoundError):
114        await artists.get_library_item(int(db_fs_only.item_id))
115    assert await artists.get_library_item(int(db_survivor.item_id))
116
117
118async def test_remove_single_provider_mapping_strips_provider_images(mass: MusicAssistant) -> None:
119    """Removing the last mapping of a provider instance also strips its images."""
120    artists = mass.music.artists
121    db_artist = await artists.add_item_to_library(
122        _artist_with_two_providers("Single Mapping Artist")
123    )
124    db_id = int(db_artist.item_id)
125
126    await artists.remove_provider_mapping(db_id, FS_INSTANCE, "fs1")
127
128    updated = await artists.get_library_item(db_id)
129    assert {pm.provider_instance for pm in updated.provider_mappings} == {STREAM_INSTANCE}
130    assert updated.metadata.images is not None
131    assert [img.provider for img in updated.metadata.images] == [STREAM_INSTANCE]
132
133
134async def test_remove_single_provider_mapping_keeps_images_if_instance_remains(
135    mass: MusicAssistant,
136) -> None:
137    """A provider's images are kept while another mapping of that same instance remains."""
138    artists = mass.music.artists
139    artist = Artist(
140        item_id="fs1",
141        provider=FS_INSTANCE,
142        name="Duplicate Mapping Artist",
143        provider_mappings={
144            ProviderMapping(
145                item_id="fs1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
146            ),
147            ProviderMapping(
148                item_id="fs2", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
149            ),
150        },
151    )
152    artist.metadata.images = UniqueList(
153        [
154            MediaItemImage(
155                type=ImageType.THUMB,
156                path="Artist/folder.jpg",
157                provider=FS_INSTANCE,
158                remotely_accessible=False,
159            )
160        ]
161    )
162    db_artist = await artists.add_item_to_library(artist)
163    db_id = int(db_artist.item_id)
164
165    # remove only one of the two mappings for the same provider instance
166    await artists.remove_provider_mapping(db_id, FS_INSTANCE, "fs1")
167
168    # the instance still maps the item (via fs2), so its image must be kept
169    updated = await artists.get_library_item(db_id)
170    assert {pm.item_id for pm in updated.provider_mappings} == {"fs2"}
171    assert updated.metadata.images is not None
172    assert [img.provider for img in updated.metadata.images] == [FS_INSTANCE]
173
174
175async def test_cleanup_suppresses_media_item_updated_events(mass: MusicAssistant) -> None:
176    """A surviving item must not emit MEDIA_ITEM_UPDATED during bulk provider cleanup."""
177    artists = mass.music.artists
178    db_artist = await artists.add_item_to_library(_artist_with_two_providers("Suppressed Artist"))
179    db_id = int(db_artist.item_id)
180
181    updated_object_ids: list[str | None] = []
182    real_signal_event = mass.signal_event
183
184    def _spy(event: EventType, object_id: str | None = None, data: object = None) -> None:
185        if event == EventType.MEDIA_ITEM_UPDATED:
186            updated_object_ids.append(object_id)
187        real_signal_event(event, object_id, data)
188
189    with patch.object(mass, "signal_event", side_effect=_spy):
190        await mass.music.cleanup_provider(FS_INSTANCE)
191
192    # no per-item update event was emitted during the bulk cleanup
193    assert updated_object_ids == []
194    # but the cleanup work still happened: the filesystem image was stripped, item kept
195    updated = await artists.get_library_item(db_id)
196    assert {pm.provider_instance for pm in updated.provider_mappings} == {STREAM_INSTANCE}
197    assert [img.provider for img in (updated.metadata.images or [])] == [STREAM_INSTANCE]
198
199
200async def test_remove_provider_mappings_emits_event_without_image_changes(
201    mass: MusicAssistant,
202) -> None:
203    """Removing mappings emits an update event even when no images were stripped."""
204    artists = mass.music.artists
205    # artist on two providers but without any images, so nothing gets stripped
206    artist = Artist(
207        item_id="fs1",
208        provider=FS_INSTANCE,
209        name="No Image Artist",
210        provider_mappings={
211            ProviderMapping(
212                item_id="fs1", provider_domain="filesystem_local", provider_instance=FS_INSTANCE
213            ),
214            ProviderMapping(
215                item_id="sp1", provider_domain="spotify", provider_instance=STREAM_INSTANCE
216            ),
217        },
218    )
219    db_artist = await artists.add_item_to_library(artist)
220    db_id = int(db_artist.item_id)
221
222    events: list[Artist] = []
223    real_signal_event = mass.signal_event
224
225    def _spy(event: EventType, object_id: str | None = None, data: object = None) -> None:
226        if event == EventType.MEDIA_ITEM_UPDATED and isinstance(data, Artist):
227            events.append(data)
228        real_signal_event(event, object_id, data)
229
230    with patch.object(mass, "signal_event", side_effect=_spy):
231        await artists.remove_provider_mappings(db_id, FS_INSTANCE)
232
233    # the update event fires purely because a provider mapping was removed, and its
234    # payload reflects the removed mapping
235    assert len(events) == 1
236    assert {pm.provider_instance for pm in events[0].provider_mappings} == {STREAM_INSTANCE}
237