music-assistant-server

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