/
/
/
1"""Tests for the Overcast provider's login, OPML fetch and resume position logic."""
2
3from __future__ import annotations
4
5import time
6from datetime import UTC, datetime
7from functools import partial
8from http.cookies import SimpleCookie
9from typing import Any, cast
10from unittest.mock import AsyncMock, MagicMock, Mock
11
12import pytest
13from music_assistant_models.enums import MediaType
14from music_assistant_models.errors import (
15 LoginFailed,
16 MediaNotFoundError,
17 ResourceTemporarilyUnavailable,
18)
19
20from music_assistant.providers.overcast import provider as overcast_provider
21from music_assistant.providers.overcast.helpers import (
22 OvercastEpisodeState,
23 OvercastSubscription,
24)
25from music_assistant.providers.overcast.provider import OvercastProvider
26
27FEED_A = "https://example.com/feed1.xml"
28FEED_B = "https://example.com/feed2.xml"
29
30
31class _FakeResponse:
32 def __init__(
33 self,
34 status: int = 200,
35 headers: dict[str, str] | None = None,
36 set_cookie: str | None = None,
37 body: str = "",
38 ) -> None:
39 self.status = status
40 self.headers = headers or {}
41 self.cookies: SimpleCookie = SimpleCookie()
42 if set_cookie is not None:
43 self.cookies.load(set_cookie)
44 self._body = body
45
46 async def text(self) -> str:
47 return self._body
48
49
50class _FakeRequestContext:
51 def __init__(self, response: _FakeResponse) -> None:
52 self._response = response
53
54 async def __aenter__(self) -> _FakeResponse:
55 return self._response
56
57 async def __aexit__(self, *exc_info: object) -> bool:
58 return False
59
60
61class _FakeSession:
62 """Fake aiohttp session returning canned responses in order."""
63
64 def __init__(self, responses: list[_FakeResponse]) -> None:
65 self._responses = responses
66 self.requests: list[tuple[str, str]] = []
67
68 def get(self, url: str, **kwargs: Any) -> _FakeRequestContext:
69 self.requests.append(("GET", url))
70 return _FakeRequestContext(self._responses.pop(0))
71
72 def post(self, url: str, **kwargs: Any) -> _FakeRequestContext:
73 self.requests.append(("POST", url))
74 return _FakeRequestContext(self._responses.pop(0))
75
76
77def _provider(responses: list[_FakeResponse]) -> MagicMock:
78 """Build a provider stub with a fake http session and recorded setup data."""
79 provider = MagicMock()
80 provider.http_session = _FakeSession(responses)
81 provider.get_setup_value = Mock(return_value="secret")
82 provider.setup_data_updates = {}
83 provider._update_setup_data = Mock(
84 side_effect=lambda key, value: provider.setup_data_updates.update({key: value})
85 )
86 # real rate limit bookkeeping, so a MagicMock is not mistaken for an active window
87 provider._rate_limited_until = None
88 provider._rate_limit_remaining = partial(OvercastProvider._rate_limit_remaining, provider)
89 return provider
90
91
92def _episode_state(**overrides: Any) -> OvercastEpisodeState:
93 state = OvercastEpisodeState(
94 overcast_id="1001",
95 title="Episode A",
96 enclosure_url="https://cdn.example.com/ep-a.mp3",
97 pub_date=None,
98 progress_s=123,
99 played=False,
100 user_updated_at=datetime(2026, 3, 1, 15, 0, 0, tzinfo=UTC),
101 )
102 for key, value in overrides.items():
103 setattr(state, key, value)
104 return state
105
106
107def _subscription(*states: OvercastEpisodeState) -> OvercastSubscription:
108 return OvercastSubscription(
109 xml_url="https://example.com/feed1.xml",
110 title="Feed One",
111 overcast_id="111",
112 episodes=list(states),
113 )
114
115
116async def test_login_success_persists_session_cookie() -> None:
117 """A 302 to /podcasts with the session cookie logs in and persists the cookie."""
118 provider = _provider(
119 [_FakeResponse(status=302, headers={"Location": "/podcasts"}, set_cookie="o=abc123")]
120 )
121 await OvercastProvider._login(cast("OvercastProvider", provider))
122 assert provider.setup_data_updates == {"session_cookie": "abc123"}
123
124
125async def test_login_rejected_credentials_raise_login_failed() -> None:
126 """A re-rendered login form (HTTP 200) raises LoginFailed."""
127 provider = _provider([_FakeResponse(status=200)])
128 with pytest.raises(LoginFailed):
129 await OvercastProvider._login(cast("OvercastProvider", provider))
130 assert provider.setup_data_updates == {}
131
132
133async def test_opml_request_returns_document() -> None:
134 """A successful export request returns the raw OPML text."""
135 provider = _provider([_FakeResponse(status=200, body="<opml/>")])
136 result = await OvercastProvider._request_opml(cast("OvercastProvider", provider))
137 assert result == "<opml/>"
138
139
140async def test_opml_rate_limit_maps_to_resource_temporarily_unavailable() -> None:
141 """A 429 with Retry-After maps to ResourceTemporarilyUnavailable with that backoff."""
142 provider = _provider([_FakeResponse(status=429, headers={"Retry-After": "900"})])
143 with pytest.raises(ResourceTemporarilyUnavailable) as err:
144 await OvercastProvider._request_opml(cast("OvercastProvider", provider))
145 assert err.value.backoff_time == 900
146
147
148async def test_rate_limit_spends_no_further_requests_while_active() -> None:
149 """After a 429 the export is not requested again until the window has elapsed."""
150 provider = _provider([_FakeResponse(status=429, headers={"Retry-After": "900"})])
151 typed_provider = cast("OvercastProvider", provider)
152 with pytest.raises(ResourceTemporarilyUnavailable):
153 await OvercastProvider._request_opml(typed_provider)
154 assert len(provider.http_session.requests) == 1
155
156 # the recorded window must fail the call outright, leaving the request count untouched
157 with pytest.raises(ResourceTemporarilyUnavailable) as err:
158 await OvercastProvider._request_opml(typed_provider)
159 assert len(provider.http_session.requests) == 1
160 assert 0 < err.value.backoff_time <= 900
161
162
163async def test_rate_limit_window_expiry_allows_a_new_request() -> None:
164 """Once the window has passed the export is requested again."""
165 provider = _provider([_FakeResponse(status=200, body="<opml/>")])
166 provider._rate_limited_until = time.monotonic() - 1
167 result = await OvercastProvider._request_opml(cast("OvercastProvider", provider))
168 assert result == "<opml/>"
169 assert provider._rate_limited_until is None
170
171
172async def test_opml_rejected_session_returns_none() -> None:
173 """A redirect to the login page signals an expired session by returning None."""
174 provider = _provider([_FakeResponse(status=302, headers={"Location": "/login"})])
175 assert await OvercastProvider._request_opml(cast("OvercastProvider", provider)) is None
176
177
178async def test_opml_fetch_relogins_once_on_expired_session() -> None:
179 """An expired session triggers exactly one re-login before the retry."""
180 provider = _provider(
181 [
182 _FakeResponse(status=302, headers={"Location": "/login"}),
183 _FakeResponse(status=302, headers={"Location": "/podcasts"}, set_cookie="o=new"),
184 _FakeResponse(status=200, body="<opml/>"),
185 ]
186 )
187 typed_provider = cast("OvercastProvider", provider)
188 provider._request_opml = partial(OvercastProvider._request_opml, typed_provider)
189 provider._login = partial(OvercastProvider._login, typed_provider)
190 # call the undecorated function to bypass the cache layer
191 fetch_opml = OvercastProvider._fetch_opml_text.__wrapped__ # type: ignore[attr-defined]
192 result = await fetch_opml(typed_provider)
193 assert result == "<opml/>"
194 assert provider.setup_data_updates == {"session_cookie": "new"}
195
196
197async def test_resume_position_maps_overcast_state() -> None:
198 """Overcast progress maps to (fully_played, position_ms, timestamp)."""
199 state = _episode_state()
200 provider = MagicMock()
201 provider._get_episode_stream_url = AsyncMock(return_value=state.enclosure_url)
202 provider._get_subscription = AsyncMock(return_value=_subscription(state))
203 result = await OvercastProvider.get_resume_position(
204 cast("OvercastProvider", provider),
205 "https://example.com/feed1.xml guid-a",
206 MediaType.PODCAST_EPISODE,
207 )
208 assert result == (False, 123000, state.user_updated_at)
209
210
211async def test_resume_position_without_progress_falls_back_to_playlog() -> None:
212 """Without any known Overcast progress the provider defers to MA's playlog."""
213 state = _episode_state(progress_s=None, played=False)
214 provider = MagicMock()
215 provider._get_episode_stream_url = AsyncMock(return_value=state.enclosure_url)
216 provider._get_subscription = AsyncMock(return_value=_subscription(state))
217 with pytest.raises(NotImplementedError):
218 await OvercastProvider.get_resume_position(
219 cast("OvercastProvider", provider),
220 "https://example.com/feed1.xml guid-a",
221 MediaType.PODCAST_EPISODE,
222 )
223
224
225async def test_apply_playback_states_respects_watermark() -> None:
226 """Only states newer than the watermark are pushed to the playlog."""
227 watermark = datetime(2026, 3, 1, 0, 0, 0, tzinfo=UTC)
228 old_state = _episode_state(
229 overcast_id="old",
230 enclosure_url="https://cdn.example.com/ep-old.mp3",
231 user_updated_at=datetime(2026, 2, 1, 0, 0, 0, tzinfo=UTC),
232 )
233 new_state = _episode_state(
234 overcast_id="new",
235 enclosure_url="https://cdn.example.com/ep-new.mp3",
236 played=True,
237 user_updated_at=datetime(2026, 3, 2, 0, 0, 0, tzinfo=UTC),
238 )
239 parsed_podcast = {
240 "title": "Feed One",
241 "cover_url": None,
242 "episodes": [
243 {"title": "Old", "guid": "g-old", "enclosures": [{"url": old_state.enclosure_url}]},
244 {"title": "New", "guid": "g-new", "enclosures": [{"url": new_state.enclosure_url}]},
245 ],
246 }
247 provider = MagicMock()
248 provider._feed_watermarks = {FEED_A: watermark}
249 provider.instance_id = "overcast--test"
250 provider.domain = "overcast"
251 provider.mass.music.mark_item_played = AsyncMock()
252 result = await OvercastProvider._apply_playback_states(
253 cast("OvercastProvider", provider),
254 "https://example.com/feed1.xml",
255 _subscription(old_state, new_state),
256 parsed_podcast,
257 )
258 assert result == new_state.user_updated_at
259 provider.mass.music.mark_item_played.assert_awaited_once()
260 call = provider.mass.music.mark_item_played.await_args
261 assert call.kwargs["fully_played"] is True
262 assert call.kwargs["seconds_played"] == 123
263 # the playlog flag is sticky, so a sync must never claim the user asked for the play
264 assert call.kwargs["user_initiated"] is False
265
266
267async def test_apply_playback_states_ignores_other_feeds_watermark() -> None:
268 """A feed is gated by its own watermark only, never by one of another feed."""
269 state = _episode_state(played=True, user_updated_at=datetime(2026, 2, 1, tzinfo=UTC))
270 parsed_podcast = {
271 "title": "Feed One",
272 "cover_url": None,
273 "episodes": [{"title": "A", "guid": "g-a", "enclosures": [{"url": state.enclosure_url}]}],
274 }
275 provider = MagicMock()
276 provider._feed_watermarks = {FEED_B: datetime(2026, 3, 1, tzinfo=UTC)}
277 provider.instance_id = "overcast--test"
278 provider.domain = "overcast"
279 provider.mass.music.mark_item_played = AsyncMock()
280 result = await OvercastProvider._apply_playback_states(
281 cast("OvercastProvider", provider),
282 FEED_A,
283 _subscription(state),
284 parsed_podcast,
285 )
286 assert result == state.user_updated_at
287 provider.mass.music.mark_item_played.assert_awaited_once()
288
289
290def _in_progress_provider(local_position_ms: int) -> MagicMock:
291 """Build a provider stub whose playlog already holds the given local position."""
292 provider = MagicMock()
293 provider._feed_watermarks = {}
294 provider.instance_id = "overcast--test"
295 provider.domain = "overcast"
296 provider.mass.music.get_resume_position = AsyncMock(return_value=(False, local_position_ms))
297 provider.mass.music.mark_item_played = AsyncMock()
298 return provider
299
300
301async def test_apply_playback_states_keeps_further_local_progress() -> None:
302 """An Overcast position behind the one MA recorded itself is not written to the playlog."""
303 state = _episode_state(progress_s=1200)
304 parsed_podcast = {
305 "title": "Feed One",
306 "cover_url": None,
307 "episodes": [{"title": "A", "guid": "g-a", "enclosures": [{"url": state.enclosure_url}]}],
308 }
309 provider = _in_progress_provider(local_position_ms=1800 * 1000)
310 result = await OvercastProvider._apply_playback_states(
311 cast("OvercastProvider", provider),
312 FEED_A,
313 _subscription(state),
314 parsed_podcast,
315 )
316 assert result is None
317 provider.mass.music.mark_item_played.assert_not_awaited()
318
319
320async def test_apply_playback_states_applies_progress_ahead_of_the_playlog() -> None:
321 """An Overcast position ahead of MA's own record is applied as usual."""
322 state = _episode_state(progress_s=1200)
323 parsed_podcast = {
324 "title": "Feed One",
325 "cover_url": None,
326 "episodes": [{"title": "A", "guid": "g-a", "enclosures": [{"url": state.enclosure_url}]}],
327 }
328 provider = _in_progress_provider(local_position_ms=600 * 1000)
329 result = await OvercastProvider._apply_playback_states(
330 cast("OvercastProvider", provider),
331 FEED_A,
332 _subscription(state),
333 parsed_podcast,
334 )
335 assert result == state.user_updated_at
336 assert provider.mass.music.mark_item_played.await_args.kwargs["seconds_played"] == 1200
337
338
339def _sync_provider(applied: datetime, monkeypatch: pytest.MonkeyPatch) -> MagicMock:
340 """Build a provider stub syncing two feeds, of which the second one is unreachable."""
341 provider = MagicMock()
342 provider._feed_watermarks = {}
343 provider.instance_id = "overcast--test"
344 provider.domain = "overcast"
345 provider.max_episodes = 0
346 provider._get_opml_subscriptions = AsyncMock(
347 return_value={FEED_A: _subscription(), FEED_B: _subscription()}
348 )
349 provider._apply_playback_states = AsyncMock(return_value=applied)
350 provider.mass.cache.set = AsyncMock()
351 provider._store_watermarks = partial(OvercastProvider._store_watermarks, provider)
352
353 async def _parsed_feed(*, feed_url: str, **_kwargs: Any) -> dict[str, Any]:
354 if feed_url == FEED_B:
355 raise MediaNotFoundError("feed is down")
356 return {"title": "Feed One", "episodes": []}
357
358 monkeypatch.setattr(overcast_provider, "refresh_cached_podcast", _parsed_feed)
359 monkeypatch.setattr(overcast_provider, "parse_podcast", Mock(return_value=MagicMock()))
360 return provider
361
362
363async def test_failing_feed_does_not_advance_another_feeds_watermark(
364 monkeypatch: pytest.MonkeyPatch,
365) -> None:
366 """An unreachable feed keeps its own (absent) watermark while others advance."""
367 applied = datetime(2026, 3, 2, tzinfo=UTC)
368 provider = _sync_provider(applied, monkeypatch)
369 podcasts = [
370 podcast
371 async for podcast in OvercastProvider.get_library_podcasts(
372 cast("OvercastProvider", provider)
373 )
374 ]
375 assert len(podcasts) == 1
376 assert provider._feed_watermarks == {FEED_A: applied}
377 assert provider.mass.cache.set.await_args.kwargs["data"] == {FEED_A: applied.isoformat()}
378
379
380async def test_watermark_of_unsubscribed_feed_is_retained(
381 monkeypatch: pytest.MonkeyPatch,
382) -> None:
383 """A feed that is no longer subscribed keeps its watermark, so it can be re-subscribed."""
384 applied = datetime(2026, 3, 2, tzinfo=UTC)
385 unsubscribed = datetime(2026, 1, 1, tzinfo=UTC)
386 provider = _sync_provider(applied, monkeypatch)
387 provider._feed_watermarks = {"https://example.com/gone.xml": unsubscribed}
388 async for _ in OvercastProvider.get_library_podcasts(cast("OvercastProvider", provider)):
389 pass
390 assert provider.mass.cache.set.await_args.kwargs["data"] == {
391 "https://example.com/gone.xml": unsubscribed.isoformat(),
392 FEED_A: applied.isoformat(),
393 }
394