/
/
/
1"""
2Tests for tagging explicit (user-initiated) plays in the playlog.
3
4These cover the player-queue side of the decision: which plays count as
5user-initiated so they surface in the Discover "Recently played" row. The pure
6decisions are exercised as plain unit tests against a bare controller instance,
7mirroring ``test_play_report_dedup`` and ``test_enqueued_album_decision``.
8"""
9
10from __future__ import annotations
11
12from typing import cast
13from unittest.mock import AsyncMock, MagicMock, Mock
14
15import pytest
16from music_assistant_models.enums import AlbumType, MediaType, QueueOption
17from music_assistant_models.media_items import (
18 Album,
19 Artist,
20 Genre,
21 ItemMapping,
22 MediaItemType,
23 Playlist,
24 Podcast,
25 ProviderMapping,
26 Track,
27)
28from music_assistant_models.player_queue import PlayerQueue
29
30from music_assistant.controllers.player_queues import PlayerQueuesController
31from music_assistant.controllers.player_queues.media_resolver import MediaResolver
32from music_assistant.controllers.player_queues.state import PlayerQueueData
33
34
35def test_directly_enqueued_track_is_user_initiated() -> None:
36 """A track the user pressed play on is user-initiated; an album track is not."""
37 album = Album(
38 item_id="ax",
39 provider="library",
40 name="X",
41 provider_mappings=set(),
42 album_type=AlbumType.ALBUM,
43 )
44 explicit_track = Track(item_id="t1", provider="library", name="T1", provider_mappings=set())
45 album_track = Track(
46 item_id="t2", provider="library", name="T2", provider_mappings=set(), album=album
47 )
48 # the user explicitly played a single track and (separately) an album
49 data = cast("PlayerQueueData", Mock(enqueued_media_items=[explicit_track, album]))
50
51 tracker = PlayerQueuesController.__new__(PlayerQueuesController)
52 # the directly-enqueued track was explicitly chosen
53 assert tracker._is_user_initiated_play(data, explicit_track) is True
54 # a track that only played as part of the enqueued album was not
55 assert tracker._is_user_initiated_play(data, album_track) is False
56
57
58async def test_mark_album_played_is_user_initiated() -> None:
59 """Crediting an enqueued album records it as a user-initiated play."""
60 tracker = PlayerQueuesController.__new__(PlayerQueuesController)
61 tracker.logger = Mock()
62 mark = AsyncMock()
63 tracker.mass = Mock()
64 tracker.mass.music.mark_item_played = mark
65 tracker.mass.music.resolve_library_artist_ids = AsyncMock(return_value=set())
66
67 album = Album(
68 item_id="a1",
69 provider="library",
70 name="A",
71 provider_mappings=set(),
72 album_type=AlbumType.ALBUM,
73 )
74 track = Track(item_id="t1", provider="library", name="T", provider_mappings=set())
75 data = cast("PlayerQueueData", Mock(userid="u1", queue=Mock(queue_id="q1")))
76
77 await tracker._mark_album_played(album, track, data)
78
79 assert mark.call_args.kwargs["user_initiated"] is True
80
81
82def _resolver() -> tuple[MediaResolver, Mock]:
83 """Create a bare resolver plus the Mock recording its ``mark_item_played`` calls."""
84 resolver = MediaResolver.__new__(MediaResolver)
85 resolver.mass = Mock()
86 resolver.mass.create_task = Mock(side_effect=lambda coro: coro)
87 mark = Mock()
88 resolver.mass.music.mark_item_played = mark
89 return resolver, mark
90
91
92# the containers that are credited at resolve time, with the resolver method each one
93# expands through: (container, resolve method name, expected credited media type)
94_CONTAINERS = [
95 (
96 Playlist(item_id="pl1", provider="library", name="Pl", provider_mappings=set()),
97 "get_playlist_tracks",
98 MediaType.PLAYLIST,
99 ),
100 (
101 Artist(item_id="ar1", provider="library", name="Ar", provider_mappings=set()),
102 "get_artist_tracks",
103 MediaType.ARTIST,
104 ),
105 (
106 Genre(item_id="g1", provider="library", name="G", provider_mappings=set()),
107 "get_genre_tracks",
108 MediaType.GENRE,
109 ),
110 (
111 Podcast(item_id="pc1", provider="library", name="Pc", provider_mappings=set()),
112 "get_next_podcast_episodes",
113 MediaType.PODCAST,
114 ),
115]
116
117
118@pytest.mark.parametrize(("container", "resolve_method", "media_type"), _CONTAINERS)
119async def test_resolve_container_marks_user_initiated(
120 container: MediaItemType, resolve_method: str, media_type: MediaType
121) -> None:
122 """Enqueuing a container records the container itself as a user-initiated play."""
123 resolver, mark = _resolver()
124 track = Track(item_id="t1", provider="library", name="T1", provider_mappings=set())
125 setattr(resolver, resolve_method, AsyncMock(return_value=[track]))
126
127 await resolver._resolve_media_items(container, userid="u1", queue_id="q1")
128
129 assert mark.call_args.kwargs["user_initiated"] is True
130 assert mark.call_args.args[0].media_type == media_type
131
132
133@pytest.mark.parametrize(
134 ("container", "resolve_method"), [(item, method) for item, method, _ in _CONTAINERS]
135)
136async def test_resolve_empty_container_is_not_marked_played(
137 container: MediaItemType, resolve_method: str
138) -> None:
139 """A container that resolves to nothing playable is never written to the play history."""
140 resolver, mark = _resolver()
141 setattr(resolver, resolve_method, AsyncMock(return_value=[]))
142
143 assert await resolver._resolve_media_items(container, userid="u1", queue_id="q1") == []
144
145 mark.assert_not_called()
146
147
148async def test_enqueued_item_mapping_counts_as_user_initiated() -> None:
149 """
150 A play requested as an ItemMapping is recorded as explicitly enqueued.
151
152 Every mapping-shaped surface -- the "Recently played" row itself, search results,
153 artist/album track listings -- would otherwise never count as a user-initiated play.
154 """
155 track = Track(
156 item_id="t1",
157 provider="library",
158 name="T1",
159 provider_mappings={
160 ProviderMapping(item_id="t1", provider_domain="library", provider_instance="library")
161 },
162 )
163 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
164 ctrl.logger = Mock()
165 ctrl.mass = Mock()
166 ctrl.mass.music.get_item_by_uri = AsyncMock(return_value=track)
167 ctrl.mass.players.get_player = Mock(return_value=Mock(extra_data={}))
168 lock_cm = MagicMock()
169 lock_cm.__aenter__ = AsyncMock(return_value=None)
170 lock_cm.__aexit__ = AsyncMock(return_value=None)
171 ctrl.mass.players.get_player_lock = Mock(return_value=lock_cm)
172 ctrl._set_transitioning = Mock() # type: ignore[method-assign]
173 ctrl.signal_update = Mock() # type: ignore[method-assign]
174 ctrl.on_player_update = Mock() # type: ignore[method-assign]
175 ctrl.store_sources = Mock() # type: ignore[method-assign]
176 ctrl._enqueue_with_option = AsyncMock() # type: ignore[method-assign]
177 ctrl._media_resolver = Mock()
178 ctrl._media_resolver._resolve_media_items = AsyncMock(return_value=[track])
179 queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
180 ctrl.get = Mock(return_value=queue) # type: ignore[method-assign]
181 ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
182
183 # the user pressed play on a row that hands back a lightweight mapping, not a full Track
184 mapping = ItemMapping(
185 item_id="t1", provider="library", name="T1", media_type=MediaType.TRACK, uri=track.uri
186 )
187 await ctrl._handle_play_media("q1", mapping, QueueOption.PLAY)
188
189 # the mapping was resolved to its full item, so the play is attributable to the user
190 enqueued = ctrl._queue_data["q1"].enqueued_media_items
191 assert enqueued == [track]
192 assert ctrl._is_user_initiated_play(ctrl._queue_data["q1"], track) is True
193
194
195def test_track_picked_from_a_provider_listing_is_user_initiated() -> None:
196 """A track picked from a provider listing counts as explicit once it resolves to the library."""
197 mapping = ProviderMapping(
198 item_id="track-prov-1", provider_domain="spotify", provider_instance="spotify--abc"
199 )
200 # the track object a provider listing hands to play_media
201 provider_track = Track(
202 item_id="track-prov-1",
203 provider="spotify--abc",
204 name="T",
205 provider_mappings={mapping},
206 )
207 # the same track as it is reported once loaded for playback
208 library_track = Track(item_id="12", provider="library", name="T", provider_mappings={mapping})
209 data = cast("PlayerQueueData", Mock(enqueued_media_items=[provider_track]))
210
211 tracker = PlayerQueuesController.__new__(PlayerQueuesController)
212 assert tracker._is_user_initiated_play(data, library_track) is True
213
214
215def test_library_ids_are_not_matched_across_media_types() -> None:
216 """The library numbers each media type from one, so an album id never matches a track id."""
217 album = Album(
218 item_id="7",
219 provider="library",
220 name="A",
221 provider_mappings=set(),
222 album_type=AlbumType.ALBUM,
223 )
224 track = Track(item_id="7", provider="library", name="T", provider_mappings=set())
225 data = cast("PlayerQueueData", Mock(enqueued_media_items=[album]))
226
227 tracker = PlayerQueuesController.__new__(PlayerQueuesController)
228 assert tracker._is_user_initiated_play(data, track) is False
229
230
231def _play_media_controller(
232 media_item: MediaItemType, default_option: str
233) -> PlayerQueuesController:
234 """Build a bare controller whose play_media resolves its enqueue option from the config."""
235 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
236 ctrl.logger = Mock()
237 ctrl.mass = Mock()
238 ctrl.mass.music.get_item_by_uri = AsyncMock(return_value=media_item)
239 ctrl.mass.players.get_player = Mock(return_value=Mock(extra_data={}))
240 lock_cm = MagicMock()
241 lock_cm.__aenter__ = AsyncMock(return_value=None)
242 lock_cm.__aexit__ = AsyncMock(return_value=None)
243 ctrl.mass.players.get_player_lock = Mock(return_value=lock_cm)
244 ctrl.get_config_value = Mock(return_value=default_option) # type: ignore[method-assign]
245 ctrl._set_transitioning = Mock() # type: ignore[method-assign]
246 ctrl.signal_update = Mock() # type: ignore[method-assign]
247 ctrl.on_player_update = Mock() # type: ignore[method-assign]
248 ctrl.store_sources = Mock() # type: ignore[method-assign]
249 ctrl._apply_shuffle = AsyncMock() # type: ignore[method-assign]
250 ctrl._enqueue_with_option = AsyncMock() # type: ignore[method-assign]
251 ctrl._media_resolver = Mock()
252 ctrl._media_resolver._resolve_media_items = AsyncMock(return_value=[media_item])
253 queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
254 ctrl.get = Mock(return_value=queue) # type: ignore[method-assign]
255 ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
256 return ctrl
257
258
259async def test_configured_add_default_keeps_the_previously_enqueued_items() -> None:
260 """A caller that leaves the enqueue option to the config still gets an add treated as one."""
261 album = Album(item_id="a1", provider="library", name="A1", provider_mappings=set())
262 track = Track(
263 item_id="t1",
264 provider="library",
265 name="T1",
266 provider_mappings={
267 ProviderMapping(
268 item_id="track-prov-1",
269 provider_domain="spotify",
270 provider_instance="spotify--abc",
271 )
272 },
273 )
274 ctrl = _play_media_controller(track, QueueOption.ADD.value)
275 ctrl._queue_data["q1"].enqueued_media_items.append(album)
276
277 await ctrl._handle_play_media("q1", track)
278
279 # the album the queue is playing is what says its tracks belong together, so an add
280 # must not drop it the way starting a new queue does
281 assert ctrl._queue_data["q1"].enqueued_media_items == [album, track]
282
283
284async def test_configured_replace_default_clears_the_previously_enqueued_items() -> None:
285 """A config default that starts a new queue drops the parents of the previous one."""
286 album = Album(item_id="a1", provider="library", name="A1", provider_mappings=set())
287 track = Track(
288 item_id="t1",
289 provider="library",
290 name="T1",
291 provider_mappings={
292 ProviderMapping(
293 item_id="track-prov-1",
294 provider_domain="spotify",
295 provider_instance="spotify--abc",
296 )
297 },
298 )
299 ctrl = _play_media_controller(track, QueueOption.REPLACE.value)
300 ctrl._queue_data["q1"].enqueued_media_items.append(album)
301 ctrl._queue_data["q1"].credited_albums.add(album)
302
303 await ctrl._handle_play_media("q1", track)
304
305 assert ctrl._queue_data["q1"].enqueued_media_items == [track]
306 # the credits only mark which enqueued albums were counted, so they go with them
307 assert ctrl._queue_data["q1"].credited_albums == set()
308
309
310async def test_configured_add_default_feeds_a_dynamic_queue() -> None:
311 """An add onto a managed pool keeps the item as a source, config default or not."""
312 track = Track(
313 item_id="t1",
314 provider="library",
315 name="T1",
316 provider_mappings={
317 ProviderMapping(
318 item_id="track-prov-1",
319 provider_domain="spotify",
320 provider_instance="spotify--abc",
321 )
322 },
323 )
324 dynamic_playlist = Playlist(
325 item_id="p1",
326 provider="spotify--abc",
327 name="Mix",
328 provider_mappings={
329 ProviderMapping(
330 item_id="p1", provider_domain="spotify", provider_instance="spotify--abc"
331 )
332 },
333 is_dynamic=True,
334 )
335 ctrl = _play_media_controller(track, QueueOption.ADD.value)
336 ctrl._enter_dynamic_mode = AsyncMock() # type: ignore[method-assign]
337 queue_data = ctrl._queue_data["q1"]
338 queue_data.source_items = [dynamic_playlist]
339 queue_data.queue.is_dynamic = True
340 ctrl.store_sources = Mock( # type: ignore[method-assign]
341 side_effect=lambda _queue, items: setattr(queue_data, "source_items", list(items))
342 )
343
344 await ctrl._handle_play_media("q1", track)
345
346 # expanding it into the queue would only have the pool rebuild discard those tracks again
347 cast("AsyncMock", ctrl._media_resolver._resolve_media_items).assert_not_called()
348 assert track in queue_data.source_items
349