/
/
/
1"""
2Tests for the stickiness of the playlog's ``user_initiated`` flag.
3
4The flag is what the Discover "Recently played" row filters on, so a play the user
5explicitly asked for must not be demoted by a later background play of the same item.
6These integration tests use the ``mass`` fixture from ``tests/conftest.py`` which
7creates a full MusicAssistant instance with a real SQLite database in a temporary
8directory.
9"""
10
11from __future__ import annotations
12
13from typing import Any
14from uuid import uuid4
15
16from music_assistant_models.enums import MediaType
17from music_assistant_models.media_items import Artist, ProviderMapping, Track
18from music_assistant_models.unique_list import UniqueList
19
20from music_assistant.constants import DB_TABLE_PLAYLOG
21from music_assistant.mass import MusicAssistant
22
23
24def _library_mapping() -> set[ProviderMapping]:
25 """Create a single library provider mapping with a unique provider item id."""
26 return {
27 ProviderMapping(
28 item_id=uuid4().hex,
29 provider_domain="library",
30 provider_instance="library",
31 in_library=True,
32 )
33 }
34
35
36async def _add_track(mass: MusicAssistant, name: str) -> Track:
37 """Add a minimal track to the library and return the (re-fetched) stored item."""
38 artist = await mass.music.artists.add_item_to_library(
39 Artist(
40 item_id="0",
41 provider="library",
42 name=f"{name} Artist",
43 provider_mappings=_library_mapping(),
44 )
45 )
46 added = await mass.music.tracks.add_item_to_library(
47 Track(
48 item_id="0",
49 provider="library",
50 name=name,
51 duration=180,
52 provider_mappings=_library_mapping(),
53 artists=UniqueList([artist]),
54 )
55 )
56 return await mass.music.tracks.get_library_item(added.item_id)
57
58
59async def _playlog_row(mass: MusicAssistant, track: Track, userid: str) -> dict[str, Any]:
60 """Read the track's playlog row for the given user."""
61 row = await mass.music.database.get_row(
62 DB_TABLE_PLAYLOG,
63 {
64 "item_id": track.item_id,
65 "provider": "library",
66 "media_type": MediaType.TRACK.value,
67 "userid": userid,
68 },
69 )
70 assert row is not None
71 return dict(row)
72
73
74async def test_explicit_track_play_survives_later_autoplay(mass: MusicAssistant) -> None:
75 """An autoplay/DSTM replay of an explicitly played track must not clear the flag."""
76 user = await mass.webserver.auth.create_user("stickytrack")
77 track = await _add_track(mass, "Sticky")
78
79 await mass.music.mark_item_played(
80 track, fully_played=True, user_initiated=True, userid=user.user_id
81 )
82 assert (await _playlog_row(mass, track, user.user_id))["user_initiated"] == 1
83
84 # the same track comes round again as part of a radio/autoplay mix
85 await mass.music.mark_item_played(
86 track, fully_played=True, user_initiated=False, seconds_played=180, userid=user.user_id
87 )
88
89 row = await _playlog_row(mass, track, user.user_id)
90 # the explicit play is remembered, so the track stays in the "Recently played" row ...
91 assert row["user_initiated"] == 1
92 # ... while everything else about the play is updated as usual
93 assert row["seconds_played"] == 180
94
95
96async def test_background_play_is_promoted_by_a_later_explicit_play(
97 mass: MusicAssistant,
98) -> None:
99 """A track first heard via autoplay becomes user-initiated once explicitly played."""
100 user = await mass.webserver.auth.create_user("promotedtrack")
101 track = await _add_track(mass, "Promoted")
102
103 await mass.music.mark_item_played(
104 track, fully_played=True, user_initiated=False, userid=user.user_id
105 )
106 assert (await _playlog_row(mass, track, user.user_id))["user_initiated"] == 0
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 _playlog_row(mass, track, user.user_id))["user_initiated"] == 1
113
114
115async def test_omitted_user_initiated_defaults_to_an_explicit_play(mass: MusicAssistant) -> None:
116 """
117 A caller that omits ``user_initiated`` records an explicit play.
118
119 ``music/mark_played`` is a public API command and clients back to ``MIN_SCHEMA_VERSION``
120 call it without the flag to report a play the user asked for, so the default is part of
121 the wire contract rather than an implementation detail. Because the flag is sticky, a
122 writer reporting playback it did not initiate must pass ``user_initiated=False`` itself.
123 """
124 user = await mass.webserver.auth.create_user("defaultedtrack")
125 track = await _add_track(mass, "Defaulted")
126
127 await mass.music.mark_item_played(track, userid=user.user_id)
128
129 assert (await _playlog_row(mass, track, user.user_id))["user_initiated"] == 1
130