/
/
/
1"""Integration tests for the Yandex Music provider with in-process Music Assistant."""
2
3from __future__ import annotations
4
5import json
6import pathlib
7from collections.abc import AsyncGenerator
8from typing import TYPE_CHECKING, Any, cast
9from unittest import mock
10
11import pytest
12from music_assistant_models.enums import ContentType, MediaType, StreamType
13from music_assistant_models.errors import ResourceTemporarilyUnavailable
14from yandex_music import Album as YandexAlbum
15from yandex_music import Artist as YandexArtist
16from yandex_music import Playlist as YandexPlaylist
17from yandex_music import Track as YandexTrack
18
19from music_assistant.mass import MusicAssistant
20from music_assistant.models.music_provider import MusicProvider
21from music_assistant.providers.yandex_music.constants import BROWSE_NAMES_EN, BROWSE_NAMES_RU
22from tests.common import wait_for_sync_completion
23
24if TYPE_CHECKING:
25 from music_assistant_models.config_entries import ProviderConfig
26
27FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures"
28_DE_JSON_CLIENT = type("ClientStub", (), {"report_unknown_fields": False})()
29
30
31def _load_json(path: pathlib.Path) -> dict[str, Any]:
32 """Load JSON fixture."""
33 with open(path) as f:
34 return cast("dict[str, Any]", json.load(f))
35
36
37def _load_yandex_objects() -> tuple[Any, Any, Any, Any]:
38 """Load Yandex Artist, Album, Track, Playlist from fixtures for mock client."""
39 artist = YandexArtist.de_json(
40 _load_json(FIXTURES_DIR / "artists" / "minimal.json"), _DE_JSON_CLIENT
41 )
42 album = YandexAlbum.de_json(
43 _load_json(FIXTURES_DIR / "albums" / "minimal.json"), _DE_JSON_CLIENT
44 )
45 track = YandexTrack.de_json(
46 _load_json(FIXTURES_DIR / "tracks" / "minimal.json"), _DE_JSON_CLIENT
47 )
48 playlist = YandexPlaylist.de_json(
49 _load_json(FIXTURES_DIR / "playlists" / "minimal.json"), _DE_JSON_CLIENT
50 )
51 return artist, album, track, playlist
52
53
54def _make_search_result(track: Any, album: Any, artist: Any, playlist: Any) -> Any:
55 """Build a Search-like object with .tracks.results, .albums.results, etc."""
56 return type(
57 "Search",
58 (),
59 {
60 "tracks": type("TracksResult", (), {"results": [track]})(),
61 "albums": type("AlbumsResult", (), {"results": [album]})(),
62 "artists": type("ArtistsResult", (), {"results": [artist]})(),
63 "playlists": type("PlaylistsResult", (), {"results": [playlist]})(),
64 },
65 )()
66
67
68def _make_download_info(
69 codec: str = "mp3",
70 direct_link: str = "https://example.com/yandex_track.mp3",
71 bitrate_in_kbps: int = 320,
72) -> Any:
73 """Build DownloadInfo-like object for streaming."""
74 return type(
75 "DownloadInfo",
76 (),
77 {
78 "direct_link": direct_link,
79 "codec": codec,
80 "bitrate_in_kbps": bitrate_in_kbps,
81 },
82 )()
83
84
85@pytest.fixture
86async def yandex_music_provider(
87 mass: MusicAssistant,
88) -> AsyncGenerator[ProviderConfig, None]:
89 """Configure Yandex Music provider with mocked API client and add to mass."""
90 artist, album, track, playlist = _load_yandex_objects()
91 search_result = _make_search_result(track, album, artist, playlist)
92 download_info = _make_download_info()
93
94 # Album with volumes for get_album_tracks
95 album_with_volumes = type(
96 "AlbumWithVolumes",
97 (),
98 {
99 "id": album.id,
100 "title": album.title,
101 "volumes": [[track]],
102 "artists": album.artists if hasattr(album, "artists") else [],
103 "year": getattr(album, "year", None),
104 "release_date": getattr(album, "release_date", None),
105 "genre": getattr(album, "genre", None),
106 "cover_uri": getattr(album, "cover_uri", None),
107 "og_image": getattr(album, "og_image", None),
108 "type": getattr(album, "type", "album"),
109 "available": getattr(album, "available", True),
110 },
111 )()
112
113 with mock.patch(
114 "music_assistant.providers.yandex_music.provider.YandexMusicClient"
115 ) as mock_client_class:
116 mock_client = mock.AsyncMock()
117 mock_client_class.return_value = mock_client
118
119 mock_client.connect = mock.AsyncMock(return_value=True)
120 mock_client.user_id = 12345
121
122 mock_client.get_liked_tracks = mock.AsyncMock(return_value=[])
123 mock_client.get_liked_albums = mock.AsyncMock(return_value=[])
124 mock_client.get_liked_artists = mock.AsyncMock(return_value=[])
125 mock_client.get_user_playlists = mock.AsyncMock(return_value=[playlist])
126
127 mock_client.search = mock.AsyncMock(return_value=search_result)
128 mock_client.get_track = mock.AsyncMock(return_value=track)
129 mock_client.get_tracks = mock.AsyncMock(return_value=[track])
130 mock_client.get_album = mock.AsyncMock(return_value=album)
131 mock_client.get_album_with_tracks = mock.AsyncMock(return_value=album_with_volumes)
132 mock_client.get_artist = mock.AsyncMock(return_value=artist)
133 mock_client.get_artist_albums = mock.AsyncMock(return_value=[album])
134 mock_client.get_artist_tracks = mock.AsyncMock(return_value=[track])
135 mock_client.get_playlist = mock.AsyncMock(return_value=playlist)
136 mock_client.get_track_download_info = mock.AsyncMock(return_value=[download_info])
137 mock_client.get_track_lyrics = mock.AsyncMock(return_value=(None, False))
138 mock_client.get_track_lyrics_from_track = mock.AsyncMock(return_value=(None, False))
139
140 async with wait_for_sync_completion(mass):
141 config = await mass.config.save_provider_config(
142 "yandex_music",
143 {"token": "mock_yandex_token", "quality": "high"},
144 )
145 await mass.music.start_sync()
146
147 yield config
148
149
150@pytest.fixture
151async def yandex_music_provider_lossless(
152 mass: MusicAssistant,
153) -> AsyncGenerator[ProviderConfig, None]:
154 """Configure Yandex Music with quality=lossless and mock returning MP3 + FLAC."""
155 artist, album, track, playlist = _load_yandex_objects()
156 search_result = _make_search_result(track, album, artist, playlist)
157 mp3_info = _make_download_info(
158 codec="mp3",
159 direct_link="https://example.com/yandex_track.mp3",
160 bitrate_in_kbps=320,
161 )
162 flac_info = _make_download_info(
163 codec="flac",
164 direct_link="https://example.com/yandex_track.flac",
165 bitrate_in_kbps=0,
166 )
167 download_infos = [mp3_info, flac_info]
168
169 album_with_volumes = type(
170 "AlbumWithVolumes",
171 (),
172 {
173 "id": album.id,
174 "title": album.title,
175 "volumes": [[track]],
176 "artists": album.artists if hasattr(album, "artists") else [],
177 "year": getattr(album, "year", None),
178 "release_date": getattr(album, "release_date", None),
179 "genre": getattr(album, "genre", None),
180 "cover_uri": getattr(album, "cover_uri", None),
181 "og_image": getattr(album, "og_image", None),
182 "type": getattr(album, "type", "album"),
183 "available": getattr(album, "available", True),
184 },
185 )()
186
187 with mock.patch(
188 "music_assistant.providers.yandex_music.provider.YandexMusicClient"
189 ) as mock_client_class:
190 mock_client = mock.AsyncMock()
191 mock_client_class.return_value = mock_client
192
193 mock_client.connect = mock.AsyncMock(return_value=True)
194 mock_client.user_id = 12345
195
196 mock_client.get_liked_tracks = mock.AsyncMock(return_value=[])
197 mock_client.get_liked_albums = mock.AsyncMock(return_value=[])
198 mock_client.get_liked_artists = mock.AsyncMock(return_value=[])
199 mock_client.get_user_playlists = mock.AsyncMock(return_value=[playlist])
200
201 mock_client.search = mock.AsyncMock(return_value=search_result)
202 mock_client.get_track = mock.AsyncMock(return_value=track)
203 mock_client.get_tracks = mock.AsyncMock(return_value=[track])
204 mock_client.get_album = mock.AsyncMock(return_value=album)
205 mock_client.get_album_with_tracks = mock.AsyncMock(return_value=album_with_volumes)
206 mock_client.get_artist = mock.AsyncMock(return_value=artist)
207 mock_client.get_artist_albums = mock.AsyncMock(return_value=[album])
208 mock_client.get_artist_tracks = mock.AsyncMock(return_value=[track])
209 mock_client.get_playlist = mock.AsyncMock(return_value=playlist)
210 # get-file-info lossless is tried first; mock returns None so we use download_info path
211 mock_client.get_track_file_info_lossless = mock.AsyncMock(return_value=None)
212 mock_client.get_track_download_info = mock.AsyncMock(return_value=download_infos)
213 mock_client.get_track_lyrics = mock.AsyncMock(return_value=(None, False))
214 mock_client.get_track_lyrics_from_track = mock.AsyncMock(return_value=(None, False))
215
216 async with wait_for_sync_completion(mass):
217 config = await mass.config.save_provider_config(
218 "yandex_music",
219 {"token": "mock_yandex_token", "quality": "lossless"},
220 )
221 await mass.music.start_sync()
222
223 yield config
224
225
226def _get_yandex_provider(mass: MusicAssistant) -> MusicProvider | None:
227 """Get Yandex Music provider instance from mass."""
228 for provider in mass.music.providers:
229 if provider.domain == "yandex_music":
230 return provider
231 return None
232
233
234@pytest.mark.usefixtures("yandex_music_provider")
235async def test_registration_and_sync(mass: MusicAssistant) -> None:
236 """Test that provider is registered and sync completes."""
237 prov = _get_yandex_provider(mass)
238 assert prov is not None
239 assert prov.domain == "yandex_music"
240 assert prov.instance_id
241
242
243@pytest.mark.usefixtures("yandex_music_provider")
244async def test_search(mass: MusicAssistant) -> None:
245 """Test search returns results from yandex_music."""
246 results = await mass.music.search("test query", [MediaType.TRACK], limit=5)
247 yandex_tracks = [t for t in results.tracks if t.provider and "yandex_music" in t.provider]
248 assert len(yandex_tracks) >= 0
249
250
251@pytest.mark.usefixtures("yandex_music_provider")
252async def test_get_artist(mass: MusicAssistant) -> None:
253 """Test getting artist by id."""
254 prov = _get_yandex_provider(mass)
255 assert prov is not None
256 artist = await prov.get_artist("100")
257 assert artist is not None
258 assert artist.name
259 assert artist.provider == prov.instance_id
260 assert artist.item_id == "100"
261
262
263@pytest.mark.usefixtures("yandex_music_provider")
264async def test_get_album(mass: MusicAssistant) -> None:
265 """Test getting album by id."""
266 prov = _get_yandex_provider(mass)
267 assert prov is not None
268 album = await prov.get_album("300")
269 assert album is not None
270 assert album.name
271 assert album.provider == prov.instance_id
272 assert album.item_id == "300"
273
274
275@pytest.mark.usefixtures("yandex_music_provider")
276async def test_get_track(mass: MusicAssistant) -> None:
277 """Test getting track by id."""
278 prov = _get_yandex_provider(mass)
279 assert prov is not None
280 track = await prov.get_track("400")
281 assert track is not None
282 assert track.name
283 assert track.provider == prov.instance_id
284 assert track.item_id == "400"
285
286
287@pytest.mark.usefixtures("yandex_music_provider")
288async def test_get_album_tracks(mass: MusicAssistant) -> None:
289 """Test getting album tracks."""
290 prov = _get_yandex_provider(mass)
291 assert prov is not None
292 tracks = await prov.get_album_tracks("300")
293 assert isinstance(tracks, list)
294 assert len(tracks) >= 0
295
296
297@pytest.mark.usefixtures("yandex_music_provider")
298async def test_get_playlist_tracks(mass: MusicAssistant) -> None:
299 """Test getting playlist tracks."""
300 prov = _get_yandex_provider(mass)
301 assert prov is not None
302 tracks = await prov.get_playlist_tracks("12345:3", page=0)
303 assert isinstance(tracks, list)
304 assert len(tracks) >= 0
305
306
307@pytest.mark.usefixtures("yandex_music_provider")
308async def test_get_stream_details(mass: MusicAssistant) -> None:
309 """Test stream details retrieval."""
310 prov = _get_yandex_provider(mass)
311 assert prov is not None
312 stream_details = await prov.get_stream_details("400", MediaType.TRACK)
313 assert stream_details is not None
314 assert stream_details.stream_type == StreamType.HTTP
315 assert stream_details.path == "https://example.com/yandex_track.mp3"
316
317
318@pytest.mark.usefixtures("yandex_music_provider_lossless")
319async def test_get_stream_details_returns_flac_when_lossless_selected(
320 mass: MusicAssistant,
321) -> None:
322 """When quality=lossless and API returns MP3+FLAC, stream details use FLAC."""
323 prov = _get_yandex_provider(mass)
324 assert prov is not None
325 stream_details = await prov.get_stream_details("400", MediaType.TRACK)
326 assert stream_details is not None
327 assert stream_details.audio_format.content_type == ContentType.FLAC
328 assert stream_details.path == "https://example.com/yandex_track.flac"
329
330
331@pytest.mark.usefixtures("yandex_music_provider")
332async def test_library_items(mass: MusicAssistant) -> None:
333 """Test library artists, albums, tracks, playlists."""
334 prov = _get_yandex_provider(mass)
335 assert prov is not None
336 instance_id = prov.instance_id
337
338 artists = await mass.music.artists.library_items()
339 yandex_artists = [a for a in artists if a.provider == instance_id]
340 assert len(yandex_artists) >= 0
341
342 albums = await mass.music.albums.library_items()
343 yandex_albums = [a for a in albums if a.provider == instance_id]
344 assert len(yandex_albums) >= 0
345
346 tracks = await mass.music.tracks.library_items()
347 yandex_tracks = [t for t in tracks if t.provider == instance_id]
348 assert len(yandex_tracks) >= 0
349
350 playlists = await mass.music.playlists.library_items()
351 yandex_playlists = [p for p in playlists if p.provider == instance_id]
352 assert len(yandex_playlists) >= 0
353
354
355@pytest.mark.usefixtures("yandex_music_provider")
356async def test_browse(mass: MusicAssistant) -> None:
357 """Test browse root and subpaths."""
358 prov = _get_yandex_provider(mass)
359 assert prov is not None
360 base_path = f"{prov.instance_id}://"
361 root_items = await prov.browse(path=base_path)
362 assert root_items is not None
363 assert isinstance(root_items, (list, tuple))
364 all_names = set(BROWSE_NAMES_RU.values()) | set(BROWSE_NAMES_EN.values())
365 if root_items:
366 first_name = getattr(root_items[0], "name", None)
367 assert first_name in all_names, (
368 f"First folder name {first_name!r} should be from locale mapping"
369 )
370
371 artists_path = f"{prov.instance_id}://artists"
372 artists_items = await prov.browse(path=artists_path)
373 assert artists_items is not None
374 assert isinstance(artists_items, (list, tuple))
375
376
377# -- Playlist edge-case tests --------------------------------------------------
378
379
380@pytest.mark.usefixtures("yandex_music_provider")
381async def test_get_playlist_tracks_page_gt_zero_returns_empty(mass: MusicAssistant) -> None:
382 """Page > 0 returns empty list (Yandex returns all tracks in one call)."""
383 prov = _get_yandex_provider(mass)
384 assert prov is not None
385 # Use a different playlist ID to avoid cache collision with test_get_playlist_tracks
386 result = await prov.get_playlist_tracks("12345:99", page=1)
387 assert result == []
388
389
390@pytest.mark.usefixtures("yandex_music_provider")
391async def test_get_playlist_tracks_fetch_tracks_async_fallback(mass: MusicAssistant) -> None:
392 """When playlist.tracks is None but track_count > 0, fetch_tracks_async is used."""
393 prov = _get_yandex_provider(mass)
394 assert prov is not None
395
396 _, _, track, _ = _load_yandex_objects()
397
398 # Build a playlist object with tracks=None and track_count=5
399 track_short = type("TrackShort", (), {"track_id": 400, "id": 400})()
400 playlist_no_tracks = type(
401 "Playlist",
402 (),
403 {
404 "owner": type("Owner", (), {"uid": 12345})(),
405 "kind": 77,
406 "title": "Fallback Playlist",
407 "tracks": None,
408 "track_count": 5,
409 "fetch_tracks_async": mock.AsyncMock(return_value=[track_short]),
410 },
411 )()
412
413 prov.client.get_playlist = mock.AsyncMock(return_value=playlist_no_tracks) # type: ignore[attr-defined]
414 prov.client.get_tracks = mock.AsyncMock(return_value=[track]) # type: ignore[attr-defined]
415
416 result = await prov.get_playlist_tracks("12345:77", page=0)
417 assert isinstance(result, list)
418 assert len(result) >= 1
419 playlist_no_tracks.fetch_tracks_async.assert_awaited_once()
420
421
422@pytest.mark.usefixtures("yandex_music_provider")
423async def test_get_playlist_tracks_empty_batch_raises(mass: MusicAssistant) -> None:
424 """Empty batch result from get_tracks raises ResourceTemporarilyUnavailable."""
425 prov = _get_yandex_provider(mass)
426 assert prov is not None
427
428 # Build a playlist with tracks that have track_ids
429 track_short = type("TrackShort", (), {"track_id": 400, "id": 400})()
430 playlist_with_tracks = type(
431 "Playlist",
432 (),
433 {
434 "owner": type("Owner", (), {"uid": 12345})(),
435 "kind": 88,
436 "title": "Batch Fail Playlist",
437 "tracks": [track_short],
438 "track_count": 1,
439 },
440 )()
441
442 prov.client.get_playlist = mock.AsyncMock(return_value=playlist_with_tracks) # type: ignore[attr-defined]
443 prov.client.get_tracks = mock.AsyncMock(return_value=[]) # type: ignore[attr-defined]
444
445 with pytest.raises(ResourceTemporarilyUnavailable):
446 await prov.get_playlist_tracks("12345:88", page=0)
447