music-assistant-server

1"""
2Tests for crediting artists and albums from track plays.
3
4The integration tests use the ``mass`` fixture from ``tests/conftest.py`` which
5creates a full MusicAssistant instance with a real SQLite database in a
6temporary directory. The decision helper is covered by a plain unit test.
7"""
8
9from __future__ import annotations
10
11from typing import TYPE_CHECKING, cast
12from unittest.mock import Mock
13from uuid import uuid4
14
15from music_assistant_models.enums import AlbumType, MediaType
16from music_assistant_models.media_items import Album, Artist, ItemMapping, ProviderMapping, Track
17from music_assistant_models.queue_item import QueueItem
18from music_assistant_models.unique_list import UniqueList
19
20from music_assistant.constants import DB_TABLE_ALBUMS, DB_TABLE_ARTISTS, DB_TABLE_PLAYLOG
21from music_assistant.controllers.player_queues import PlayerQueuesController
22from music_assistant.controllers.player_queues.state import PlayerQueueData
23from music_assistant.mass import MusicAssistant
24
25if TYPE_CHECKING:
26    from music_assistant_models.player_queue import PlayerQueue
27
28
29def _library_mapping() -> set[ProviderMapping]:
30    """Create a single library provider mapping with a unique provider item id."""
31    return {
32        ProviderMapping(
33            item_id=uuid4().hex,
34            provider_domain="library",
35            provider_instance="library",
36            in_library=True,
37        )
38    }
39
40
41async def _add_artist(mass: MusicAssistant, name: str) -> Artist:
42    """Add a minimal artist to the library and return the stored item."""
43    return await mass.music.artists.add_item_to_library(
44        Artist(item_id="0", provider="library", name=name, provider_mappings=_library_mapping())
45    )
46
47
48async def _add_track(mass: MusicAssistant, name: str, artists: list[Artist]) -> Track:
49    """Add a minimal track with the given artists and return the (re-fetched) stored item."""
50    added = await mass.music.tracks.add_item_to_library(
51        Track(
52            item_id="0",
53            provider="library",
54            name=name,
55            provider_mappings=_library_mapping(),
56            artists=UniqueList(artists),
57        )
58    )
59    return await mass.music.tracks.get_library_item(added.item_id)
60
61
62async def _add_album(mass: MusicAssistant, name: str, artists: list[Artist]) -> Album:
63    """Add a minimal album with the given album artists and return the (re-fetched) stored item."""
64    added = await mass.music.albums.add_item_to_library(
65        Album(
66            item_id="0",
67            provider="library",
68            name=name,
69            provider_mappings=_library_mapping(),
70            album_type=AlbumType.ALBUM,
71            artists=UniqueList(artists),
72        )
73    )
74    return await mass.music.albums.get_library_item(added.item_id)
75
76
77async def _play_count(mass: MusicAssistant, table: str, item_id: str) -> int:
78    """Read the play_count for a library item directly from the database."""
79    row = await mass.music.database.get_row(table, {"item_id": item_id})
80    assert row is not None
81    return int(row["play_count"])
82
83
84async def _artist_user_initiated(mass: MusicAssistant, item_id: str, userid: str) -> int:
85    """Read the user_initiated flag of an artist playlog row."""
86    row = await mass.music.database.get_row(
87        DB_TABLE_PLAYLOG,
88        {"media_type": MediaType.ARTIST.value, "item_id": item_id, "userid": userid},
89    )
90    assert row is not None
91    return int(row["user_initiated"])
92
93
94async def test_played_track_credits_all_track_artists(mass: MusicAssistant) -> None:
95    """A fully played track credits every one of its artists (not just the primary)."""
96    user = await mass.webserver.auth.create_user("trackcredit")
97    primary = await _add_artist(mass, "Primary")
98    featured = await _add_artist(mass, "Featured")
99    track = await _add_track(mass, "Collab", [primary, featured])
100
101    await mass.music.mark_item_played(
102        track, fully_played=True, user_initiated=True, userid=user.user_id
103    )
104
105    assert await _play_count(mass, DB_TABLE_ARTISTS, primary.item_id) == 1
106    assert await _play_count(mass, DB_TABLE_ARTISTS, featured.item_id) == 1
107    for artist in (primary, featured):
108        rows = await mass.music.database.get_rows(
109            DB_TABLE_PLAYLOG,
110            {
111                "media_type": MediaType.ARTIST.value,
112                "item_id": artist.item_id,
113                "userid": user.user_id,
114            },
115        )
116        assert len(rows) == 1
117
118
119async def test_album_play_credits_album_artists_with_dedup(mass: MusicAssistant) -> None:
120    """Marking an album played credits its album artists, skipping any in skip_artist_ids."""
121    kept = await _add_artist(mass, "Kept Artist")
122    skipped = await _add_artist(mass, "Skipped Artist")
123    album = await _add_album(mass, "An Album", [kept, skipped])
124
125    await mass.music.mark_item_played(
126        album, fully_played=True, user_initiated=True, skip_artist_ids=[skipped.item_id]
127    )
128
129    assert await _play_count(mass, DB_TABLE_ALBUMS, album.item_id) == 1
130    assert await _play_count(mass, DB_TABLE_ARTISTS, kept.item_id) == 1
131    assert await _play_count(mass, DB_TABLE_ARTISTS, skipped.item_id) == 0
132
133
134async def test_track_credit_does_not_flag_artist_user_initiated(mass: MusicAssistant) -> None:
135    """A track play credits its artist as a side-effect, never as user-initiated."""
136    user = await mass.webserver.auth.create_user("trackcreditui")
137    artist = await _add_artist(mass, "Sideeffect")
138    track = await _add_track(mass, "A Song", [artist])
139
140    await mass.music.mark_item_played(
141        track, fully_played=True, user_initiated=True, userid=user.user_id
142    )
143
144    assert await _artist_user_initiated(mass, artist.item_id, user.user_id) == 0
145
146
147async def test_explicit_artist_play_survives_track_credit(mass: MusicAssistant) -> None:
148    """An explicit artist play stays user-initiated after later side-effect credits."""
149    user = await mass.webserver.auth.create_user("artiststicky")
150    artist = await _add_artist(mass, "Chosen")
151    track = await _add_track(mass, "Their Song", [artist])
152
153    # user explicitly plays the artist
154    await mass.music.mark_item_played(
155        artist, fully_played=True, user_initiated=True, userid=user.user_id
156    )
157    assert await _artist_user_initiated(mass, artist.item_id, user.user_id) == 1
158
159    # later, one of the artist's tracks auto-plays (not user-initiated)
160    await mass.music.mark_item_played(
161        track, fully_played=True, user_initiated=False, userid=user.user_id
162    )
163    assert await _artist_user_initiated(mass, artist.item_id, user.user_id) == 1
164
165
166def test_enqueued_album_decision() -> None:
167    """An album is credited only when enqueued and only on the first track of its run."""
168    album_x = Album(
169        item_id="ax",
170        provider="library",
171        name="X",
172        provider_mappings=set(),
173        album_type=AlbumType.ALBUM,
174    )
175    album_y = Album(
176        item_id="ay",
177        provider="library",
178        name="Y",
179        provider_mappings=set(),
180        album_type=AlbumType.ALBUM,
181    )
182    t1 = Track(item_id="t1", provider="library", name="T1", provider_mappings=set(), album=album_x)
183    t2 = Track(item_id="t2", provider="library", name="T2", provider_mappings=set(), album=album_x)
184    t3 = Track(item_id="t3", provider="library", name="T3", provider_mappings=set(), album=album_y)
185    items = [QueueItem.from_media_item("q1", track) for track in (t1, t2, t3)]
186
187    tracker = PlayerQueuesController.__new__(PlayerQueuesController)
188    queue = cast("PlayerQueue", Mock(queue_id="q1"))
189    data = PlayerQueueData(queue=queue, items=items, enqueued_media_items=[album_x])
190    tracker._queue_data = {"q1": data}
191
192    # first track of the enqueued album -> credit it
193    assert tracker._enqueued_album_for_track(data, items[0], t1) is album_x
194    # second track shares the previous queue item's album -> already handled
195    assert tracker._enqueued_album_for_track(data, items[1], t2) is None
196    # a track whose album was never enqueued -> not credited
197    assert tracker._enqueued_album_for_track(data, items[2], t3) is None
198
199
200def test_enqueued_provider_album_credits_its_library_tracks() -> None:
201    """An album picked from a provider listing is credited, though its tracks are library ones."""
202    mapping = ProviderMapping(
203        item_id="album-prov-1", provider_domain="spotify", provider_instance="spotify--abc"
204    )
205    # the album object a provider listing hands to play_media
206    provider_album = Album(
207        item_id="album-prov-1",
208        provider="spotify--abc",
209        name="Kind of Blue",
210        provider_mappings={mapping},
211        album_type=AlbumType.ALBUM,
212    )
213    # the album its tracks carry, because the album is also in the library
214    library_album = Album(
215        item_id="7",
216        provider="library",
217        name="Kind of Blue",
218        provider_mappings={mapping},
219        album_type=AlbumType.ALBUM,
220    )
221    other_album = Album(
222        item_id="8",
223        provider="library",
224        name="Sketches of Spain",
225        provider_mappings={
226            ProviderMapping(
227                item_id="album-prov-2",
228                provider_domain="spotify",
229                provider_instance="spotify--abc",
230            )
231        },
232        album_type=AlbumType.ALBUM,
233    )
234    t1 = Track(item_id="t1", provider="library", name="T1", provider_mappings=set())
235    t2 = Track(item_id="t2", provider="library", name="T2", provider_mappings=set())
236    t3 = Track(item_id="t3", provider="library", name="T3", provider_mappings=set())
237    items = [QueueItem.from_media_item("q1", track) for track in (t1, t2, t3)]
238    # loading an item for playback puts the full library album on it
239    t1.album = library_album
240    t2.album = library_album
241    t3.album = other_album
242
243    tracker = PlayerQueuesController.__new__(PlayerQueuesController)
244    queue = cast("PlayerQueue", Mock(queue_id="q1"))
245    data = PlayerQueueData(queue=queue, items=items, enqueued_media_items=[provider_album])
246    tracker._queue_data = {"q1": data}
247
248    # the enqueued album is credited on the first track of its run, under the library
249    # identity so it shares the row an explicit library play writes
250    assert tracker._enqueued_album_for_track(data, items[0], t1) is library_album
251    # and only once for that run
252    assert tracker._enqueued_album_for_track(data, items[1], t2) is None
253    # a track from an album the user never enqueued is not credited
254    assert tracker._enqueued_album_for_track(data, items[2], t3) is None
255
256
257def test_album_outside_the_library_is_credited_as_the_provider_album() -> None:
258    """An album that is not in the library is credited under the provider identity."""
259    provider_album = Album(
260        item_id="album-prov-1",
261        provider="spotify--abc",
262        name="Kind of Blue",
263        provider_mappings={
264            ProviderMapping(
265                item_id="album-prov-1",
266                provider_domain="spotify",
267                provider_instance="spotify--abc",
268            )
269        },
270        album_type=AlbumType.ALBUM,
271    )
272    track = Track(item_id="t1", provider="spotify--abc", name="T1", provider_mappings=set())
273    items = [QueueItem.from_media_item("q1", track)]
274    # with no library album to swap in, the track keeps the provider album as a mapping
275    track.album = ItemMapping.from_item(provider_album)
276
277    tracker = PlayerQueuesController.__new__(PlayerQueuesController)
278    queue = cast("PlayerQueue", Mock(queue_id="q1"))
279    data = PlayerQueueData(queue=queue, items=items, enqueued_media_items=[provider_album])
280    tracker._queue_data = {"q1": data}
281
282    assert tracker._enqueued_album_for_track(data, items[0], track) is provider_album
283