/
/
/
1"""
2Tests for the playlog updated event.
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.
7"""
8
9from __future__ import annotations
10
11import asyncio
12from typing import TYPE_CHECKING
13from uuid import uuid4
14
15from music_assistant_models.enums import EventType, MediaType
16from music_assistant_models.media_items import (
17 Artist,
18 Podcast,
19 PodcastEpisode,
20 ProviderMapping,
21 Track,
22)
23from music_assistant_models.playlog_update import PlaylogUpdate
24from music_assistant_models.unique_list import UniqueList
25
26from music_assistant.mass import MusicAssistant
27
28if TYPE_CHECKING:
29 from music_assistant_models.event import MassEvent
30
31
32def _provider_mapping(provider: str = "test_prov") -> set[ProviderMapping]:
33 """Create a single provider mapping with a unique item id."""
34 return {
35 ProviderMapping(
36 item_id=uuid4().hex,
37 provider_domain=provider,
38 provider_instance=provider,
39 )
40 }
41
42
43def _collect_events(mass: MusicAssistant) -> list[MassEvent]:
44 """Return a list that collects the playlog updated events as they are signalled."""
45 events: list[MassEvent] = []
46 mass.subscribe(events.append, EventType.PLAYLOG_UPDATED)
47 return events
48
49
50async def _updates(events: list[MassEvent]) -> list[PlaylogUpdate]:
51 """Let the pending event callbacks run and return the collected payloads."""
52 await asyncio.sleep(0)
53 return [event.data for event in events]
54
55
56async def test_mark_played_signals_playlog_updated(mass: MusicAssistant) -> None:
57 """Marking an item played signals the new playlog state for that item."""
58 user = await mass.webserver.auth.create_user("playloguser")
59 track = Track(
60 item_id="track-001",
61 provider="test_prov",
62 name="Test Track",
63 provider_mappings=_provider_mapping(),
64 duration=200,
65 )
66 assert track.uri is not None
67 events = _collect_events(mass)
68
69 await mass.music.mark_item_played(track, fully_played=True, userid=user.user_id)
70
71 updates = await _updates(events)
72 assert len(updates) == 1
73 assert events[0].object_id == track.uri
74 assert updates[0] == PlaylogUpdate(
75 uri=track.uri,
76 media_type=MediaType.TRACK,
77 fully_played=True,
78 seconds_played=0,
79 userid=user.user_id,
80 )
81
82
83async def test_progress_report_signals_resume_position(mass: MusicAssistant) -> None:
84 """A partial play reports the resume position, so clients can update an in progress row."""
85 user = await mass.webserver.auth.create_user("playlogpartial")
86 episode = PodcastEpisode(
87 item_id="ep-010",
88 provider="test_podcast_prov",
89 name="Episode 10",
90 provider_mappings=_provider_mapping("test_podcast_prov"),
91 position=1,
92 podcast=Podcast(
93 item_id="show-011",
94 provider="test_podcast_prov",
95 name="Progress Show",
96 provider_mappings=_provider_mapping("test_podcast_prov"),
97 ),
98 )
99 assert episode.uri is not None
100 events = _collect_events(mass)
101
102 await mass.music.mark_item_played(
103 episode, fully_played=False, seconds_played=120, userid=user.user_id
104 )
105
106 updates = await _updates(events)
107 assert updates == [
108 PlaylogUpdate(
109 uri=episode.uri,
110 media_type=MediaType.PODCAST_EPISODE,
111 fully_played=False,
112 seconds_played=120,
113 userid=user.user_id,
114 )
115 ]
116
117
118async def test_playing_progress_report_signals_nothing(mass: MusicAssistant) -> None:
119 """The frequent progress reports during playback do not touch the playlog, so stay quiet."""
120 user = await mass.webserver.auth.create_user("playlogplaying")
121 track = Track(
122 item_id="track-002",
123 provider="test_prov",
124 name="Playing Track",
125 provider_mappings=_provider_mapping(),
126 duration=200,
127 )
128 events = _collect_events(mass)
129
130 await mass.music.mark_item_played(
131 track, fully_played=False, seconds_played=30, is_playing=True, userid=user.user_id
132 )
133
134 assert await _updates(events) == []
135
136
137async def test_mark_unplayed_signals_playlog_updated(mass: MusicAssistant) -> None:
138 """Marking an item unplayed signals it as neither played nor in progress."""
139 user = await mass.webserver.auth.create_user("playlogunplayed")
140 episode = PodcastEpisode(
141 item_id="ep-011",
142 provider="test_podcast_prov",
143 name="Episode 11",
144 provider_mappings=_provider_mapping("test_podcast_prov"),
145 position=1,
146 podcast=Podcast(
147 item_id="show-012",
148 provider="test_podcast_prov",
149 name="Unplayed Show",
150 provider_mappings=_provider_mapping("test_podcast_prov"),
151 ),
152 )
153 assert episode.uri is not None
154 await mass.music.mark_item_played(
155 episode, fully_played=False, seconds_played=120, userid=user.user_id
156 )
157 events = _collect_events(mass)
158
159 await mass.music.mark_item_unplayed(episode, userid=user.user_id)
160
161 assert await _updates(events) == [
162 PlaylogUpdate(
163 uri=episode.uri,
164 media_type=MediaType.PODCAST_EPISODE,
165 fully_played=False,
166 seconds_played=0,
167 userid=user.user_id,
168 )
169 ]
170
171
172async def test_mark_played_without_user_applies_to_all_users(mass: MusicAssistant) -> None:
173 """Without a determinable user the playlog changes for everyone, so userid is None."""
174 track = Track(
175 item_id="track-003",
176 provider="test_prov",
177 name="Shared Track",
178 provider_mappings=_provider_mapping(),
179 duration=200,
180 )
181 events = _collect_events(mass)
182
183 await mass.music.mark_item_played(track, fully_played=True)
184
185 updates = await _updates(events)
186 assert len(updates) == 1
187 assert updates[0].userid is None
188
189
190async def test_credited_artist_signals_playlog_updated(mass: MusicAssistant) -> None:
191 """An artist credited by a played track gets its own playlog update."""
192 user = await mass.webserver.auth.create_user("playlogartist")
193 artist = await mass.music.artists.add_item_to_library(
194 Artist(
195 item_id="0", provider="library", name="Credited", provider_mappings=_provider_mapping()
196 )
197 )
198 added = await mass.music.tracks.add_item_to_library(
199 Track(
200 item_id="0",
201 provider="library",
202 name="Credited Track",
203 provider_mappings=_provider_mapping(),
204 artists=UniqueList([artist]),
205 )
206 )
207 track = await mass.music.tracks.get_library_item(added.item_id)
208 events = _collect_events(mass)
209
210 await mass.music.mark_item_played(track, fully_played=True, userid=user.user_id)
211
212 updates = await _updates(events)
213 assert [update.uri for update in updates] == [track.uri, artist.uri]
214 assert updates[1].media_type == MediaType.ARTIST
215 assert updates[1].userid == user.user_id
216
217
218async def test_credited_podcast_signals_playlog_updated(mass: MusicAssistant) -> None:
219 """The parent podcast credited by a played episode gets its own playlog update."""
220 user = await mass.webserver.auth.create_user("playlogpodcast")
221 podcast = Podcast(
222 item_id="show-010",
223 provider="test_podcast_prov",
224 name="Credited Show",
225 provider_mappings=_provider_mapping("test_podcast_prov"),
226 )
227 episode = PodcastEpisode(
228 item_id="ep-012",
229 provider="test_podcast_prov",
230 name="Episode 12",
231 provider_mappings=_provider_mapping("test_podcast_prov"),
232 position=1,
233 podcast=podcast,
234 )
235 events = _collect_events(mass)
236
237 await mass.music.mark_item_played(episode, fully_played=True, userid=user.user_id)
238
239 updates = await _updates(events)
240 assert [update.uri for update in updates] == [episode.uri, podcast.uri]
241 assert updates[1].media_type == MediaType.PODCAST
242