/
/
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
10
11from music_assistant.providers.pocketcasts import PocketCastsProvider
12
13
14@pytest.fixture
15def client() -> AsyncMock:
16 """Return a mocked Pocket Casts API client."""
17 return AsyncMock()
18
19
20@pytest.fixture
21def provider(client: AsyncMock) -> PocketCastsProvider:
22 """Return a PocketCastsProvider backed by the mocked API client."""
23 manifest = MagicMock()
24 manifest.domain = "pocketcasts"
25 config = MagicMock()
26 config.instance_id = "pocketcasts"
27 config.get_value.return_value = None
28 prov = PocketCastsProvider(MagicMock(), manifest, config)
29 prov._client = client
30 return prov
31
32
33def _feed_episode(**overrides: Any) -> dict[str, Any]:
34 """Build a full-podcast feed episode (snake_case schema, no playback status)."""
35 return {
36 "uuid": "episode-1",
37 "title": "Episode 1",
38 "url": "https://example.com/ep1.mp3",
39 "file_type": "audio/mpeg",
40 "duration": 1800,
41 **overrides,
42 }
43
44
45async def test_sync_survives_episode_without_duration(
46 provider: PocketCastsProvider, client: AsyncMock
47) -> None:
48 """A feed episode with a null duration must not abort the episode listing."""
49 client.get_podcast_episodes.return_value = [
50 _feed_episode(uuid="episode-1", duration=None),
51 _feed_episode(uuid="episode-2", duration=1800),
52 ]
53 client.get_in_progress_episodes.return_value = []
54 client.get_history.return_value = []
55
56 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
57
58 assert [episode.item_id for episode in episodes] == [
59 "podcast-1:episode-1",
60 "podcast-1:episode-2",
61 ]
62 assert episodes[0].duration == 0
63 assert episodes[0].fully_played is False
64 assert episodes[0].resume_position_ms == 0
65
66
67async def test_sync_survives_null_status_fields(
68 provider: PocketCastsProvider, client: AsyncMock
69) -> None:
70 """Null playedUpTo/duration on an in-progress entry must not abort the listing."""
71 client.get_podcast_episodes.return_value = [_feed_episode(uuid="episode-1", duration=None)]
72 client.get_in_progress_episodes.return_value = [
73 {"uuid": "episode-1", "playedUpTo": None, "duration": None}
74 ]
75 client.get_history.return_value = []
76
77 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
78
79 assert len(episodes) == 1
80 assert episodes[0].fully_played is False
81 assert episodes[0].resume_position_ms == 0
82
83
84async def test_sync_still_marks_played_episodes(
85 provider: PocketCastsProvider, client: AsyncMock
86) -> None:
87 """An episode played past the threshold is still reported as fully played."""
88 client.get_podcast_episodes.return_value = [_feed_episode(duration=1000)]
89 client.get_in_progress_episodes.return_value = [
90 {"uuid": "episode-1", "playedUpTo": 950, "duration": 1000}
91 ]
92 client.get_history.return_value = []
93
94 episodes = [episode async for episode in provider.get_podcast_episodes("podcast-1")]
95
96 assert episodes[0].fully_played is True
97 assert episodes[0].resume_position_ms == 0
98
99
100async def test_sync_reports_resume_position(
101 provider: PocketCastsProvider, client: AsyncMock
102) -> None:
103 """A partially played episode keeps its resume position."""
104 client.get_podcast_episodes.return_value = [_feed_episode(duration=1000)]
105 client.get_in_progress_episodes.return_value = [
106 {"uuid": "episode-1", "playedUpTo": 300, "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 False
113 assert episodes[0].resume_position_ms == 300000
114
115
116async def test_get_podcast_episode_handles_null_fields(
117 provider: PocketCastsProvider, client: AsyncMock
118) -> None:
119 """The single-episode endpoint must tolerate null duration/playedUpTo."""
120 client.get_episode_details.return_value = {
121 "uuid": "episode-1",
122 "title": "Episode 1",
123 "url": "https://example.com/ep1.mp3",
124 "fileType": "audio/mpeg",
125 "duration": None,
126 "playedUpTo": None,
127 "playingStatus": 1,
128 }
129
130 # call the undecorated function so the @use_cache wrapper stays out of the test
131 get_podcast_episode = cast("Any", PocketCastsProvider.get_podcast_episode).__wrapped__
132 episode = await get_podcast_episode(provider, "podcast-1:episode-1")
133
134 assert episode.fully_played is False
135 assert episode.resume_position_ms == 0
136
137
138async def test_get_resume_position_handles_null_fields(
139 provider: PocketCastsProvider, client: AsyncMock
140) -> None:
141 """A null playedUpTo/duration on an in-progress entry yields a zero resume point."""
142 client.get_in_progress_episodes.return_value = [
143 {"uuid": "episode-1", "playedUpTo": None, "duration": None}
144 ]
145
146 assert await provider.get_resume_position("podcast-1:episode-1", MediaType.PODCAST_EPISODE) == (
147 False,
148 0,
149 None,
150 )
151