/
/
/
1"""Test Tidal recommendations: payload manager + two-method provider contract."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, Mock, patch
8
9import pytest
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import ResourceTemporarilyUnavailable
12from music_assistant_models.media_items import (
13 ItemMapping,
14 RecommendationFolder,
15 UniqueList,
16)
17
18from music_assistant.providers.tidal.provider import TidalProvider
19from music_assistant.providers.tidal.recommendations import TidalRecommendationManager
20
21INSTANCE_ID = "tidal_test"
22
23
24@pytest.fixture
25def recommendation_manager(provider_mock: Mock) -> TidalRecommendationManager:
26 """Return a TidalRecommendationManager instance."""
27 return TidalRecommendationManager(provider_mock)
28
29
30@pytest.fixture
31def mass_mock() -> Mock:
32 """Return a mock MusicAssistant with a dict-backed cache and real background tasks."""
33 mass = Mock()
34 mass.http_session = AsyncMock()
35 mass.metadata.locale = "en_US"
36
37 cache_store: dict[str, Any] = {}
38 background_tasks: list[asyncio.Future[Any]] = []
39
40 async def _cache_get(key: str, **_kwargs: Any) -> tuple[Any, bool, bool]:
41 if key in cache_store:
42 return cache_store[key], True, True
43 return None, False, False
44
45 async def _cache_set(key: str, data: Any, **_kwargs: Any) -> None:
46 cache_store[key] = data
47
48 def _create_task(target: Any, *_args: Any, **_kwargs: Any) -> asyncio.Future[Any]:
49 task: asyncio.Future[Any] = asyncio.ensure_future(target)
50 background_tasks.append(task)
51 return task
52
53 mass.cache.get = AsyncMock(return_value=None)
54 mass.cache.get_with_freshness = AsyncMock(side_effect=_cache_get)
55 mass.cache.set = AsyncMock(side_effect=_cache_set)
56 mass.create_task = Mock(side_effect=_create_task)
57 mass.background_tasks = background_tasks
58 return mass
59
60
61@pytest.fixture
62def provider(mass_mock: Mock) -> TidalProvider:
63 """Return a TidalProvider instance for provider-level tests."""
64 manifest = Mock()
65 manifest.domain = "tidal"
66
67 config = Mock()
68 config.name = "Tidal Test"
69 config.instance_id = INSTANCE_ID
70 config.enabled = True
71 config.get_value.side_effect = lambda key: "INFO" if "log" in key else None
72
73 return TidalProvider(mass_mock, manifest, config)
74
75
76def _make_payload() -> list[RecommendationFolder]:
77 """Build a two-folder payload, shaped like the manager's merged page modules."""
78 return [
79 RecommendationFolder(
80 item_id="12345_for_you",
81 provider=INSTANCE_ID,
82 name="For You",
83 translation_key="12345_for_you",
84 icon="mdi-playlist-music",
85 subtitle="From Home ⢠1 items",
86 items=UniqueList(
87 [
88 ItemMapping(
89 media_type=MediaType.PLAYLIST,
90 item_id="p1",
91 provider=INSTANCE_ID,
92 name="Daily Mix",
93 )
94 ]
95 ),
96 ),
97 RecommendationFolder(
98 item_id="12345_new_albums",
99 provider=INSTANCE_ID,
100 name="New Albums",
101 translation_key="12345_new_albums",
102 icon="mdi-album",
103 subtitle="From Explore New Music ⢠1 items",
104 items=UniqueList(
105 [
106 ItemMapping(
107 media_type=MediaType.ALBUM,
108 item_id="a1",
109 provider=INSTANCE_ID,
110 name="New Album",
111 )
112 ]
113 ),
114 ),
115 ]
116
117
118async def test_get_recommendations_returns_rows_without_items(provider: TidalProvider) -> None:
119 """get_recommendations returns the payload folders as rows, stripped of items."""
120 payload = _make_payload()
121 with patch.object(
122 provider.recommendations_manager, "get_recommendations", new_callable=AsyncMock
123 ) as mock_fetch:
124 mock_fetch.return_value = payload
125
126 rows = await provider.get_recommendations()
127
128 mock_fetch.assert_awaited_once()
129 assert [row.item_id for row in rows] == ["12345_for_you", "12345_new_albums"]
130 for row, folder in zip(rows, payload, strict=True):
131 assert row.name == folder.name
132 assert row.translation_key == folder.translation_key
133 assert row.icon == folder.icon
134 assert len(row.items) == 0
135
136
137async def test_rows_and_items_share_one_payload_fetch(
138 provider: TidalProvider, mass_mock: Mock
139) -> None:
140 """Rows and both items calls are served from a single cached payload fetch."""
141 payload = _make_payload()
142 with patch.object(
143 provider.recommendations_manager, "get_recommendations", new_callable=AsyncMock
144 ) as mock_fetch:
145 mock_fetch.return_value = payload
146
147 await provider.get_recommendations()
148 # let the background cache-store task complete before the next calls
149 await asyncio.gather(*mass_mock.background_tasks)
150 items_for_you = await provider.get_recommendation_items("12345_for_you")
151 items_new_albums = await provider.get_recommendation_items("12345_new_albums")
152
153 mock_fetch.assert_awaited_once()
154 assert items_for_you == payload[0].items
155 assert items_new_albums == payload[1].items
156
157
158async def test_get_recommendation_items_unknown_id_returns_empty(
159 provider: TidalProvider,
160) -> None:
161 """An unknown row item_id yields an empty UniqueList."""
162 with patch.object(
163 provider.recommendations_manager, "get_recommendations", new_callable=AsyncMock
164 ) as mock_fetch:
165 mock_fetch.return_value = _make_payload()
166
167 result = await provider.get_recommendation_items("bogus_row")
168
169 assert isinstance(result, UniqueList)
170 assert len(result) == 0
171
172
173@pytest.mark.usefixtures("provider_mock")
174async def test_manager_builds_payload_folders(
175 recommendation_manager: TidalRecommendationManager,
176) -> None:
177 """The manager merges page modules into folders with items and stable item_ids."""
178 # Mock get_page_content to return a mock parser
179 mock_parser = Mock()
180 mock_parser.modules = [{"title": "Test Module"}]
181 mock_parser.get_module_items.return_value = (
182 [Mock(item_id="rec_1", name="Recommendation 1")],
183 MediaType.PLAYLIST,
184 )
185
186 with patch.object(
187 recommendation_manager, "get_page_content", new_callable=AsyncMock
188 ) as mock_get_page:
189 mock_get_page.return_value = mock_parser
190
191 recommendations = await recommendation_manager.get_recommendations()
192
193 assert len(recommendations) == 1
194 assert recommendations[0].name == "Test Module"
195 assert recommendations[0].item_id == "12345_test_module"
196 assert len(recommendations[0].items) == 1
197
198 # Should fetch pages
199 assert mock_get_page.call_count >= 1
200
201
202async def test_manager_filters_out_video_modules(
203 recommendation_manager: TidalRecommendationManager,
204) -> None:
205 """Test video modules never surface, by VIDEO_LIST type or a video-mentioning title."""
206 parser = Mock()
207 parser.modules = [
208 {"title": "Video Playlists", "type": "PLAYLIST_LIST"}, # title-based drop
209 {"title": "New Videos", "type": "VIDEO_LIST"}, # both
210 {"title": "Clips", "type": "VIDEO_LIST"}, # type-based drop
211 {"title": "Playlists", "type": "PLAYLIST_LIST"}, # survives
212 ]
213 parser.get_module_items.return_value = (
214 [Mock(item_id="p1", name="P1")],
215 MediaType.PLAYLIST,
216 )
217
218 with patch.object(
219 recommendation_manager, "get_page_content", new_callable=AsyncMock
220 ) as mock_get_page:
221 mock_get_page.return_value = parser
222
223 recommendations = await recommendation_manager.get_recommendations()
224
225 names = {r.name for r in recommendations}
226 assert names == {"Playlists"}
227
228
229async def test_manager_strips_at_symbol_when_multiple_instances(
230 recommendation_manager: TidalRecommendationManager, provider_mock: Mock
231) -> None:
232 """Test that username is included and '@' is stripped when multiple instances exist."""
233 provider_mock.auth.user = Mock(profile_name="[email protected]", user_name=None)
234
235 provider_mock.mass.config.get_provider_configs = AsyncMock(
236 return_value=[
237 Mock(domain="tidal", instance_id="tidal_instance_1"),
238 Mock(domain="tidal", instance_id="tidal_instance_2"),
239 Mock(domain="other", instance_id="other_instance"),
240 ]
241 )
242
243 parser_with_module = Mock()
244 parser_with_module.modules = [{"title": "Test Module"}]
245 parser_with_module.get_module_items.return_value = (
246 [Mock(item_id="rec_1", name="Recommendation 1")],
247 MediaType.PLAYLIST,
248 )
249
250 parser_empty = Mock()
251 parser_empty.modules = []
252 parser_empty.get_module_items = Mock()
253
254 with patch.object(
255 recommendation_manager, "get_page_content", new_callable=AsyncMock
256 ) as mock_get_page:
257 # Only first page returns the module, remaining pages return no modules
258 mock_get_page.side_effect = [parser_with_module] + [parser_empty] * 4
259
260 recommendations = await recommendation_manager.get_recommendations()
261
262 assert len(recommendations) == 1
263 assert recommendations[0].name == "Test Module (john)"
264 assert "@" not in recommendations[0].name
265 assert len(recommendations[0].items) == 1
266
267
268async def test_get_page_content(
269 recommendation_manager: TidalRecommendationManager, provider_mock: Mock
270) -> None:
271 """Test get_page_content."""
272 with patch(
273 "music_assistant.providers.tidal.recommendations.TidalPageParser"
274 ) as mock_parser_cls:
275 # Configure from_cache to be async and return None
276 mock_parser_cls.from_cache = AsyncMock(return_value=None)
277
278 # Configure parser instance
279 mock_parser_instance = mock_parser_cls.return_value
280 mock_parser_instance.parse_page_structure = Mock() # Ensure it's a synchronous mock
281 mock_parser_instance.to_cache = Mock(
282 return_value={"module_map": [], "content_map": {}, "parsed_at": 1234567890}
283 )
284
285 # Mock API response
286 provider_mock.api.get.return_value = {"rows": []}
287
288 parser = await recommendation_manager.get_page_content("pages/home")
289
290 assert parser == mock_parser_instance
291
292 # Should check cache
293 mock_parser_cls.from_cache.assert_called_with(provider_mock, "pages/home")
294
295 # Should fetch from API
296 provider_mock.api.get.assert_called()
297
298 # Should parse structure
299 mock_parser_instance.parse_page_structure.assert_called()
300
301 # Should cache result
302 provider_mock.mass.cache.set.assert_called()
303
304
305async def test_get_page_content_propagates_api_errors(
306 recommendation_manager: TidalRecommendationManager, provider_mock: Mock
307) -> None:
308 """Test API failures propagate so an empty result is never cached."""
309 with patch(
310 "music_assistant.providers.tidal.recommendations.TidalPageParser"
311 ) as mock_parser_cls:
312 mock_parser_cls.from_cache = AsyncMock(return_value=None)
313 provider_mock.api.get.side_effect = ResourceTemporarilyUnavailable("API error")
314
315 with pytest.raises(ResourceTemporarilyUnavailable):
316 await recommendation_manager.get_page_content("pages/home")
317
318 provider_mock.mass.cache.set.assert_not_called()
319