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