/
/
/
1"""Test Deezer's two-method recommendations contract (rows + per-row items)."""
2
3from __future__ import annotations
4
5import asyncio
6from types import SimpleNamespace
7from typing import Any
8from unittest.mock import AsyncMock, Mock
9
10import pytest
11from deezer_python_gql.generated.get_recently_played import (
12 GetRecentlyPlayedMeRecentlyPlayedEdgesNodeFlow,
13)
14from music_assistant_models.unique_list import UniqueList
15
16from music_assistant.providers.deezer.provider import DeezerProvider
17from tests.common import use_real_create_task
18
19ALL_ROW_IDS = [
20 "made_for_you",
21 "recommended_playlists",
22 "recommended_artist_playlists",
23 "recommended_tracks",
24 "new_releases",
25 "mood_flows",
26 "genre_flows",
27 "recently_played",
28]
29
30
31def _playlist_node(node_id: str, title: str) -> SimpleNamespace:
32 return SimpleNamespace(id=node_id, title=title, picture=None, owner=None)
33
34
35def _recs_data() -> SimpleNamespace:
36 """Minimal get_recommendations result feeding all four shared-fetch rows."""
37 track = SimpleNamespace(
38 id="tr1",
39 title="Hot Song",
40 duration=200,
41 contributors=SimpleNamespace(edges=[]),
42 album=None,
43 media=None,
44 is_explicit=False,
45 )
46 album = SimpleNamespace(
47 id="al1",
48 display_title="New Album",
49 contributors=SimpleNamespace(edges=[]),
50 cover=None,
51 type_=None,
52 release_date=None,
53 )
54 return SimpleNamespace(
55 recommendations=SimpleNamespace(
56 playlists=SimpleNamespace(
57 edges=[SimpleNamespace(node=_playlist_node("pl1", "Editorial"))]
58 ),
59 artist_playlists=SimpleNamespace(
60 edges=[SimpleNamespace(node=_playlist_node("pl2", "Artist Mix"))]
61 ),
62 hot_tracks=[track],
63 new_releases=SimpleNamespace(edges=[SimpleNamespace(node=album)]),
64 )
65 )
66
67
68def _flow_configs_data() -> SimpleNamespace:
69 """Minimal get_flow_configs result with one mood and one genre flow."""
70
71 def config_edge(node_id: str, title: str) -> SimpleNamespace:
72 return SimpleNamespace(
73 node=SimpleNamespace(
74 id=node_id,
75 title=title,
76 visuals=SimpleNamespace(hardware_square_icon=None),
77 )
78 )
79
80 return SimpleNamespace(
81 flow_configs=SimpleNamespace(
82 moods=SimpleNamespace(edges=[config_edge("chill", "Chill")]),
83 genres=SimpleNamespace(edges=[config_edge("genre-rock", "Rock")]),
84 )
85 )
86
87
88def _recently_played_data() -> SimpleNamespace:
89 """Minimal get_recently_played result with a single Flow node."""
90 flow_node = GetRecentlyPlayedMeRecentlyPlayedEdgesNodeFlow.model_construct(
91 typename__="Flow", id="flow", title="My Flow", cover=None
92 )
93 return SimpleNamespace(recently_played=SimpleNamespace(edges=[SimpleNamespace(node=flow_node)]))
94
95
96def _stub_gql_client(provider: DeezerProvider) -> Mock:
97 """Attach a gql_client stub with canned data for every recommendations fetch."""
98 gql = Mock()
99 gql.get_recommendations = AsyncMock(return_value=_recs_data())
100 gql.get_flow_config_tracks = AsyncMock(return_value=None) # Flow cover -> None
101 gql.get_made_for_me = AsyncMock(return_value=None) # smart tracklists -> []
102 gql.get_flow_configs = AsyncMock(return_value=_flow_configs_data())
103 gql.get_recently_played = AsyncMock(return_value=_recently_played_data())
104 provider.gql_client = gql
105 return gql
106
107
108def _install_cache_mocks(provider: DeezerProvider) -> None:
109 """Make the @use_cache decorator treat every call as a cache miss."""
110 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
111 return_value=(None, False, False)
112 )
113 provider.mass.cache.set = AsyncMock() # type: ignore[method-assign]
114 use_real_create_task(provider.mass)
115
116
117class _FakeCache:
118 """Dict-backed stand-in for the mass cache, so a second call is a cache hit."""
119
120 def __init__(self, provider: DeezerProvider) -> None:
121 self._store: dict[str, Any] = {}
122 self.background_tasks: list[asyncio.Future[Any]] = []
123 provider.mass.cache.get_with_freshness = AsyncMock( # type: ignore[method-assign]
124 side_effect=self._get
125 )
126 provider.mass.cache.set = AsyncMock(side_effect=self._set) # type: ignore[method-assign]
127 provider.mass.create_task = Mock( # type: ignore[method-assign]
128 side_effect=self._create_task
129 )
130
131 async def _get(self, key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
132 if key in self._store:
133 return self._store[key], True, True
134 return None, False, False
135
136 async def _set(self, key: str, data: Any, **kwargs: Any) -> None:
137 self._store[key] = data
138
139 def _create_task(self, target: Any, *args: Any, **kwargs: Any) -> asyncio.Future[Any]:
140 task: asyncio.Future[Any] = asyncio.ensure_future(target)
141 self.background_tasks.append(task)
142 return task
143
144 async def flush(self) -> None:
145 """Let pending background cache-store tasks complete."""
146 await asyncio.gather(*self.background_tasks)
147 self.background_tasks.clear()
148
149
150@pytest.mark.asyncio
151async def test_get_recommendations_static_rows_zero_backend_calls(
152 provider: DeezerProvider,
153) -> None:
154 """get_recommendations returns the static row descriptors without any backend fetch."""
155 gql = _stub_gql_client(provider)
156
157 rows = await provider.get_recommendations()
158
159 assert [row.item_id for row in rows] == ALL_ROW_IDS
160 assert [row.translation_key for row in rows] == ALL_ROW_IDS
161 assert all(row.provider == provider.instance_id for row in rows)
162 assert all(len(row.items) == 0 for row in rows)
163 gql.get_recommendations.assert_not_awaited()
164 gql.get_flow_config_tracks.assert_not_awaited()
165 gql.get_made_for_me.assert_not_awaited()
166 gql.get_flow_configs.assert_not_awaited()
167 gql.get_recently_played.assert_not_awaited()
168
169
170@pytest.mark.asyncio
171async def test_items_made_for_you(provider: DeezerProvider) -> None:
172 """made_for_you items trigger only the Flow cover and made-for-me fetches."""
173 _install_cache_mocks(provider)
174 gql = _stub_gql_client(provider)
175
176 items = await provider.get_recommendation_items("made_for_you")
177
178 gql.get_flow_config_tracks.assert_awaited_once()
179 gql.get_made_for_me.assert_awaited_once()
180 gql.get_recommendations.assert_not_awaited()
181 gql.get_flow_configs.assert_not_awaited()
182 gql.get_recently_played.assert_not_awaited()
183 assert [item.name for item in items] == ["Flow"]
184
185
186@pytest.mark.asyncio
187async def test_items_recently_played(provider: DeezerProvider) -> None:
188 """recently_played items trigger only the recently-played fetch."""
189 _install_cache_mocks(provider)
190 gql = _stub_gql_client(provider)
191
192 items = await provider.get_recommendation_items("recently_played")
193
194 gql.get_recently_played.assert_awaited_once()
195 gql.get_recommendations.assert_not_awaited()
196 gql.get_flow_config_tracks.assert_not_awaited()
197 gql.get_made_for_me.assert_not_awaited()
198 gql.get_flow_configs.assert_not_awaited()
199 assert [item.name for item in items] == ["My Flow"]
200
201
202@pytest.mark.asyncio
203async def test_items_payload_rows_share_one_gql_fetch(provider: DeezerProvider) -> None:
204 """The four gql-backed rows are all served from one cached get_recommendations fetch."""
205 cache = _FakeCache(provider)
206 gql = _stub_gql_client(provider)
207
208 playlists = await provider.get_recommendation_items("recommended_playlists")
209 await cache.flush()
210 artist_playlists = await provider.get_recommendation_items("recommended_artist_playlists")
211 tracks = await provider.get_recommendation_items("recommended_tracks")
212 albums = await provider.get_recommendation_items("new_releases")
213
214 gql.get_recommendations.assert_awaited_once()
215 assert [item.name for item in playlists] == ["Editorial"]
216 assert [item.name for item in artist_playlists] == ["Artist Mix"]
217 assert [item.name for item in tracks] == ["Hot Song"]
218 assert [item.name for item in albums] == ["New Album"]
219 gql.get_flow_config_tracks.assert_not_awaited()
220 gql.get_made_for_me.assert_not_awaited()
221 gql.get_flow_configs.assert_not_awaited()
222 gql.get_recently_played.assert_not_awaited()
223
224
225@pytest.mark.asyncio
226async def test_items_flow_rows_share_one_fetch(provider: DeezerProvider) -> None:
227 """mood_flows and genre_flows are both served from one cached get_flow_configs fetch."""
228 cache = _FakeCache(provider)
229 gql = _stub_gql_client(provider)
230
231 mood_items = await provider.get_recommendation_items("mood_flows")
232 await cache.flush()
233 genre_items = await provider.get_recommendation_items("genre_flows")
234
235 gql.get_flow_configs.assert_awaited_once()
236 assert [item.name for item in mood_items] == ["Flow: Chill"]
237 assert [item.name for item in genre_items] == ["Flow: Rock"]
238 gql.get_recommendations.assert_not_awaited()
239 gql.get_flow_config_tracks.assert_not_awaited()
240 gql.get_made_for_me.assert_not_awaited()
241 gql.get_recently_played.assert_not_awaited()
242
243
244@pytest.mark.asyncio
245async def test_items_unknown_id_returns_empty(provider: DeezerProvider) -> None:
246 """An unknown row item_id yields an empty result without any backend fetch."""
247 gql = _stub_gql_client(provider)
248
249 result = await provider.get_recommendation_items("bogus_row")
250
251 assert isinstance(result, UniqueList)
252 assert len(result) == 0
253 gql.get_recommendations.assert_not_awaited()
254 gql.get_flow_config_tracks.assert_not_awaited()
255 gql.get_made_for_me.assert_not_awaited()
256 gql.get_flow_configs.assert_not_awaited()
257 gql.get_recently_played.assert_not_awaited()
258