/
/
/
1"""Unit tests for the Spotify provider's curated browse implementation."""
2
3from collections.abc import Generator
4from typing import Any
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8from music_assistant_models.errors import MediaNotFoundError
9from music_assistant_models.media_items import Album, BrowseFolder, Playlist
10
11from music_assistant.models.music_provider import MusicProvider
12from music_assistant.providers.spotify.provider import SpotifyProvider
13from tests.common import use_real_create_task
14
15
16def _make_album_obj(album_id: str, name: str) -> dict[str, Any]:
17 return {
18 "id": album_id,
19 "name": name,
20 "album_type": "album",
21 "external_ids": {},
22 "external_urls": {"spotify": f"https://open.spotify.com/album/{album_id}"},
23 "artists": [
24 {
25 "id": "artist1",
26 "name": "Test Artist",
27 "external_urls": {"spotify": "https://open.spotify.com/artist/artist1"},
28 }
29 ],
30 "images": [],
31 }
32
33
34def _make_playlist_obj(playlist_id: str, name: str) -> dict[str, Any]:
35 return {
36 "id": playlist_id,
37 "name": name,
38 "collaborative": False,
39 "owner": {"id": "spotify", "display_name": "Spotify"},
40 "external_urls": {"spotify": f"https://open.spotify.com/playlist/{playlist_id}"},
41 "images": [],
42 }
43
44
45def _make_category_obj(category_id: str, name: str) -> dict[str, Any]:
46 return {"id": category_id, "name": name}
47
48
49@pytest.fixture
50def get_data() -> AsyncMock:
51 """Return an AsyncMock standing in for SpotifyProvider._get_data."""
52 return AsyncMock()
53
54
55@pytest.fixture
56def provider(get_data: AsyncMock, monkeypatch: pytest.MonkeyPatch) -> SpotifyProvider:
57 """Return a SpotifyProvider with mocked mass/cache, bypassing __init__."""
58 prov = object.__new__(SpotifyProvider)
59 # instance_id and domain are read-only properties backed by config/manifest
60 prov.config = MagicMock(instance_id="spotify--test")
61 prov.manifest = MagicMock(domain="spotify")
62 prov.logger = MagicMock()
63 prov._sp_user = None
64
65 mass = MagicMock()
66 mass.metadata.locale = "de_DE"
67 # bypass the use_cache decorator: always miss
68 mass.cache.get = AsyncMock(return_value=None)
69 mass.cache.get_with_freshness = AsyncMock(return_value=(None, False, False))
70 mass.cache.set = AsyncMock()
71 use_real_create_task(mass)
72 prov.mass = mass
73
74 monkeypatch.setattr(prov, "_get_data", get_data)
75 return prov
76
77
78@pytest.fixture
79def patch_super_browse() -> Generator[AsyncMock]:
80 """Patch MusicProvider.browse so the root listing does not need a full provider."""
81 original = MusicProvider.browse
82 mock = AsyncMock(return_value=[])
83 MusicProvider.browse = mock # type: ignore[method-assign]
84 try:
85 yield mock
86 finally:
87 MusicProvider.browse = original # type: ignore[method-assign]
88
89
90@pytest.mark.asyncio
91async def test_get_new_releases_returns_albums(
92 provider: SpotifyProvider, get_data: AsyncMock
93) -> None:
94 """_get_new_releases parses albums from the browse/new-releases response."""
95 get_data.return_value = {
96 "albums": {"items": [_make_album_obj("a1", "Album 1"), _make_album_obj("a2", "Album 2")]}
97 }
98
99 result = await provider._get_new_releases()
100
101 get_data.assert_awaited_once_with("browse/new-releases", limit=50)
102 assert len(result) == 2
103 assert all(isinstance(a, Album) for a in result)
104
105
106@pytest.mark.asyncio
107async def test_get_new_releases_skips_items_without_id(
108 provider: SpotifyProvider, get_data: AsyncMock
109) -> None:
110 """_get_new_releases ignores malformed album entries."""
111 get_data.return_value = {"albums": {"items": [{"name": "no id"}, None]}}
112
113 result = await provider._get_new_releases()
114
115 assert result == []
116
117
118@pytest.mark.asyncio
119async def test_get_new_releases_handles_not_found(
120 provider: SpotifyProvider, get_data: AsyncMock
121) -> None:
122 """_get_new_releases returns an empty list when the endpoint is unavailable."""
123 get_data.side_effect = MediaNotFoundError("nope")
124
125 result = await provider._get_new_releases()
126
127 assert result == []
128
129
130@pytest.mark.asyncio
131async def test_get_categories_returns_browse_folders(
132 provider: SpotifyProvider, get_data: AsyncMock
133) -> None:
134 """_get_categories maps Spotify categories onto browse folders with stable paths."""
135 get_data.return_value = {
136 "categories": {
137 "items": [_make_category_obj("pop", "Pop"), _make_category_obj("rock", "Rock")]
138 }
139 }
140
141 result = await provider._get_categories("de_DE")
142
143 get_data.assert_awaited_once_with("browse/categories", locale="de_DE", limit=50)
144 assert all(isinstance(f, BrowseFolder) for f in result)
145 pop = result[0]
146 assert pop.item_id == "pop"
147 assert pop.name == "Pop"
148 assert pop.path == "spotify--test://categories/pop"
149 assert pop.is_playable is False
150
151
152@pytest.mark.asyncio
153async def test_get_categories_skips_incomplete(
154 provider: SpotifyProvider, get_data: AsyncMock
155) -> None:
156 """_get_categories ignores categories missing an id or name."""
157 get_data.return_value = {"categories": {"items": [{"id": "x"}, {"name": "y"}, None]}}
158
159 result = await provider._get_categories("de_DE")
160
161 assert result == []
162
163
164@pytest.mark.asyncio
165async def test_get_categories_handles_not_found(
166 provider: SpotifyProvider, get_data: AsyncMock
167) -> None:
168 """_get_categories returns an empty list when the endpoint is unavailable."""
169 get_data.side_effect = MediaNotFoundError("nope")
170
171 result = await provider._get_categories("de_DE")
172
173 assert result == []
174
175
176@pytest.mark.asyncio
177async def test_get_category_playlists_returns_playlists(
178 provider: SpotifyProvider, get_data: AsyncMock
179) -> None:
180 """_get_category_playlists parses playlists via the grandfathered global session."""
181 get_data.return_value = {"playlists": {"items": [_make_playlist_obj("p1", "Playlist 1")]}}
182
183 result = await provider._get_category_playlists("pop", "de_DE")
184
185 get_data.assert_awaited_once_with(
186 "browse/categories/pop/playlists",
187 locale="de_DE",
188 limit=50,
189 use_global_session=True,
190 )
191 assert len(result) == 1
192 assert isinstance(result[0], Playlist)
193
194
195@pytest.mark.asyncio
196async def test_get_category_playlists_handles_not_found(
197 provider: SpotifyProvider, get_data: AsyncMock
198) -> None:
199 """_get_category_playlists returns an empty list when the endpoint is unavailable."""
200 get_data.side_effect = MediaNotFoundError("nope")
201
202 result = await provider._get_category_playlists("pop", "de_DE")
203
204 assert result == []
205
206
207@pytest.mark.asyncio
208async def test_browse_root_prepends_curated_folders(
209 provider: SpotifyProvider, patch_super_browse: AsyncMock
210) -> None:
211 """Browsing the root adds the curated folders before the standard library folders."""
212 patch_super_browse.return_value = [
213 BrowseFolder(item_id="artists", provider=provider.instance_id, name="Artists")
214 ]
215
216 result = await provider.browse("spotify--test://")
217
218 assert [f.item_id for f in result] == ["new-releases", "categories", "artists"]
219 new_releases = next(
220 f for f in result if isinstance(f, BrowseFolder) and f.item_id == "new-releases"
221 )
222 categories = next(
223 f for f in result if isinstance(f, BrowseFolder) and f.item_id == "categories"
224 )
225 assert new_releases.translation_key == "new_releases"
226 assert new_releases.path == "spotify--test://new-releases"
227 assert new_releases.is_playable is True
228 assert categories.translation_key == "genres_and_moods"
229 assert categories.path == "spotify--test://categories"
230 assert categories.is_playable is False
231
232
233@pytest.mark.asyncio
234async def test_browse_dispatches_to_helpers(
235 provider: SpotifyProvider, monkeypatch: pytest.MonkeyPatch
236) -> None:
237 """Browse delegates each curated subpath to the matching cached helper."""
238 new_releases: list[Album] = []
239 folders: list[BrowseFolder] = []
240 playlists: list[Playlist] = []
241 get_new_releases = AsyncMock(return_value=new_releases)
242 get_categories = AsyncMock(return_value=folders)
243 get_category_playlists = AsyncMock(return_value=playlists)
244 monkeypatch.setattr(provider, "_get_new_releases", get_new_releases)
245 monkeypatch.setattr(provider, "_get_categories", get_categories)
246 monkeypatch.setattr(provider, "_get_category_playlists", get_category_playlists)
247
248 assert await provider.browse("spotify--test://new-releases") is new_releases
249 get_new_releases.assert_awaited_once_with()
250
251 assert await provider.browse("spotify--test://categories") is folders
252 get_categories.assert_awaited_once_with("de_DE")
253
254 assert await provider.browse("spotify--test://categories/pop") is playlists
255 get_category_playlists.assert_awaited_once_with("pop", "de_DE")
256