/
/
/
1"""
2Tests that the playlog commands accept the minimized items the Discover rows hand out.
3
4Recommendation rows such as 'In progress' and 'Recently played' return ItemMapping objects
5rather than full media items, and the frontend posts one of those straight back when the
6user picks 'Mark as played'. See music-assistant/support#6131.
7"""
8
9from __future__ import annotations
10
11from collections.abc import Mapping
12from typing import Any
13from unittest.mock import MagicMock, patch
14
15from music_assistant_models.enums import MediaType
16from music_assistant_models.media_items import (
17 Audiobook,
18 ItemMapping,
19 Podcast,
20 PodcastEpisode,
21 ProviderMapping,
22)
23
24from music_assistant.constants import DB_TABLE_PLAYLOG
25from music_assistant.helpers.api import parse_arguments
26from music_assistant.mass import MusicAssistant
27from music_assistant.models.music_provider import MusicProvider
28
29PROVIDER_ID = "podcastfeed--LxfKPLhS"
30PODCAST_ID = "stay-forever"
31EPISODE_ID = "49d77a72faa63de8145f1d8d7b3a17d9"
32EPISODE_NAME = "Forbidden Forest (SF 130)"
33AUDIOBOOK_PROVIDER = "filesystem_local--AbCd"
34AUDIOBOOK_ID = "book-001"
35
36
37class _StubPodcastProvider(MusicProvider):
38 """Minimal music provider that resolves a single podcast episode."""
39
40 def __init__(self) -> None:
41 """Initialize the stub provider."""
42 self.config = MagicMock()
43 self.config.instance_id = PROVIDER_ID
44 self.manifest = MagicMock()
45 self.manifest.domain = PROVIDER_ID
46 self.logger = MagicMock()
47
48 async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
49 """Return the full episode for the given provider episode id."""
50 return PodcastEpisode(
51 item_id=prov_episode_id,
52 provider=PROVIDER_ID,
53 name=EPISODE_NAME,
54 position=130,
55 duration=7200,
56 provider_mappings={
57 ProviderMapping(
58 item_id=prov_episode_id,
59 provider_domain=PROVIDER_ID,
60 provider_instance=PROVIDER_ID,
61 )
62 },
63 podcast=Podcast(
64 item_id=PODCAST_ID,
65 provider=PROVIDER_ID,
66 name="Stay Forever",
67 provider_mappings={
68 ProviderMapping(
69 item_id=PODCAST_ID,
70 provider_domain=PROVIDER_ID,
71 provider_instance=PROVIDER_ID,
72 )
73 },
74 ),
75 )
76
77
78def _minimized_payload(*, fully_played: bool) -> dict[str, Any]:
79 """
80 Build the media_item payload the frontend posts for an 'In progress' row item.
81
82 :param fully_played: The played state the frontend stamps onto the item before sending.
83 """
84 return {
85 "item_id": EPISODE_ID,
86 "provider": PROVIDER_ID,
87 "name": EPISODE_NAME,
88 "version": "",
89 "sort_name": "forbidden forest (sf 130)",
90 "uri": f"{PROVIDER_ID}://podcast_episode/{EPISODE_ID}",
91 "external_ids": [],
92 "is_playable": True,
93 "media_type": "podcast_episode",
94 "available": True,
95 "image": None,
96 "year": None,
97 "fully_played": fully_played,
98 "resume_position_ms": 0,
99 }
100
101
102def _minimized_episode() -> ItemMapping:
103 """Return the episode as in_progress_items hands it to the frontend."""
104 return ItemMapping(
105 item_id=EPISODE_ID,
106 provider=PROVIDER_ID,
107 name=EPISODE_NAME,
108 media_type=MediaType.PODCAST_EPISODE,
109 )
110
111
112async def _playlog_rows(mass: MusicAssistant) -> list[Mapping[str, Any]]:
113 """Return every playlog row for the episode under test."""
114 return await mass.music.database.get_rows(
115 DB_TABLE_PLAYLOG,
116 {"media_type": MediaType.PODCAST_EPISODE.value, "item_id": EPISODE_ID},
117 )
118
119
120async def test_mark_played_command_accepts_a_minimized_media_item(
121 mass: MusicAssistant,
122) -> None:
123 """The raw payload posted for an 'In progress' row parses into the command's arguments."""
124 handler = mass.command_handlers["music/mark_played"]
125
126 args = parse_arguments(
127 handler.signature,
128 handler.type_hints,
129 {"media_item": _minimized_payload(fully_played=True), "fully_played": True},
130 )
131
132 assert isinstance(args["media_item"], ItemMapping)
133 assert args["media_item"].uri == f"{PROVIDER_ID}://podcast_episode/{EPISODE_ID}"
134
135
136async def test_mark_unplayed_command_accepts_a_minimized_media_item(
137 mass: MusicAssistant,
138) -> None:
139 """The same payload parses for the mark_unplayed command."""
140 handler = mass.command_handlers["music/mark_unplayed"]
141
142 args = parse_arguments(
143 handler.signature,
144 handler.type_hints,
145 {"media_item": _minimized_payload(fully_played=False)},
146 )
147
148 assert isinstance(args["media_item"], ItemMapping)
149
150
151async def test_mark_played_records_a_minimized_podcast_episode(mass: MusicAssistant) -> None:
152 """Marking an 'In progress' row item as played writes its playlog row."""
153 user = await mass.webserver.auth.create_user("markplayedmapping")
154
155 with patch.object(mass, "get_provider", return_value=_StubPodcastProvider()):
156 await mass.music.mark_item_played(
157 _minimized_episode(), fully_played=True, userid=user.user_id
158 )
159
160 rows = await _playlog_rows(mass)
161 assert len(rows) == 1
162 assert rows[0]["fully_played"]
163
164
165async def test_mark_unplayed_clears_a_minimized_podcast_episode(mass: MusicAssistant) -> None:
166 """Marking an 'In progress' row item as unplayed removes its playlog row again."""
167 user = await mass.webserver.auth.create_user("markunplayedmapping")
168
169 with patch.object(mass, "get_provider", return_value=_StubPodcastProvider()):
170 await mass.music.mark_item_played(
171 _minimized_episode(), fully_played=True, userid=user.user_id
172 )
173 assert await _playlog_rows(mass)
174
175 await mass.music.mark_item_unplayed(_minimized_episode(), userid=user.user_id)
176
177 assert not await _playlog_rows(mass)
178
179
180async def test_mark_played_keeps_the_provider_identity_of_a_library_backed_audiobook(
181 mass: MusicAssistant,
182) -> None:
183 """A row referencing the provider identity is written under that identity, not the library one."""
184 user = await mass.webserver.auth.create_user("markplayedaudiobook")
185 db_book = await mass.music.audiobooks.add_item_to_library(
186 Audiobook(
187 item_id=AUDIOBOOK_ID,
188 provider=AUDIOBOOK_PROVIDER,
189 name="A Library Audiobook",
190 provider_mappings={
191 ProviderMapping(
192 item_id=AUDIOBOOK_ID,
193 provider_domain="filesystem_local",
194 provider_instance=AUDIOBOOK_PROVIDER,
195 )
196 },
197 )
198 )
199
200 await mass.music.mark_item_played(
201 ItemMapping(
202 item_id=AUDIOBOOK_ID,
203 provider=AUDIOBOOK_PROVIDER,
204 name="A Library Audiobook",
205 media_type=MediaType.AUDIOBOOK,
206 ),
207 fully_played=True,
208 userid=user.user_id,
209 )
210
211 rows = await mass.music.database.get_rows(
212 DB_TABLE_PLAYLOG, {"media_type": MediaType.AUDIOBOOK.value}
213 )
214 assert [(row["item_id"], row["provider"]) for row in rows] == [
215 (AUDIOBOOK_ID, AUDIOBOOK_PROVIDER)
216 ]
217 assert db_book.item_id != AUDIOBOOK_ID
218
219
220async def test_mark_unplayed_clears_the_provider_identity_of_a_library_backed_audiobook(
221 mass: MusicAssistant,
222) -> None:
223 """An existing row under the provider identity is the one removed again."""
224 user = await mass.webserver.auth.create_user("markunplayedaudiobook")
225 await mass.music.audiobooks.add_item_to_library(
226 Audiobook(
227 item_id=AUDIOBOOK_ID,
228 provider=AUDIOBOOK_PROVIDER,
229 name="A Library Audiobook",
230 provider_mappings={
231 ProviderMapping(
232 item_id=AUDIOBOOK_ID,
233 provider_domain="filesystem_local",
234 provider_instance=AUDIOBOOK_PROVIDER,
235 )
236 },
237 )
238 )
239 # the row an 'In progress' tile is built from, written under the provider identity
240 await mass.music.database.insert(
241 DB_TABLE_PLAYLOG,
242 {
243 "item_id": AUDIOBOOK_ID,
244 "provider": AUDIOBOOK_PROVIDER,
245 "media_type": MediaType.AUDIOBOOK.value,
246 "name": "A Library Audiobook",
247 "userid": user.user_id,
248 "seconds_played": 120,
249 "fully_played": False,
250 "timestamp": 1000,
251 },
252 allow_replace=True,
253 )
254
255 await mass.music.mark_item_unplayed(
256 ItemMapping(
257 item_id=AUDIOBOOK_ID,
258 provider=AUDIOBOOK_PROVIDER,
259 name="A Library Audiobook",
260 media_type=MediaType.AUDIOBOOK,
261 ),
262 userid=user.user_id,
263 )
264
265 assert not await mass.music.database.get_rows(
266 DB_TABLE_PLAYLOG, {"media_type": MediaType.AUDIOBOOK.value}
267 )
268