/
/
/
1"""Tests for the podcastfeed -> Mass parsing helpers."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any, cast
6
7from aiohttp.client import ClientError
8from music_assistant_models.enums import LinkType
9
10from music_assistant.helpers.podcast_parsers import (
11 enrich_episode_chapters,
12 find_episode_stream_url,
13 get_cached_podcast,
14 get_episode_positions,
15 get_podcastparser_dict,
16 get_stream_url_from_episode,
17 parse_chapters_from_json,
18 parse_podcast_episode,
19 parse_podcast_persons,
20 rank_episodes_by_date,
21 refresh_cached_podcast,
22)
23
24if TYPE_CHECKING:
25 import aiohttp
26 from music_assistant_models.media_items import PodcastEpisode
27
28 from music_assistant.mass import MusicAssistant
29
30
31def _episode(**overrides: Any) -> dict[str, Any]:
32 """Return a minimal podcastparser episode dict with a valid enclosure."""
33 episode: dict[str, Any] = {
34 "title": "Episode 1",
35 "enclosures": [{"url": "https://example.com/ep1.mp3"}],
36 }
37 episode.update(overrides)
38 return episode
39
40
41def _parse(episode: dict[str, Any]) -> PodcastEpisode | None:
42 """Parse an episode dict with the required boilerplate args."""
43 return parse_podcast_episode(
44 episode=episode,
45 prov_podcast_id="podcast-1",
46 position=1,
47 instance_id="podcastfeed--test",
48 domain="podcastfeed",
49 )
50
51
52class _FakeResponse:
53 """Minimal stand-in for an aiohttp response yielding a fixed JSON payload."""
54
55 def __init__(self, payload: Any) -> None:
56 self._payload = payload
57
58 async def json(self, **kwargs: Any) -> Any:
59 return self._payload
60
61
62class _FakeGetContext:
63 """
64 Stand-in for aiohttp's request context manager.
65
66 Mirrors aiohttp where the error (e.g. raise_for_status / timeout) surfaces on
67 entering the context, so enrichment must guard the ``async with`` itself.
68 """
69
70 def __init__(self, session: _FakeSession) -> None:
71 self._session = session
72
73 async def __aenter__(self) -> _FakeResponse:
74 if self._session._error is not None:
75 raise self._session._error
76 return _FakeResponse(self._session._payload)
77
78 async def __aexit__(self, *exc_info: object) -> bool:
79 return False
80
81
82class _FakeSession:
83 """Minimal stand-in for an aiohttp ClientSession used by chapter enrichment."""
84
85 def __init__(self, *, payload: Any = None, error: Exception | None = None) -> None:
86 self._payload = payload
87 self._error = error
88 self.calls = 0
89 self.last_url: str | None = None
90
91 def get(self, url: str, **kwargs: Any) -> _FakeGetContext:
92 self.calls += 1
93 self.last_url = url
94 return _FakeGetContext(self)
95
96
97# --- enclosure / stream url selection ---------------------------------------------------------
98
99
100def test_stream_url_prefers_audio_over_leading_image() -> None:
101 """A leading image media:content enclosure is skipped in favor of the audio one (#5920)."""
102 audio_url = "https://rss.wbur.org/the-midnight-rebellion/ep1.mp3"
103 cover_url = "https://rss.wbur.org/the-midnight-rebellion/cover.jpg"
104 episode = _episode(
105 enclosures=[
106 {"url": cover_url, "mime_type": "image/jpeg"},
107 {"url": audio_url, "mime_type": "audio/mpeg"},
108 {"url": audio_url, "mime_type": "audio/mpeg"},
109 ]
110 )
111 assert get_stream_url_from_episode(episode=episode) == audio_url
112
113
114def test_stream_url_accepts_bogus_mime_type() -> None:
115 """A real audio enclosure with a bogus declared mime type is still used (#5692)."""
116 episode = _episode(
117 enclosures=[{"url": "https://example.com/ep1.mp3", "mime_type": "application/octet-stream"}]
118 )
119 assert get_stream_url_from_episode(episode=episode) == "https://example.com/ep1.mp3"
120
121
122def test_stream_url_accepts_video_enclosure() -> None:
123 """A video/mp4 enclosure is accepted as a playable stream."""
124 video_url = "https://example.com/ep1.mp4"
125 episode = _episode(enclosures=[{"url": video_url, "mime_type": "video/mp4"}])
126 assert get_stream_url_from_episode(episode=episode) == video_url
127
128
129def test_stream_url_skips_enclosures_without_url() -> None:
130 """An enclosure missing a url is skipped in favor of a later usable one."""
131 episode = _episode(
132 enclosures=[
133 {"mime_type": "audio/mpeg"},
134 {"url": "https://example.com/ep1.mp3", "mime_type": "audio/mpeg"},
135 ]
136 )
137 assert get_stream_url_from_episode(episode=episode) == "https://example.com/ep1.mp3"
138
139
140def test_stream_url_image_only_returns_none() -> None:
141 """An episode whose only enclosure is an image has no playable stream."""
142 episode = _episode(
143 enclosures=[{"url": "https://example.com/cover.jpg", "mime_type": "image/jpeg"}]
144 )
145 assert get_stream_url_from_episode(episode=episode) is None
146 assert _parse(episode) is None
147
148
149def test_stream_url_first_audio_enclosure_wins() -> None:
150 """When several audio enclosures are declared, the first one wins."""
151 episode = _episode(
152 enclosures=[
153 {"url": "https://example.com/ep1-first.mp3", "mime_type": "audio/mpeg"},
154 {"url": "https://example.com/ep1-second.mp3", "mime_type": "audio/mpeg"},
155 ]
156 )
157 assert get_stream_url_from_episode(episode=episode) == "https://example.com/ep1-first.mp3"
158
159
160def test_stream_url_missing_mime_type_falls_back_to_url() -> None:
161 """The default `_episode()` enclosure has no mime_type key and still resolves (fallback)."""
162 assert get_stream_url_from_episode(episode=_episode()) == "https://example.com/ep1.mp3"
163
164
165# --- description -----------------------------------------------------------------------------
166
167
168def test_description_is_populated() -> None:
169 """A non-empty episode description ends up on the episode metadata."""
170 mass_episode = _parse(_episode(description="All about parsing podcasts."))
171 assert mass_episode is not None
172 assert mass_episode.metadata.description == "All about parsing podcasts."
173
174
175def test_empty_description_left_unset() -> None:
176 """An empty description (podcastparser's default) leaves metadata.description as None."""
177 mass_episode = _parse(_episode(description=""))
178 assert mass_episode is not None
179 assert mass_episode.metadata.description is None
180
181
182def test_missing_description_left_unset() -> None:
183 """An absent description key leaves metadata.description as None."""
184 mass_episode = _parse(_episode())
185 assert mass_episode is not None
186 assert mass_episode.metadata.description is None
187
188
189# --- parent podcast reference ----------------------------------------------------------------
190
191
192def test_podcast_reference_uses_podcast_name() -> None:
193 """The parent-podcast reference is named after the podcast, not the episode."""
194 mass_episode = parse_podcast_episode(
195 episode=_episode(title="Episode 1"),
196 prov_podcast_id="podcast-1",
197 position=1,
198 podcast_name="My Show",
199 instance_id="podcastfeed--test",
200 domain="podcastfeed",
201 )
202 assert mass_episode is not None
203 assert mass_episode.podcast.name == "My Show"
204 assert mass_episode.name == "Episode 1"
205
206
207def test_podcast_reference_falls_back_to_episode_title() -> None:
208 """Without a podcast name, the reference falls back to the episode title."""
209 mass_episode = _parse(_episode(title="Some Episode"))
210 assert mass_episode is not None
211 assert mass_episode.podcast.name == "Some Episode"
212
213
214# --- episode position -----------------------------------------------------------------------
215
216
217def test_episode_position_uses_caller_supplied_position() -> None:
218 """The episode position is determined by the caller."""
219 mass_episode = parse_podcast_episode(
220 episode=_episode(),
221 prov_podcast_id="podcast-1",
222 position=5,
223 instance_id="podcastfeed--test",
224 domain="podcastfeed",
225 )
226 assert mass_episode is not None
227 assert mass_episode.position == 5
228
229
230def test_positions_use_episode_numbers_when_all_numbered() -> None:
231 """A fully numbered feed keeps its own itunes:episode numbers."""
232 episodes = [_episode(number=3), _episode(number=1), _episode(number=2)]
233 assert get_episode_positions(episodes) == [3, 1, 2]
234
235
236def test_positions_ignore_partial_numbering() -> None:
237 """A feed that numbers only some episodes has all of its numbers ignored."""
238 episodes = [
239 _episode(number=1, published=300),
240 _episode(published=100),
241 _episode(number=3, published=200),
242 ]
243 assert get_episode_positions(episodes) == [3, 1, 2]
244
245
246def test_positions_rank_newest_highest_by_published() -> None:
247 """Dated feeds are ranked oldest to newest regardless of the order they are listed in."""
248 episodes = [_episode(published=300), _episode(published=100), _episode(published=200)]
249 assert get_episode_positions(episodes) == [3, 1, 2]
250
251
252def test_positions_rank_serial_feeds_oldest_first() -> None:
253 """An oldest-first (itunes:type=serial) feed still gives the newest episode the top spot."""
254 episodes = [_episode(published=100), _episode(published=200), _episode(published=300)]
255 assert get_episode_positions(episodes) == [1, 2, 3]
256
257
258def test_positions_ignore_numbering_that_restarts_each_season() -> None:
259 """Seasoned feeds restart their numbering, so the dates decide instead."""
260 episodes = [
261 _episode(number=1, season=2, published=300),
262 _episode(number=2, season=1, published=100),
263 _episode(number=1, season=1, published=200),
264 ]
265 assert get_episode_positions(episodes) == [3, 1, 2]
266
267
268def test_positions_use_numbers_for_a_single_season() -> None:
269 """A feed that declares one season keeps its own episode numbers."""
270 episodes = [_episode(number=2, season=1), _episode(number=1, season=1)]
271 assert get_episode_positions(episodes) == [2, 1]
272
273
274def test_positions_prefer_numbers_over_dates() -> None:
275 """Episode numbers win over the publication dates when the feed carries both."""
276 episodes = [_episode(number=7, published=100), _episode(number=9, published=200)]
277 assert get_episode_positions(episodes) == [7, 9]
278
279
280def test_rank_by_date_keeps_feed_order_among_undated() -> None:
281 """Undated episodes rank oldest and hold their listing order between themselves."""
282 assert rank_episodes_by_date([None, 300, None, 100]) == [1, 4, 2, 3]
283
284
285def test_rank_by_date_accepts_string_dates() -> None:
286 """Providers reporting dates as sortable strings are ranked the same way."""
287 assert rank_episodes_by_date(["2026-03-01", "2024-01-01", "2025-02-01"]) == [3, 1, 2]
288
289
290def test_positions_rank_undated_episodes_as_oldest() -> None:
291 """Episodes missing a publication date rank below every dated one."""
292 episodes = [_episode(published=200), _episode(), _episode(published=100)]
293 assert get_episode_positions(episodes) == [3, 1, 2]
294
295
296def test_positions_assume_newest_first_without_dates() -> None:
297 """Without publication dates the feed order is assumed to be newest-first."""
298 episodes = [_episode(), _episode(), _episode()]
299 assert get_episode_positions(episodes) == [3, 2, 1]
300
301
302def test_positions_empty_feed() -> None:
303 """An empty feed yields no positions."""
304 assert get_episode_positions([]) == []
305
306
307# --- inline (Podlove Simple Chapters) --------------------------------------------------------
308
309
310def test_inline_chapters_populated() -> None:
311 """Inline chapters are attached, including a chapter that starts at 0."""
312 episode = _episode(
313 chapters=[
314 {"title": "Intro", "start": 0},
315 {"title": "Topic", "start": 90.0},
316 ]
317 )
318 mass_episode = _parse(episode)
319 assert mass_episode is not None
320 chapters = mass_episode.metadata.chapters
321 assert chapters is not None
322 assert [c.name for c in chapters] == ["Intro", "Topic"]
323 # the opening chapter (start == 0) must survive; a truthiness check would drop it
324 assert chapters[0].start == 0
325
326
327def test_inline_chapters_absent_leaves_none() -> None:
328 """No inline chapters key leaves metadata.chapters as None."""
329 mass_episode = _parse(_episode())
330 assert mass_episode is not None
331 assert mass_episode.metadata.chapters is None
332
333
334# --- podcast:chapters JSON parsing -----------------------------------------------------------
335
336
337def test_parse_chapters_from_json_maps_fields() -> None:
338 """Valid chapters JSON maps startTime/endTime/title onto MediaItemChapters."""
339 data = {
340 "version": "1.2.0",
341 "chapters": [
342 {"startTime": 0, "title": "Intro"},
343 {"startTime": 90, "endTime": 120, "title": "Sponsor"},
344 ],
345 }
346 chapters = parse_chapters_from_json(data)
347 assert [(c.position, c.name, c.start, c.end) for c in chapters] == [
348 (1, "Intro", 0.0, None),
349 (2, "Sponsor", 90.0, 120.0),
350 ]
351
352
353def test_parse_chapters_from_json_skips_invalid_and_hidden() -> None:
354 """Entries missing startTime/title, hidden (toc: false), or non-dict are skipped."""
355 data = {
356 "chapters": [
357 {"title": "no start"},
358 {"startTime": 10},
359 {"startTime": 20, "title": "hidden", "toc": False},
360 "not-a-dict",
361 {"startTime": 30, "title": "kept"},
362 ]
363 }
364 chapters = parse_chapters_from_json(data)
365 assert [c.name for c in chapters] == ["kept"]
366 assert chapters[0].position == 1
367
368
369def test_parse_chapters_from_json_malformed_returns_empty() -> None:
370 """A missing or non-list chapters key yields no chapters."""
371 assert parse_chapters_from_json({}) == []
372 assert parse_chapters_from_json({"chapters": "nope"}) == []
373
374
375def test_parse_chapters_from_json_skips_malformed_values() -> None:
376 """Non-numeric/non-finite startTimes and non-string titles are dropped, not stored."""
377 data = {
378 "chapters": [
379 {"startTime": "01:30", "title": "non-numeric start"},
380 {"startTime": float("nan"), "title": "nan start"},
381 {"startTime": float("inf"), "title": "inf start"},
382 {"startTime": 10, "title": {"x": 1}},
383 {"startTime": 20, "title": 123},
384 {"startTime": 30, "title": "kept", "endTime": float("nan")},
385 ]
386 }
387 chapters = parse_chapters_from_json(data)
388 # only the well-formed entry survives, and its non-finite endTime collapses to None
389 assert [(c.name, c.start, c.end) for c in chapters] == [("kept", 30.0, None)]
390
391
392# --- chapter enrichment (external JSON fetch) ------------------------------------------------
393
394
395async def test_enrich_fetches_json_chapters() -> None:
396 """Chapters are fetched from chapters_json_url when the episode has none inline."""
397 mass_episode = _parse(_episode())
398 assert mass_episode is not None
399 session = _FakeSession(payload={"chapters": [{"startTime": 0, "title": "Intro"}]})
400 await enrich_episode_chapters(
401 session=cast("aiohttp.ClientSession", session),
402 chapters_json_url="https://example.com/ch.json",
403 mass_episode=mass_episode,
404 )
405 assert mass_episode.metadata.chapters is not None
406 assert [c.name for c in mass_episode.metadata.chapters] == ["Intro"]
407
408
409async def test_enrich_swallows_fetch_error() -> None:
410 """A failed chapter fetch is ignored and leaves the episode without chapters."""
411 mass_episode = _parse(_episode())
412 assert mass_episode is not None
413 session = _FakeSession(error=ClientError("boom"))
414 await enrich_episode_chapters(
415 session=cast("aiohttp.ClientSession", session),
416 chapters_json_url="https://example.com/ch.json",
417 mass_episode=mass_episode,
418 )
419 assert mass_episode.metadata.chapters is None
420
421
422async def test_enrich_swallows_timeout() -> None:
423 """A timed-out chapter fetch is ignored: TimeoutError is not a ClientError."""
424 mass_episode = _parse(_episode())
425 assert mass_episode is not None
426 session = _FakeSession(error=TimeoutError())
427 await enrich_episode_chapters(
428 session=cast("aiohttp.ClientSession", session),
429 chapters_json_url="https://example.com/ch.json",
430 mass_episode=mass_episode,
431 )
432 assert mass_episode.metadata.chapters is None
433
434
435async def test_enrich_skips_when_inline_chapters_present() -> None:
436 """Inline chapters win: no JSON fetch is performed when chapters already exist."""
437 mass_episode = _parse(_episode(chapters=[{"title": "Inline", "start": 0}]))
438 assert mass_episode is not None
439 session = _FakeSession(payload={"chapters": [{"startTime": 5, "title": "FromJson"}]})
440 await enrich_episode_chapters(
441 session=cast("aiohttp.ClientSession", session),
442 chapters_json_url="https://example.com/ch.json",
443 mass_episode=mass_episode,
444 )
445 assert session.calls == 0
446 assert mass_episode.metadata.chapters is not None
447 assert [c.name for c in mass_episode.metadata.chapters] == ["Inline"]
448
449
450async def test_enrich_skips_when_no_url() -> None:
451 """A falsy chapters URL is a no-op: no fetch, no chapters."""
452 mass_episode = _parse(_episode())
453 assert mass_episode is not None
454 session = _FakeSession(payload={"chapters": [{"startTime": 0, "title": "Intro"}]})
455 await enrich_episode_chapters(
456 session=cast("aiohttp.ClientSession", session),
457 chapters_json_url=None,
458 mass_episode=mass_episode,
459 )
460 assert session.calls == 0
461 assert mass_episode.metadata.chapters is None
462
463
464async def test_enrich_ignores_non_dict_payload() -> None:
465 """A JSON payload that is not an object (e.g. a list) leaves the episode without chapters."""
466 mass_episode = _parse(_episode())
467 assert mass_episode is not None
468 session = _FakeSession(payload=[{"startTime": 0, "title": "Intro"}])
469 await enrich_episode_chapters(
470 session=cast("aiohttp.ClientSession", session),
471 chapters_json_url="https://example.com/ch.json",
472 mass_episode=mass_episode,
473 )
474 assert session.calls == 1
475 assert mass_episode.metadata.chapters is None
476
477
478async def test_enrich_ignores_empty_chapters_payload() -> None:
479 """A document whose chapters all filter out leaves metadata.chapters as None."""
480 mass_episode = _parse(_episode())
481 assert mass_episode is not None
482 session = _FakeSession(payload={"chapters": []})
483 await enrich_episode_chapters(
484 session=cast("aiohttp.ClientSession", session),
485 chapters_json_url="https://example.com/ch.json",
486 mass_episode=mass_episode,
487 )
488 assert session.calls == 1
489 assert mass_episode.metadata.chapters is None
490
491
492# --- episode webpage, explicit flag, performers ----------------------------------------------
493
494
495def test_episode_link_maps_to_website_link() -> None:
496 """An item <link> becomes a WEBSITE link on the episode metadata."""
497 mass_episode = _parse(_episode(link="https://example.com/ep1"))
498 assert mass_episode is not None
499 assert mass_episode.metadata.links is not None
500 link = next(iter(mass_episode.metadata.links))
501 assert (link.type, link.url) == (LinkType.WEBSITE, "https://example.com/ep1")
502
503
504def test_episode_missing_link_leaves_links_unset() -> None:
505 """Without an item <link>, metadata.links stays None."""
506 mass_episode = _parse(_episode())
507 assert mass_episode is not None
508 assert mass_episode.metadata.links is None
509
510
511def test_episode_explicit_only_set_when_declared() -> None:
512 """The explicit flag is set from the feed value, and left untouched when absent."""
513 explicit_ep = _parse(_episode(explicit=True))
514 assert explicit_ep is not None
515 assert explicit_ep.metadata.explicit is True
516 not_explicit = _parse(_episode(explicit=False))
517 assert not_explicit is not None
518 assert not_explicit.metadata.explicit is False
519 # absent key must not coerce to False over the model default
520 unknown = _parse(_episode())
521 assert unknown is not None
522 assert unknown.metadata.explicit is None
523
524
525def test_episode_performers_from_persons() -> None:
526 """podcast:person entries land on metadata.performers as names."""
527 mass_episode = _parse(
528 _episode(persons=[{"name": "Jane Host", "role": "host"}, {"name": "Joe Guest"}])
529 )
530 assert mass_episode is not None
531 assert mass_episode.metadata.performers == {"Jane Host", "Joe Guest"}
532
533
534# --- parse_podcast_persons -------------------------------------------------------------------
535
536
537def test_parse_podcast_persons_dedupes_and_trims() -> None:
538 """Names are trimmed, de-duplicated case-insensitively, and order-preserved."""
539 persons = [{"name": " Alice "}, {"name": "alice"}, {"name": "Bob"}]
540 assert parse_podcast_persons(persons) == ["Alice", "Bob"]
541
542
543def test_parse_podcast_persons_skips_blank_and_malformed() -> None:
544 """Blank names, missing names, and non-dict/non-str entries are skipped."""
545 persons = [{"name": ""}, {"role": "host"}, {"name": 123}, "Carol", 7]
546 assert parse_podcast_persons(persons) == ["Carol"]
547
548
549def test_parse_podcast_persons_non_list_returns_empty() -> None:
550 """Any non-list input yields no names, so callers need not guard."""
551 assert parse_podcast_persons(None) == []
552 assert parse_podcast_persons("nope") == []
553
554
555# --- find_episode_stream_url -----------------------------------------------------------------
556
557
558def test_find_episode_stream_url_matches_guid() -> None:
559 """An episode with a usable guid is found by that guid."""
560 feed = {"episodes": [_episode(guid="ep-1"), _episode(guid="ep-2", enclosures=[{"url": "b"}])]}
561 assert find_episode_stream_url(parsed_feed=feed, guid_or_stream_url="ep-2") == "b"
562
563
564def test_find_episode_stream_url_falls_back_to_stream_url() -> None:
565 """A guid containing a space is unusable as an id, so the stream url identifies it."""
566 feed = {"episodes": [_episode(guid="not a guid")]}
567 assert (
568 find_episode_stream_url(parsed_feed=feed, guid_or_stream_url="https://example.com/ep1.mp3")
569 == "https://example.com/ep1.mp3"
570 )
571 # the unusable guid must not match
572 assert find_episode_stream_url(parsed_feed=feed, guid_or_stream_url="not a guid") is None
573
574
575def test_find_episode_stream_url_skips_episodes_without_enclosure() -> None:
576 """An episode without a playable enclosure does not stop the search."""
577 feed = {"episodes": [{"title": "no audio"}, _episode(guid="ep-2", enclosures=[{"url": "b"}])]}
578 assert find_episode_stream_url(parsed_feed=feed, guid_or_stream_url="ep-2") == "b"
579
580
581def test_find_episode_stream_url_unknown_returns_none() -> None:
582 """An unknown identifier yields None rather than raising."""
583 assert find_episode_stream_url(parsed_feed={"episodes": []}, guid_or_stream_url="x") is None
584
585
586# --- feed retrieval and caching ----------------------------------------------------------------
587
588
589FEED_URL = "https://example.com/feed.xml"
590FEED_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
591<rss version="2.0"><channel>
592<title>Feed One</title>
593<item>
594<title>Episode 1</title>
595<guid>ep-1</guid>
596<enclosure url="https://example.com/ep1.mp3" type="audio/mpeg" length="1"/>
597</item>
598</channel></rss>
599"""
600
601
602class _FakeFeedResponse:
603 """Minimal stand-in for an aiohttp response yielding raw feed bytes."""
604
605 def __init__(self, body: bytes) -> None:
606 self._body = body
607
608 async def read(self) -> bytes:
609 return self._body
610
611
612class _FakeFeedGetContext:
613 """Request context manager recording whether the response was released again."""
614
615 def __init__(self, session: _FakeFeedSession) -> None:
616 self._session = session
617
618 async def __aenter__(self) -> _FakeFeedResponse:
619 error = self._session.errors.pop(0) if self._session.errors else None
620 if error is not None:
621 raise error
622 return _FakeFeedResponse(self._session.body)
623
624 async def __aexit__(self, *exc_info: object) -> bool:
625 self._session.released += 1
626 return False
627
628
629class _FakeFeedSession:
630 """Session stand-in serving a fixed feed body, optionally failing the first attempts."""
631
632 def __init__(self, *, body: bytes = FEED_XML, errors: list[Exception] | None = None) -> None:
633 self.body = body
634 self.errors = errors or []
635 self.calls = 0
636 self.released = 0
637 self.headers: list[dict[str, str]] = []
638
639 def get(self, url: str, headers: dict[str, str], **kwargs: Any) -> _FakeFeedGetContext:
640 self.calls += 1
641 self.headers.append(headers)
642 return _FakeFeedGetContext(self)
643
644
645class _FakeCache:
646 """In-memory stand-in for the cache controller, keyed like the real one."""
647
648 def __init__(self) -> None:
649 self.store: dict[tuple[str, str, int], Any] = {}
650 self.sets = 0
651
652 async def get(self, key: str, provider: str, category: int, default: Any = None) -> Any:
653 return self.store.get((key, provider, category), default)
654
655 async def set(self, key: str, provider: str, category: int, data: Any, expiration: int) -> None:
656 self.sets += 1
657 self.store[(key, provider, category)] = data
658
659
660class _FakeMass:
661 """Stand-in exposing only what the podcast cache helpers use."""
662
663 def __init__(self, session: _FakeFeedSession) -> None:
664 self.http_session = session
665 self.cache = _FakeCache()
666
667
668def _fake_mass(session: _FakeFeedSession) -> MusicAssistant:
669 return cast("MusicAssistant", _FakeMass(session))
670
671
672async def test_get_podcastparser_dict_releases_the_response() -> None:
673 """The feed response is released again, on the retry path as well."""
674 session = _FakeFeedSession(errors=[ClientError("no user agent allowed")])
675 parsed_feed = await get_podcastparser_dict(
676 session=cast("aiohttp.ClientSession", session), feed_url=FEED_URL
677 )
678 assert parsed_feed["title"] == "Feed One"
679 # the first attempt failed on entering the context, so only the second one is released
680 assert session.calls == 2
681 assert session.released == 1
682 assert session.headers[0] == {"User-Agent": "Mozilla/5.0"}
683
684
685async def test_get_cached_podcast_stores_and_reuses_the_feed() -> None:
686 """A miss retrieves and caches the feed, a subsequent call is served from the cache."""
687 session = _FakeFeedSession()
688 mass = _fake_mass(session)
689 parsed_feed = await get_cached_podcast(
690 mass=mass, provider_instance_id="podcastfeed--test", feed_url=FEED_URL
691 )
692 assert parsed_feed["title"] == "Feed One"
693 assert session.calls == 1
694 await get_cached_podcast(mass=mass, provider_instance_id="podcastfeed--test", feed_url=FEED_URL)
695 assert session.calls == 1
696
697
698async def test_refresh_cached_podcast_always_updates_the_cache() -> None:
699 """A sync must refresh the cached feed, also when a valid entry exists."""
700 session = _FakeFeedSession()
701 mass = _fake_mass(session)
702 await get_cached_podcast(mass=mass, provider_instance_id="podcastfeed--test", feed_url=FEED_URL)
703 assert session.calls == 1
704 session.body = FEED_XML.replace(b"Feed One", b"Feed Renamed")
705 parsed_feed = await refresh_cached_podcast(
706 mass=mass, provider_instance_id="podcastfeed--test", feed_url=FEED_URL
707 )
708 assert session.calls == 2
709 assert parsed_feed["title"] == "Feed Renamed"
710 # the refreshed feed is what subsequent (cached) reads see
711 cached_feed = await get_cached_podcast(
712 mass=mass, provider_instance_id="podcastfeed--test", feed_url=FEED_URL
713 )
714 assert cached_feed["title"] == "Feed Renamed"
715 assert session.calls == 2
716
717
718def test_find_episode_stream_url_matches_empty_guid() -> None:
719 """An empty guid is used as episode id by the parser, so it must resolve as one."""
720 feed = {"episodes": [_episode(guid=""), _episode(guid="ep-2", enclosures=[{"url": "b"}])]}
721 assert find_episode_stream_url(parsed_feed=feed, guid_or_stream_url="") == (
722 "https://example.com/ep1.mp3"
723 )
724