/
/
/
1"""
2Tests for crediting the parent podcast show when a podcast episode is fully played.
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
11from uuid import uuid4
12
13from music_assistant_models.enums import MediaType
14from music_assistant_models.media_items import Podcast, PodcastEpisode, ProviderMapping
15
16from music_assistant.constants import DB_TABLE_PLAYLOG
17from music_assistant.mass import MusicAssistant
18
19
20def _provider_mapping(provider: str = "test_podcast_prov") -> set[ProviderMapping]:
21 """Create a single provider mapping with a unique item id."""
22 return {
23 ProviderMapping(
24 item_id=uuid4().hex,
25 provider_domain=provider,
26 provider_instance=provider,
27 )
28 }
29
30
31async def test_fully_played_episode_credits_parent_podcast(mass: MusicAssistant) -> None:
32 """A fully played episode writes a playlog row for its parent podcast show."""
33 user = await mass.webserver.auth.create_user("podcastcredit")
34
35 podcast = Podcast(
36 item_id="show-001",
37 provider="test_podcast_prov",
38 name="My Podcast Show",
39 provider_mappings=_provider_mapping(),
40 )
41 episode = PodcastEpisode(
42 item_id="ep-001",
43 provider="test_podcast_prov",
44 name="Episode 1",
45 provider_mappings=_provider_mapping(),
46 position=1,
47 podcast=podcast,
48 )
49
50 await mass.music.mark_item_played(
51 episode, fully_played=True, user_initiated=False, userid=user.user_id
52 )
53
54 row = await mass.music.database.get_row(
55 DB_TABLE_PLAYLOG,
56 {
57 "media_type": MediaType.PODCAST.value,
58 "item_id": podcast.item_id,
59 "userid": user.user_id,
60 },
61 )
62 assert row is not None, "Expected a playlog row for the parent podcast show"
63 assert int(row["user_initiated"]) == 0
64
65
66async def test_user_initiated_show_play_stays_sticky_across_episode_credit(
67 mass: MusicAssistant,
68) -> None:
69 """An explicit user play of the show stays user_initiated after a later episode credit."""
70 user = await mass.webserver.auth.create_user("podcaststicky")
71
72 podcast = Podcast(
73 item_id="show-003",
74 provider="test_podcast_prov",
75 name="Sticky Show",
76 provider_mappings=_provider_mapping(),
77 )
78 episode = PodcastEpisode(
79 item_id="ep-003",
80 provider="test_podcast_prov",
81 name="Episode 3",
82 provider_mappings=_provider_mapping(),
83 position=1,
84 podcast=podcast,
85 )
86
87 # the user explicitly plays the show -> a user_initiated podcast row
88 await mass.music.mark_item_played(
89 podcast, fully_played=True, user_initiated=True, userid=user.user_id
90 )
91 # a later episode credit writes user_initiated=False; the ON CONFLICT merge must keep it sticky
92 await mass.music.mark_item_played(
93 episode, fully_played=True, user_initiated=False, userid=user.user_id
94 )
95
96 row = await mass.music.database.get_row(
97 DB_TABLE_PLAYLOG,
98 {
99 "media_type": MediaType.PODCAST.value,
100 "item_id": podcast.item_id,
101 "userid": user.user_id,
102 },
103 )
104 assert row is not None
105 assert int(row["user_initiated"]) == 1, (
106 "an explicit show play must not be downgraded by a credit"
107 )
108
109
110async def test_partial_episode_does_not_credit_podcast(mass: MusicAssistant) -> None:
111 """A partially played episode does NOT write a playlog row for the parent podcast."""
112 user = await mass.webserver.auth.create_user("podcastnocredit")
113
114 podcast = Podcast(
115 item_id="show-002",
116 provider="test_podcast_prov",
117 name="Another Podcast Show",
118 provider_mappings=_provider_mapping(),
119 )
120 episode = PodcastEpisode(
121 item_id="ep-002",
122 provider="test_podcast_prov",
123 name="Episode 2",
124 provider_mappings=_provider_mapping(),
125 position=1,
126 podcast=podcast,
127 )
128
129 await mass.music.mark_item_played(
130 episode, fully_played=False, seconds_played=30, user_initiated=False, userid=user.user_id
131 )
132
133 row = await mass.music.database.get_row(
134 DB_TABLE_PLAYLOG,
135 {
136 "media_type": MediaType.PODCAST.value,
137 "item_id": podcast.item_id,
138 "userid": user.user_id,
139 },
140 )
141 assert row is None, "Expected no playlog row for the parent podcast on partial play"
142
143
144async def test_library_podcast_episode_credits_library_scoped_row(
145 mass: MusicAssistant,
146) -> None:
147 """An episode of a library podcast credits the library row, not a provider-scoped duplicate."""
148 user = await mass.webserver.auth.create_user("podcastlibraryscoped")
149
150 ext_item_id = uuid4().hex
151 library_podcast = await mass.music.podcasts.add_item_to_library(
152 Podcast(
153 item_id="0",
154 provider="library",
155 name="Library Podcast Show",
156 provider_mappings={
157 ProviderMapping(
158 item_id=ext_item_id,
159 provider_domain="test_podcast_prov",
160 provider_instance="test_podcast_prov",
161 )
162 },
163 )
164 )
165
166 # an explicit user play of the library show -> a user_initiated library-scoped row
167 await mass.music.mark_item_played(
168 library_podcast, fully_played=True, user_initiated=True, userid=user.user_id
169 )
170
171 # the episode's parent podcast carries provider-scoped ids, as returned by the provider layer
172 provider_scoped_podcast = Podcast(
173 item_id=ext_item_id,
174 provider="test_podcast_prov",
175 name="Library Podcast Show",
176 provider_mappings=_provider_mapping(),
177 )
178 episode = PodcastEpisode(
179 item_id="ep-lib-001",
180 provider="test_podcast_prov",
181 name="Episode 1",
182 provider_mappings=_provider_mapping(),
183 position=1,
184 podcast=provider_scoped_podcast,
185 )
186 await mass.music.mark_item_played(
187 episode, fully_played=True, user_initiated=False, userid=user.user_id
188 )
189
190 row = await mass.music.database.get_row(
191 DB_TABLE_PLAYLOG,
192 {
193 "media_type": MediaType.PODCAST.value,
194 "item_id": library_podcast.item_id,
195 "provider": "library",
196 "userid": user.user_id,
197 },
198 )
199 assert row is not None, "Expected the episode credit to land on the library-scoped row"
200 assert int(row["user_initiated"]) == 1, (
201 "the episode credit must merge into the library row, not downgrade it"
202 )
203
204 dup_row = await mass.music.database.get_row(
205 DB_TABLE_PLAYLOG,
206 {
207 "media_type": MediaType.PODCAST.value,
208 "item_id": ext_item_id,
209 "userid": user.user_id,
210 },
211 )
212 assert dup_row is None, "Should not create a separate provider-scoped duplicate row"
213
214
215async def test_non_library_podcast_episode_credits_provider_scoped_row(
216 mass: MusicAssistant,
217) -> None:
218 """An episode of a podcast that isn't in the library still credits the provider-scoped row."""
219 user = await mass.webserver.auth.create_user("podcastnonlibraryscoped")
220
221 podcast = Podcast(
222 item_id="show-004",
223 provider="test_podcast_prov",
224 name="Non-Library Show",
225 provider_mappings=_provider_mapping(),
226 )
227 episode = PodcastEpisode(
228 item_id="ep-004",
229 provider="test_podcast_prov",
230 name="Episode 4",
231 provider_mappings=_provider_mapping(),
232 position=1,
233 podcast=podcast,
234 )
235
236 await mass.music.mark_item_played(
237 episode, fully_played=True, user_initiated=False, userid=user.user_id
238 )
239
240 row = await mass.music.database.get_row(
241 DB_TABLE_PLAYLOG,
242 {
243 "media_type": MediaType.PODCAST.value,
244 "item_id": podcast.item_id,
245 "provider": podcast.provider,
246 "userid": user.user_id,
247 },
248 )
249 assert row is not None, "Expected a provider-scoped playlog row for the non-library podcast"
250