music-assistant-server

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