/
/
/
1"""Tests for Podcast Index episode metadata (persons/links) and chapter enrichment."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any, cast
6from unittest.mock import AsyncMock, MagicMock
7
8from music_assistant_models.enums import LinkType
9
10from music_assistant.providers.podcast_index.helpers import parse_episode_from_data
11from music_assistant.providers.podcast_index.provider import PodcastIndexProvider
12
13if TYPE_CHECKING:
14 from music_assistant_models.media_items import PodcastEpisode
15
16
17def _episode_data(**overrides: Any) -> dict[str, Any]:
18 """Return minimal Podcast Index episode API data with a valid enclosure."""
19 data: dict[str, Any] = {
20 "id": 123,
21 "title": "Episode 1",
22 "enclosureUrl": "https://example.com/ep1.mp3",
23 "enclosureType": "audio/mpeg",
24 }
25 data.update(overrides)
26 return data
27
28
29def _parse(data: dict[str, Any]) -> PodcastEpisode | None:
30 return parse_episode_from_data(data, "feed-1", "podcast_index--test", "podcast_index")
31
32
33# --- parse_episode_from_data: persons / links ------------------------------------------------
34
35
36def test_persons_map_to_performers() -> None:
37 """The API persons array becomes performer names on the episode metadata."""
38 episode = _parse(
39 _episode_data(persons=[{"name": "Jane Host", "role": "host"}, {"name": "Joe Guest"}])
40 )
41 assert episode is not None
42 assert episode.metadata.performers == {"Jane Host", "Joe Guest"}
43
44
45def test_link_maps_to_website_link() -> None:
46 """The API episode link becomes a WEBSITE link."""
47 episode = _parse(_episode_data(link="https://example.com/ep1"))
48 assert episode is not None
49 assert episode.metadata.links is not None
50 link = next(iter(episode.metadata.links))
51 assert (link.type, link.url) == (LinkType.WEBSITE, "https://example.com/ep1")
52
53
54def test_missing_persons_and_link_leave_metadata_unset() -> None:
55 """Without persons/link, performers and links stay None and parsing still succeeds."""
56 episode = _parse(_episode_data())
57 assert episode is not None
58 assert episode.metadata.performers is None
59 assert episode.metadata.links is None
60
61
62# --- chapter enrichment on the single-episode path -------------------------------------------
63
64
65class _FakeResponse:
66 def __init__(self, payload: Any) -> None:
67 self._payload = payload
68
69 async def json(self, **kwargs: Any) -> Any:
70 return self._payload
71
72
73class _FakeGetContext:
74 def __init__(self, session: _FakeSession) -> None:
75 self._session = session
76
77 async def __aenter__(self) -> _FakeResponse:
78 return _FakeResponse(self._session.payload)
79
80 async def __aexit__(self, *exc_info: object) -> bool:
81 return False
82
83
84class _FakeSession:
85 def __init__(self, payload: Any) -> None:
86 self.payload = payload
87 self.calls = 0
88
89 def get(self, url: str, **kwargs: Any) -> _FakeGetContext:
90 self.calls += 1
91 return _FakeGetContext(self)
92
93
94def _provider(episode_data: dict[str, Any], session: _FakeSession) -> MagicMock:
95 """Build a provider stub sufficient for get_podcast_episode's single lookup."""
96 provider = MagicMock()
97 provider.instance_id = "podcast_index--test"
98 provider.domain = "podcast_index"
99 provider.logger = MagicMock()
100 provider.mass.http_session = session
101 provider._api_request = AsyncMock(return_value={"episode": episode_data})
102 return provider
103
104
105async def _call_get_episode(provider: MagicMock, prov_episode_id: str) -> PodcastEpisode:
106 # bypass the @use_cache wrapper to drive the real method directly
107 func: Any = PodcastIndexProvider.get_podcast_episode.__wrapped__ # type: ignore[attr-defined]
108 result = await func(cast("PodcastIndexProvider", provider), prov_episode_id)
109 return cast("PodcastEpisode", result)
110
111
112async def test_get_podcast_episode_enriches_chapters() -> None:
113 """A chaptersUrl on the single-episode path populates metadata.chapters."""
114 session = _FakeSession(payload={"chapters": [{"startTime": 0, "title": "Intro"}]})
115 provider = _provider(_episode_data(chaptersUrl="https://example.com/ch.json"), session)
116 episode = await _call_get_episode(provider, "feed-1|123")
117 assert session.calls == 1
118 assert episode.metadata.chapters is not None
119 assert [c.name for c in episode.metadata.chapters] == ["Intro"]
120
121
122async def test_get_podcast_episode_without_chapters_url() -> None:
123 """No chaptersUrl: the episode resolves normally with no chapters and no fetch."""
124 session = _FakeSession(payload={"chapters": [{"startTime": 0, "title": "Intro"}]})
125 provider = _provider(_episode_data(), session)
126 episode = await _call_get_episode(provider, "feed-1|123")
127 assert session.calls == 0
128 assert episode.metadata.chapters is None
129