/
/
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, 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