/
/
/
1"""Test the iTunes Podcasts two-method recommendations contract."""
2
3from unittest.mock import AsyncMock, Mock, patch
4
5import pytest
6
7from music_assistant.providers.itunes_podcasts import (
8 RECOMMENDATION_ROW_TOP_PODCASTS,
9 SUPPORTED_FEATURES,
10 ITunesPodcastsProvider,
11)
12from music_assistant.providers.itunes_podcasts.schema import PodcastSearchResult, TopPodcastsHelper
13
14
15@pytest.fixture
16def mass_mock() -> Mock:
17 """Return a mock MusicAssistant instance."""
18 mass = Mock()
19 mass.http_session = AsyncMock()
20 mass.cache.get = AsyncMock(return_value=None)
21 mass.cache.set = AsyncMock()
22 return mass
23
24
25@pytest.fixture
26def manifest_mock() -> Mock:
27 """Return a mock provider manifest."""
28 manifest = Mock()
29 manifest.domain = "itunes_podcasts"
30 return manifest
31
32
33@pytest.fixture
34def config_mock() -> Mock:
35 """Return a mock provider config."""
36 config = Mock()
37 config.name = "iTunes Podcasts Test"
38 config.instance_id = "itunes_podcasts_test"
39 config.enabled = True
40 config.get_value.side_effect = lambda key, default=None: {
41 "locale": "us",
42 "explicit": True,
43 "num_episodes": 0,
44 "log_level": "INFO",
45 }.get(key, default)
46 return config
47
48
49@pytest.fixture
50async def provider(
51 mass_mock: Mock, manifest_mock: Mock, config_mock: Mock
52) -> ITunesPodcastsProvider:
53 """Return an ITunesPodcastsProvider instance."""
54 provider = ITunesPodcastsProvider(mass_mock, manifest_mock, config_mock, SUPPORTED_FEATURES)
55 await provider.handle_async_init()
56 return provider
57
58
59def _search_result() -> PodcastSearchResult:
60 """Return a minimal top-podcast search result."""
61 return PodcastSearchResult(
62 track_name="Test Podcast",
63 artist_name="Test Publisher",
64 feed_url="https://example.com/feed.xml",
65 artwork_url_600="https://example.com/artwork600.jpg",
66 )
67
68
69async def test_get_recommendations_static_row_without_backend_calls(
70 provider: ITunesPodcastsProvider, mass_mock: Mock
71) -> None:
72 """get_recommendations() returns the single static row without any backend or cache I/O."""
73 result = await provider.get_recommendations()
74
75 assert [folder.item_id for folder in result] == [RECOMMENDATION_ROW_TOP_PODCASTS]
76 folder = result[0]
77 assert folder.name == "Trending Podcasts"
78 assert folder.translation_key == "trending_podcasts"
79 assert folder.icon == "mdi-trending-up"
80 assert folder.provider == "itunes_podcasts_test"
81 assert len(folder.items) == 0
82 mass_mock.http_session.get.assert_not_called()
83 mass_mock.cache.get.assert_not_called()
84
85
86async def test_get_recommendation_items_fetches_top_podcasts(
87 provider: ITunesPodcastsProvider,
88) -> None:
89 """get_recommendation_items(<row>) triggers the top-podcasts fetch and returns its podcasts."""
90 with patch.object(
91 provider, "_cache_get_top_podcasts", new_callable=AsyncMock
92 ) as mock_top_podcasts:
93 mock_top_podcasts.return_value = [_search_result()]
94
95 items = await provider.get_recommendation_items(RECOMMENDATION_ROW_TOP_PODCASTS)
96
97 mock_top_podcasts.assert_awaited_once_with()
98 assert [item.item_id for item in items] == ["https://example.com/feed.xml"]
99 assert items[0].name == "Test Podcast"
100
101
102async def test_get_recommendation_items_served_from_cache(
103 provider: ITunesPodcastsProvider, mass_mock: Mock
104) -> None:
105 """A cached top-podcasts payload serves the row's items without any http calls."""
106 helper = TopPodcastsHelper(top_podcasts=[_search_result()])
107 mass_mock.cache.get = AsyncMock(return_value=helper.to_dict())
108
109 items = await provider.get_recommendation_items(RECOMMENDATION_ROW_TOP_PODCASTS)
110
111 assert [item.item_id for item in items] == ["https://example.com/feed.xml"]
112 mass_mock.http_session.get.assert_not_called()
113
114
115async def test_get_recommendation_items_unknown_id_returns_empty(
116 provider: ITunesPodcastsProvider, mass_mock: Mock
117) -> None:
118 """An unknown row item_id returns an empty result without triggering any fetch."""
119 with patch.object(
120 provider, "_cache_get_top_podcasts", new_callable=AsyncMock
121 ) as mock_top_podcasts:
122 items = await provider.get_recommendation_items("unknown-row")
123
124 assert len(items) == 0
125 mock_top_podcasts.assert_not_called()
126 mass_mock.http_session.get.assert_not_called()
127