/
/
/
1"""Tests for Pocket Casts playback status handling."""
2
3from __future__ import annotations
4
5from typing import Any, cast
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.enums import MediaType
10from music_assistant_models.errors import RetriesExhausted
11from music_assistant_models.media_items import PodcastEpisode
12
13from music_assistant.providers.pocketcasts import PocketCastsProvider
14from tests.common import use_real_create_task
15
16
17@pytest.fixture
18def client() -> AsyncMock:
19 """Return a mocked Pocket Casts API client with no show notes on offer."""
20 client = AsyncMock()
21 client.get_show_notes.return_value = {}
22 client.get_podcast.return_value = {"uuid": "podcast-1", "title": "Podcast One"}
23 return client
24
25
26@pytest.fixture
27def provider(client: AsyncMock) -> PocketCastsProvider:
28 """Return a PocketCastsProvider backed by the mocked API client and a cold cache."""
29 mass = AsyncMock()
30 # force a cache miss so the wrapped fetches always run
31 mass.cache.get_with_freshness = AsyncMock(return_value=(None, False, False))
32 use_real_create_task(mass)
33 manifest = MagicMock()
34 manifest.domain = "pocketcasts"
35 config = MagicMock()
36 config.instance_id = "pocketcasts"
37 config.get_value.return_value = None
38 prov = PocketCastsProvider(mass, manifest, config)
39 prov._client = client
40 return prov
41
42
43def _feed_episode(**overrides: Any) -> dict[str, Any]:
44 """Build a full-podcast feed episode (snake_case schema, no playback status)."""
45 return {
46 "uuid": "episode-1",
47 "title": "Episode 1",
48 "url": "https://example.com/ep1.mp3",
49 "file_type": "audio/mpeg",
50 "duration": 1800,
51 **overrides,
52 }
53
54
55async def test_sync_survives_episode_without_duration(
56 provider: PocketCastsProvider, client: AsyncMock
57) -> None:
58 """A feed episode with a null duration must not abort the episode listing."""
59 client.get_podcast_episodes.return_value = (
60 "Podcast One",
61 [
62 _feed_episode(uuid="episode-1", duration=None),
63 _feed_episode(uuid="episode-2", duration=1800),
64 ],
65 )
66 client.get_in_progress_episodes.return_value = []
67 client.get_history.return_value = []
68
69 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
70
71 assert [episode.item_id for episode in episodes] == [
72 "podcast-1:episode-1",
73 "podcast-1:episode-2",
74 ]
75 assert episodes[0].duration == 0
76 assert episodes[0].fully_played is False
77 assert episodes[0].resume_position_ms == 0
78
79
80async def test_sync_survives_null_status_fields(
81 provider: PocketCastsProvider, client: AsyncMock
82) -> None:
83 """Null playedUpTo/duration on an in-progress entry must not abort the listing."""
84 client.get_podcast_episodes.return_value = (
85 "Podcast One",
86 [_feed_episode(uuid="episode-1", duration=None)],
87 )
88 client.get_in_progress_episodes.return_value = [
89 {"uuid": "episode-1", "playedUpTo": None, "duration": None}
90 ]
91 client.get_history.return_value = []
92
93 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
94
95 assert len(episodes) == 1
96 assert episodes[0].fully_played is False
97 assert episodes[0].resume_position_ms == 0
98
99
100async def test_sync_still_marks_played_episodes(
101 provider: PocketCastsProvider, client: AsyncMock
102) -> None:
103 """An episode played past the threshold is still reported as fully played."""
104 client.get_podcast_episodes.return_value = ("Podcast One", [_feed_episode(duration=1000)])
105 client.get_in_progress_episodes.return_value = [
106 {"uuid": "episode-1", "playedUpTo": 950, "duration": 1000}
107 ]
108 client.get_history.return_value = []
109
110 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
111
112 assert episodes[0].fully_played is True
113 assert episodes[0].resume_position_ms == 0
114
115
116async def test_sync_reports_resume_position(
117 provider: PocketCastsProvider, client: AsyncMock
118) -> None:
119 """A partially played episode keeps its resume position."""
120 client.get_podcast_episodes.return_value = ("Podcast One", [_feed_episode(duration=1000)])
121 client.get_in_progress_episodes.return_value = [
122 {"uuid": "episode-1", "playedUpTo": 300, "duration": 1000}
123 ]
124 client.get_history.return_value = []
125
126 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
127
128 assert episodes[0].fully_played is False
129 assert episodes[0].resume_position_ms == 300000
130
131
132async def test_get_podcast_episode_handles_null_fields(
133 provider: PocketCastsProvider, client: AsyncMock
134) -> None:
135 """The single-episode endpoint must tolerate null duration/playedUpTo."""
136 client.get_episode_details.return_value = {
137 "uuid": "episode-1",
138 "title": "Episode 1",
139 "url": "https://example.com/ep1.mp3",
140 "fileType": "audio/mpeg",
141 "duration": None,
142 "playedUpTo": None,
143 "playingStatus": 1,
144 }
145
146 # call the undecorated function so the @use_cache wrapper stays out of the test
147 get_podcast_episode = cast("Any", PocketCastsProvider.get_podcast_episode).__wrapped__
148 episode = await get_podcast_episode(provider, "podcast-1:episode-1")
149
150 assert episode.fully_played is False
151 assert episode.resume_position_ms == 0
152
153
154async def test_get_resume_position_handles_null_fields(
155 provider: PocketCastsProvider, client: AsyncMock
156) -> None:
157 """A null playedUpTo/duration on an in-progress entry yields a zero resume point."""
158 client.get_in_progress_episodes.return_value = [
159 {"uuid": "episode-1", "playedUpTo": None, "duration": None}
160 ]
161
162 assert await provider.get_resume_position("podcast-1:episode-1", MediaType.PODCAST_EPISODE) == (
163 False,
164 0,
165 None,
166 )
167
168
169async def test_episodes_get_their_own_description_and_artwork(
170 provider: PocketCastsProvider, client: AsyncMock
171) -> None:
172 """An episode uses its own show notes and artwork when the podcast supplies them."""
173 client.get_podcast_episodes.return_value = (
174 "Podcast One",
175 [
176 _feed_episode(uuid="episode-1"),
177 _feed_episode(uuid="episode-2"),
178 ],
179 )
180 client.get_in_progress_episodes.return_value = []
181 client.get_history.return_value = []
182 client.get_show_notes.return_value = {
183 "episode-1": {
184 "description": "<p>All about episode one.</p>",
185 "image": "https://example.com/ep1.jpg",
186 }
187 }
188
189 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
190
191 assert episodes[0].metadata.description == "<p>All about episode one.</p>"
192 assert episodes[0].metadata.images is not None
193 assert episodes[0].metadata.images[0].path == "https://example.com/ep1.jpg"
194
195
196async def test_episodes_without_artwork_keep_the_podcast_cover(
197 provider: PocketCastsProvider, client: AsyncMock
198) -> None:
199 """Only some episodes have their own artwork, the rest fall back to the podcast cover."""
200 client.get_podcast_episodes.return_value = ("Podcast One", [_feed_episode(uuid="episode-2")])
201 client.get_in_progress_episodes.return_value = []
202 client.get_history.return_value = []
203 client.get_show_notes.return_value = {"episode-2": {"description": "Just notes."}}
204
205 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
206
207 assert episodes[0].metadata.description == "Just notes."
208 assert episodes[0].metadata.images is not None
209 assert episodes[0].metadata.images[0].path.endswith("/podcast-1.jpg")
210
211
212async def test_sync_survives_unavailable_show_notes(
213 provider: PocketCastsProvider, client: AsyncMock
214) -> None:
215 """A failing show notes lookup must not abort the episode listing."""
216 client.get_podcast_episodes.return_value = ("Podcast One", [_feed_episode(uuid="episode-1")])
217 client.get_in_progress_episodes.return_value = []
218 client.get_history.return_value = []
219 client.get_show_notes.side_effect = RetriesExhausted("gave up")
220
221 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
222
223 assert [episode.item_id for episode in episodes] == ["podcast-1:episode-1"]
224 assert episodes[0].metadata.description is None
225
226
227async def test_single_episode_gets_its_description(
228 provider: PocketCastsProvider, client: AsyncMock
229) -> None:
230 """Fetching one episode also fills in its description."""
231 client.get_episode_details.return_value = {
232 "uuid": "episode-1",
233 "title": "Episode 1",
234 "url": "https://example.com/ep1.mp3",
235 "duration": 1800,
236 }
237 client.get_show_notes.return_value = {"episode-1": {"description": "All about it."}}
238
239 episode = await provider.get_podcast_episode("podcast-1:episode-1")
240
241 assert episode.metadata.description == "All about it."
242
243
244async def test_episodes_name_their_podcast(
245 provider: PocketCastsProvider, client: AsyncMock
246) -> None:
247 """An episode carries its podcast's name, which players show next to the title."""
248 client.get_podcast_episodes.return_value = ("Podcast One", [_feed_episode(uuid="episode-1")])
249 client.get_in_progress_episodes.return_value = []
250 client.get_history.return_value = []
251
252 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
253
254 assert episodes[0].podcast.name == "Podcast One"
255
256
257async def test_single_episode_names_its_podcast(
258 provider: PocketCastsProvider, client: AsyncMock
259) -> None:
260 """Fetching one episode also names its podcast."""
261 client.get_episode_details.return_value = {
262 "uuid": "episode-1",
263 "title": "Episode 1",
264 "url": "https://example.com/ep1.mp3",
265 "duration": 1800,
266 }
267
268 episode = await provider.get_podcast_episode("podcast-1:episode-1")
269
270 assert episode.podcast.name == "Podcast One"
271
272
273async def test_special_folder_episodes_name_their_podcast(
274 provider: PocketCastsProvider, client: AsyncMock
275) -> None:
276 """Episodes in the mixed folders name the podcast they came from."""
277 client.get_starred_episodes.return_value = [
278 {
279 "uuid": "episode-1",
280 "title": "Episode 1",
281 "url": "https://example.com/ep1.mp3",
282 "podcast": {"uuid": "podcast-1", "title": "Podcast One"},
283 }
284 ]
285
286 items = await provider.browse("pocketcasts://starred")
287
288 episodes = [item for item in items if isinstance(item, PodcastEpisode)]
289 assert [episode.podcast.name for episode in episodes] == ["Podcast One"]
290
291
292async def test_special_folder_episodes_use_the_name_from_the_payload(
293 provider: PocketCastsProvider, client: AsyncMock
294) -> None:
295 """A podcast title carried by the folder payload is used without a podcast lookup."""
296 client.get_history.return_value = [
297 {
298 "uuid": "episode-1",
299 "title": "Episode 1",
300 "url": "https://example.com/ep1.mp3",
301 "podcastUuid": "podcast-1",
302 "podcastTitle": "Podcast One",
303 }
304 ]
305
306 items = await provider.browse("pocketcasts://history")
307
308 episodes = [item for item in items if isinstance(item, PodcastEpisode)]
309 assert [episode.podcast.name for episode in episodes] == ["Podcast One"]
310 client.get_podcast.assert_not_called()
311
312
313async def test_special_folder_looks_each_podcast_up_once(
314 provider: PocketCastsProvider, client: AsyncMock
315) -> None:
316 """Episodes of the same podcast share a single name lookup for the whole folder."""
317 client.get_history.return_value = [
318 {
319 "uuid": f"episode-{index}",
320 "title": f"Episode {index}",
321 "url": f"https://example.com/ep{index}.mp3",
322 "podcast": "podcast-1",
323 }
324 for index in (1, 2, 3)
325 ]
326
327 items = await provider.browse("pocketcasts://history")
328
329 episodes = [item for item in items if isinstance(item, PodcastEpisode)]
330 assert [episode.podcast.name for episode in episodes] == ["Podcast One"] * 3
331 assert client.get_podcast.await_count == 1
332