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.unique_list import UniqueList
18
19from music_assistant.constants import DB_TABLE_ALBUMS, DB_TABLE_ARTISTS, DB_TABLE_PLAYLOG
20from music_assistant.controllers.player_queues import PlayerQueuesController
21from music_assistant.controllers.player_queues.state import PlayerQueueData
22from music_assistant.mass import MusicAssistant
23
24if TYPE_CHECKING:
25    from music_assistant_models.player_queue import PlayerQueue
26
27
28def _library_mapping() -> set[ProviderMapping]:
29    """Create a single library provider mapping with a unique provider item id."""
30    return {
31        ProviderMapping(
32            item_id=uuid4().hex,
33            provider_domain="library",
34            provider_instance="library",
35            in_library=True,
36        )
37    }
38
39
40async def _add_artist(mass: MusicAssistant, name: str) -> Artist:
41    """Add a minimal artist to the library and return the stored item."""
42    return await mass.music.artists.add_item_to_library(
43        Artist(item_id="0", provider="library", name=name, provider_mappings=_library_mapping())
44    )
45
46
47async def _add_track(mass: MusicAssistant, name: str, artists: list[Artist]) -> Track:
48    """Add a minimal track with the given artists and return the (re-fetched) stored item."""
49    added = await mass.music.tracks.add_item_to_library(
50        Track(
51            item_id="0",
52            provider="library",
53            name=name,
54            provider_mappings=_library_mapping(),
55            artists=UniqueList(artists),
56        )
57    )
58    return await mass.music.tracks.get_library_item(added.item_id)
59
60
61async def _add_album(mass: MusicAssistant, name: str, artists: list[Artist]) -> Album:
62    """Add a minimal album with the given album artists and return the (re-fetched) stored item."""
63    added = await mass.music.albums.add_item_to_library(
64        Album(
65            item_id="0",
66            provider="library",
67            name=name,
68            provider_mappings=_library_mapping(),
69            album_type=AlbumType.ALBUM,
70            artists=UniqueList(artists),
71        )
72    )
73    return await mass.music.albums.get_library_item(added.item_id)
74
75
76def _queue_data(enqueued: list[Album]) -> PlayerQueueData:
77    """Build the queue record the album-credit decision reads (only the enqueued items matter)."""
78    return PlayerQueueData(
79        queue=cast("PlayerQueue", Mock(queue_id="q1")),
80        enqueued_media_items=list(enqueued),
81    )
82
83
84async def _play_count(mass: MusicAssistant, table: str, item_id: str) -> int:
85    """Read the play_count for a library item directly from the database."""
86    row = await mass.music.database.get_row(table, {"item_id": item_id})
87    assert row is not None
88    return int(row["play_count"])
89
90
91async def _artist_user_initiated(mass: MusicAssistant, item_id: str, userid: str) -> int:
92    """Read the user_initiated flag of an artist playlog row."""
93    row = await mass.music.database.get_row(
94        DB_TABLE_PLAYLOG,
95        {"media_type": MediaType.ARTIST.value, "item_id": item_id, "userid": userid},
96    )
97    assert row is not None
98    return int(row["user_initiated"])
99
100
101async def test_played_track_credits_all_track_artists(mass: MusicAssistant) -> None:
102    """A fully played track credits every one of its artists (not just the primary)."""
103    user = await mass.webserver.auth.create_user("trackcredit")
104    primary = await _add_artist(mass, "Primary")
105    featured = await _add_artist(mass, "Featured")
106    track = await _add_track(mass, "Collab", [primary, featured])
107
108    await mass.music.mark_item_played(
109        track, fully_played=True, user_initiated=True, userid=user.user_id
110    )
111
112    assert await _play_count(mass, DB_TABLE_ARTISTS, primary.item_id) == 1
113    assert await _play_count(mass, DB_TABLE_ARTISTS, featured.item_id) == 1
114    for artist in (primary, featured):
115        rows = await mass.music.database.get_rows(
116            DB_TABLE_PLAYLOG,
117            {
118                "media_type": MediaType.ARTIST.value,
119                "item_id": artist.item_id,
120                "userid": user.user_id,
121            },
122        )
123        assert len(rows) == 1
124
125
126async def test_album_play_credits_album_artists_with_dedup(mass: MusicAssistant) -> None:
127    """Marking an album played credits its album artists, skipping any in skip_artist_ids."""
128    kept = await _add_artist(mass, "Kept Artist")
129    skipped = await _add_artist(mass, "Skipped Artist")
130    album = await _add_album(mass, "An Album", [kept, skipped])
131
132    await mass.music.mark_item_played(
133        album, fully_played=True, user_initiated=True, skip_artist_ids=[skipped.item_id]
134    )
135
136    assert await _play_count(mass, DB_TABLE_ALBUMS, album.item_id) == 1
137    assert await _play_count(mass, DB_TABLE_ARTISTS, kept.item_id) == 1
138    assert await _play_count(mass, DB_TABLE_ARTISTS, skipped.item_id) == 0
139
140
141async def test_track_credit_does_not_flag_artist_user_initiated(mass: MusicAssistant) -> None:
142    """A track play credits its artist as a side-effect, never as user-initiated."""
143    user = await mass.webserver.auth.create_user("trackcreditui")
144    artist = await _add_artist(mass, "Sideeffect")
145    track = await _add_track(mass, "A Song", [artist])
146
147    await mass.music.mark_item_played(
148        track, fully_played=True, user_initiated=True, userid=user.user_id
149    )
150
151    assert await _artist_user_initiated(mass, artist.item_id, user.user_id) == 0
152
153
154async def test_explicit_artist_play_survives_track_credit(mass: MusicAssistant) -> None:
155    """An explicit artist play stays user-initiated after later side-effect credits."""
156    user = await mass.webserver.auth.create_user("artiststicky")
157    artist = await _add_artist(mass, "Chosen")
158    track = await _add_track(mass, "Their Song", [artist])
159
160    # user explicitly plays the artist
161    await mass.music.mark_item_played(
162        artist, fully_played=True, user_initiated=True, userid=user.user_id
163    )
164    assert await _artist_user_initiated(mass, artist.item_id, user.user_id) == 1
165
166    # later, one of the artist's tracks auto-plays (not user-initiated)
167    await mass.music.mark_item_played(
168        track, fully_played=True, user_initiated=False, userid=user.user_id
169    )
170    assert await _artist_user_initiated(mass, artist.item_id, user.user_id) == 1
171
172
173def test_enqueued_album_decision() -> None:
174    """An album is credited only when enqueued, and only once per enqueue."""
175    album_x = Album(
176        item_id="ax",
177        provider="library",
178        name="X",
179        provider_mappings=set(),
180        album_type=AlbumType.ALBUM,
181    )
182    album_y = Album(
183        item_id="ay",
184        provider="library",
185        name="Y",
186        provider_mappings=set(),
187        album_type=AlbumType.ALBUM,
188    )
189    t1 = Track(item_id="t1", provider="library", name="T1", provider_mappings=set(), album=album_x)
190    t2 = Track(item_id="t2", provider="library", name="T2", provider_mappings=set(), album=album_x)
191    t3 = Track(item_id="t3", provider="library", name="T3", provider_mappings=set(), album=album_y)
192
193    tracker = PlayerQueuesController.__new__(PlayerQueuesController)
194    data = _queue_data(enqueued=[album_x])
195
196    # first of the album's tracks to complete -> credit it
197    assert tracker._claim_enqueued_album_credit(data, t1) is album_x
198    # any further track of the same album -> already credited for this enqueue
199    assert tracker._claim_enqueued_album_credit(data, t2) is None
200    # a track whose album was never enqueued -> not credited
201    assert tracker._claim_enqueued_album_credit(data, t3) is None
202
203
204def test_enqueued_album_credited_once_however_its_tracks_are_ordered() -> None:
205    """Album tracks split up or reordered in the queue still credit the album exactly once."""
206    album_x = Album(
207        item_id="ax",
208        provider="library",
209        name="X",
210        provider_mappings=set(),
211        album_type=AlbumType.ALBUM,
212    )
213    album_y = Album(
214        item_id="ay",
215        provider="library",
216        name="Y",
217        provider_mappings=set(),
218        album_type=AlbumType.ALBUM,
219    )
220    x_tracks = [
221        Track(
222            item_id=f"x{i}",
223            provider="library",
224            name=f"X{i}",
225            provider_mappings=set(),
226            album=album_x,
227        )
228        for i in range(4)
229    ]
230    y_track = Track(
231        item_id="y1", provider="library", name="Y1", provider_mappings=set(), album=album_y
232    )
233
234    tracker = PlayerQueuesController.__new__(PlayerQueuesController)
235    # 'play next' on a track from another album splits the album's run in two
236    data = _queue_data(enqueued=[album_x])
237    played = [x_tracks[0], x_tracks[1], y_track, x_tracks[2], x_tracks[3]]
238    credited = [tracker._claim_enqueued_album_credit(data, track) for track in played]
239    assert [c for c in credited if c is not None] == [album_x]
240
241    # the album's first track is skipped, so it never completes and the claim arrives from a
242    # later track that sits behind a same-album predecessor; the album is still credited once
243    data = _queue_data(enqueued=[album_x])
244    played = x_tracks[1:]
245    credited = [tracker._claim_enqueued_album_credit(data, track) for track in played]
246    assert [c for c in credited if c is not None] == [album_x]
247
248
249def test_reenqueued_album_in_another_shape_is_credited_again() -> None:
250    """Re-enqueueing an album from a different listing arms its credit again."""
251    mapping = ProviderMapping(
252        item_id="album-prov-1", provider_domain="spotify", provider_instance="spotify--abc"
253    )
254    provider_album = Album(
255        item_id="album-prov-1",
256        provider="spotify--abc",
257        name="Kind of Blue",
258        provider_mappings={mapping},
259        album_type=AlbumType.ALBUM,
260    )
261    library_album = Album(
262        item_id="7",
263        provider="library",
264        name="Kind of Blue",
265        provider_mappings={mapping},
266        album_type=AlbumType.ALBUM,
267    )
268    track = Track(
269        item_id="t1", provider="library", name="T1", provider_mappings=set(), album=library_album
270    )
271
272    tracker = PlayerQueuesController.__new__(PlayerQueuesController)
273    # played from the provider listing first, then queued again from the library view
274    data = _queue_data(enqueued=[provider_album])
275    assert tracker._claim_enqueued_album_credit(data, track) is library_album
276    data.enqueued_media_items.append(library_album)
277    data.credited_albums.discard(library_album)
278
279    assert tracker._claim_enqueued_album_credit(data, track) is library_album
280
281
282def test_enqueued_provider_album_credits_its_library_tracks() -> None:
283    """An album picked from a provider listing is credited, though its tracks are library ones."""
284    mapping = ProviderMapping(
285        item_id="album-prov-1", provider_domain="spotify", provider_instance="spotify--abc"
286    )
287    # the album object a provider listing hands to play_media
288    provider_album = Album(
289        item_id="album-prov-1",
290        provider="spotify--abc",
291        name="Kind of Blue",
292        provider_mappings={mapping},
293        album_type=AlbumType.ALBUM,
294    )
295    # the album its tracks carry, because the album is also in the library
296    library_album = Album(
297        item_id="7",
298        provider="library",
299        name="Kind of Blue",
300        provider_mappings={mapping},
301        album_type=AlbumType.ALBUM,
302    )
303    other_album = Album(
304        item_id="8",
305        provider="library",
306        name="Sketches of Spain",
307        provider_mappings={
308            ProviderMapping(
309                item_id="album-prov-2",
310                provider_domain="spotify",
311                provider_instance="spotify--abc",
312            )
313        },
314        album_type=AlbumType.ALBUM,
315    )
316    t1 = Track(item_id="t1", provider="library", name="T1", provider_mappings=set())
317    t2 = Track(item_id="t2", provider="library", name="T2", provider_mappings=set())
318    t3 = Track(item_id="t3", provider="library", name="T3", provider_mappings=set())
319    # loading an item for playback puts the full library album on it
320    t1.album = library_album
321    t2.album = library_album
322    t3.album = other_album
323
324    tracker = PlayerQueuesController.__new__(PlayerQueuesController)
325    data = _queue_data(enqueued=[provider_album])
326
327    # the enqueued album is credited on the first of its tracks to complete, under the
328    # library identity so it shares the row an explicit library play writes
329    assert tracker._claim_enqueued_album_credit(data, t1) is library_album
330    # and only once for that enqueue
331    assert tracker._claim_enqueued_album_credit(data, t2) is None
332    # a track from an album the user never enqueued is not credited
333    assert tracker._claim_enqueued_album_credit(data, t3) is None
334
335
336def test_album_outside_the_library_is_credited_as_the_provider_album() -> None:
337    """An album that is not in the library is credited under the provider identity."""
338    provider_album = Album(
339        item_id="album-prov-1",
340        provider="spotify--abc",
341        name="Kind of Blue",
342        provider_mappings={
343            ProviderMapping(
344                item_id="album-prov-1",
345                provider_domain="spotify",
346                provider_instance="spotify--abc",
347            )
348        },
349        album_type=AlbumType.ALBUM,
350    )
351    track = Track(item_id="t1", provider="spotify--abc", name="T1", provider_mappings=set())
352    # with no library album to swap in, the track keeps the provider album as a mapping
353    track.album = ItemMapping.from_item(provider_album)
354
355    tracker = PlayerQueuesController.__new__(PlayerQueuesController)
356    data = _queue_data(enqueued=[provider_album])
357
358    assert tracker._claim_enqueued_album_credit(data, track) is provider_album
359