/
/
/
1"""Tests for gpodder external chapter enrichment on the single-episode path."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any, cast
6from unittest.mock import AsyncMock, MagicMock
7
8from music_assistant.helpers.podcast_parsers import parse_podcast_episode
9from music_assistant.providers.gpodder import GPodder
10
11if TYPE_CHECKING:
12 from music_assistant_models.media_items import PodcastEpisode
13
14
15class _FakeResponse:
16 def __init__(self, payload: Any) -> None:
17 self._payload = payload
18
19 async def json(self, **kwargs: Any) -> Any:
20 return self._payload
21
22
23class _FakeGetContext:
24 def __init__(self, session: _FakeSession) -> None:
25 self._session = session
26
27 async def __aenter__(self) -> _FakeResponse:
28 return _FakeResponse(self._session.payload)
29
30 async def __aexit__(self, *exc_info: object) -> bool:
31 return False
32
33
34class _FakeSession:
35 def __init__(self, payload: Any) -> None:
36 self.payload = payload
37 self.calls = 0
38
39 def get(self, url: str, **kwargs: Any) -> _FakeGetContext:
40 self.calls += 1
41 return _FakeGetContext(self)
42
43
44def _mass_episode() -> PodcastEpisode:
45 """Build a resolved episode with no chapters, as the enrichment input."""
46 episode = parse_podcast_episode(
47 episode={
48 "title": "Episode 1",
49 "enclosures": [{"url": "https://example.com/ep1.mp3"}],
50 "guid": "guid-1",
51 },
52 prov_podcast_id="pod-1",
53 position=1,
54 instance_id="gpodder--test",
55 domain="gpodder",
56 )
57 assert episode is not None
58 return episode
59
60
61def _provider(podcast: dict[str, Any], session: _FakeSession) -> MagicMock:
62 """Build a gpodder provider stub sufficient for _enrich_episode_chapters."""
63 provider = MagicMock()
64 provider._cache_get_podcast = AsyncMock(return_value=podcast)
65 provider.mass.http_session = session
66 return provider
67
68
69async def _enrich(provider: MagicMock, guid_or_stream_url: str, episode: PodcastEpisode) -> None:
70 await GPodder._enrich_episode_chapters(
71 cast("GPodder", provider), "pod-1", guid_or_stream_url, episode
72 )
73
74
75async def test_enriches_matching_episode_and_skips_enclosure_less() -> None:
76 """An enclosure-less raw episode is skipped; the matching one's chapters are fetched."""
77 podcast = {
78 "episodes": [
79 # enclosure-less: get_stream_url_and_guid_from_episode raises ValueError -> skipped
80 {"enclosures": []},
81 {
82 "enclosures": [{"url": "https://example.com/ep1.mp3"}],
83 "guid": "guid-1",
84 "chapters_json_url": "https://example.com/ch.json",
85 },
86 ]
87 }
88 session = _FakeSession(payload={"chapters": [{"startTime": 0, "title": "Intro"}]})
89 mass_episode = _mass_episode()
90 await _enrich(_provider(podcast, session), "guid-1", mass_episode)
91 assert session.calls == 1
92 assert mass_episode.metadata.chapters is not None
93 assert [c.name for c in mass_episode.metadata.chapters] == ["Intro"]
94
95
96async def test_no_matching_episode_performs_no_fetch() -> None:
97 """When no raw episode matches the id, nothing is fetched and chapters stay None."""
98 podcast = {
99 "episodes": [
100 {"enclosures": [{"url": "https://example.com/other.mp3"}], "guid": "other"},
101 ]
102 }
103 session = _FakeSession(payload={"chapters": [{"startTime": 0, "title": "Intro"}]})
104 mass_episode = _mass_episode()
105 await _enrich(_provider(podcast, session), "guid-1", mass_episode)
106 assert session.calls == 0
107 assert mass_episode.metadata.chapters is None
108