music-assistant-server
16.9 KB•PY
test_integration.py
16.9 KB • 443 lines • python
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
138 async with wait_for_sync_completion(mass):
139 config = await mass.config.save_provider_config(
140 "yandex_music",
141 {"token": "mock_yandex_token", "quality": "high"},
142 )
143 await mass.music.start_sync()
144
145 yield config
146
147
148@pytest.fixture
149async def yandex_music_provider_lossless(
150 mass: MusicAssistant,
151) -> AsyncGenerator[ProviderConfig, None]:
152 """Configure Yandex Music with quality=lossless and mock returning MP3 + FLAC."""
153 artist, album, track, playlist = _load_yandex_objects()
154 search_result = _make_search_result(track, album, artist, playlist)
155 mp3_info = _make_download_info(
156 codec="mp3",
157 direct_link="https://example.com/yandex_track.mp3",
158 bitrate_in_kbps=320,
159 )
160 flac_info = _make_download_info(
161 codec="flac",
162 direct_link="https://example.com/yandex_track.flac",
163 bitrate_in_kbps=0,
164 )
165 download_infos = [mp3_info, flac_info]
166
167 album_with_volumes = type(
168 "AlbumWithVolumes",
169 (),
170 {
171 "id": album.id,
172 "title": album.title,
173 "volumes": [[track]],
174 "artists": album.artists if hasattr(album, "artists") else [],
175 "year": getattr(album, "year", None),
176 "release_date": getattr(album, "release_date", None),
177 "genre": getattr(album, "genre", None),
178 "cover_uri": getattr(album, "cover_uri", None),
179 "og_image": getattr(album, "og_image", None),
180 "type": getattr(album, "type", "album"),
181 "available": getattr(album, "available", True),
182 },
183 )()
184
185 with mock.patch(
186 "music_assistant.providers.yandex_music.provider.YandexMusicClient"
187 ) as mock_client_class:
188 mock_client = mock.AsyncMock()
189 mock_client_class.return_value = mock_client
190
191 mock_client.connect = mock.AsyncMock(return_value=True)
192 mock_client.user_id = 12345
193
194 mock_client.get_liked_tracks = mock.AsyncMock(return_value=[])
195 mock_client.get_liked_albums = mock.AsyncMock(return_value=[])
196 mock_client.get_liked_artists = mock.AsyncMock(return_value=[])
197 mock_client.get_user_playlists = mock.AsyncMock(return_value=[playlist])
198
199 mock_client.search = mock.AsyncMock(return_value=search_result)
200 mock_client.get_track = mock.AsyncMock(return_value=track)
201 mock_client.get_tracks = mock.AsyncMock(return_value=[track])
202 mock_client.get_album = mock.AsyncMock(return_value=album)
203 mock_client.get_album_with_tracks = mock.AsyncMock(return_value=album_with_volumes)
204 mock_client.get_artist = mock.AsyncMock(return_value=artist)
205 mock_client.get_artist_albums = mock.AsyncMock(return_value=[album])
206 mock_client.get_artist_tracks = mock.AsyncMock(return_value=[track])
207 mock_client.get_playlist = mock.AsyncMock(return_value=playlist)
208 # get-file-info lossless is tried first; mock returns None so we use download_info path
209 mock_client.get_track_file_info_lossless = mock.AsyncMock(return_value=None)
210 mock_client.get_track_download_info = mock.AsyncMock(return_value=download_infos)
211
212 async with wait_for_sync_completion(mass):
213 config = await mass.config.save_provider_config(
214 "yandex_music",
215 {"token": "mock_yandex_token", "quality": "lossless"},
216 )
217 await mass.music.start_sync()
218
219 yield config
220
221
222def _get_yandex_provider(mass: MusicAssistant) -> MusicProvider | None:
223 """Get Yandex Music provider instance from mass."""
224 for provider in mass.music.providers:
225 if provider.domain == "yandex_music":
226 return provider
227 return None
228
229
230@pytest.mark.usefixtures("yandex_music_provider")
231async def test_registration_and_sync(mass: MusicAssistant) -> None:
232 """Test that provider is registered and sync completes."""
233 prov = _get_yandex_provider(mass)
234 assert prov is not None
235 assert prov.domain == "yandex_music"
236 assert prov.instance_id
237
238
239@pytest.mark.usefixtures("yandex_music_provider")
240async def test_search(mass: MusicAssistant) -> None:
241 """Test search returns results from yandex_music."""
242 results = await mass.music.search("test query", [MediaType.TRACK], limit=5)
243 yandex_tracks = [t for t in results.tracks if t.provider and "yandex_music" in t.provider]
244 assert len(yandex_tracks) >= 0
245
246
247@pytest.mark.usefixtures("yandex_music_provider")
248async def test_get_artist(mass: MusicAssistant) -> None:
249 """Test getting artist by id."""
250 prov = _get_yandex_provider(mass)
251 assert prov is not None
252 artist = await prov.get_artist("100")
253 assert artist is not None
254 assert artist.name
255 assert artist.provider == prov.instance_id
256 assert artist.item_id == "100"
257
258
259@pytest.mark.usefixtures("yandex_music_provider")
260async def test_get_album(mass: MusicAssistant) -> None:
261 """Test getting album by id."""
262 prov = _get_yandex_provider(mass)
263 assert prov is not None
264 album = await prov.get_album("300")
265 assert album is not None
266 assert album.name
267 assert album.provider == prov.instance_id
268 assert album.item_id == "300"
269
270
271@pytest.mark.usefixtures("yandex_music_provider")
272async def test_get_track(mass: MusicAssistant) -> None:
273 """Test getting track by id."""
274 prov = _get_yandex_provider(mass)
275 assert prov is not None
276 track = await prov.get_track("400")
277 assert track is not None
278 assert track.name
279 assert track.provider == prov.instance_id
280 assert track.item_id == "400"
281
282
283@pytest.mark.usefixtures("yandex_music_provider")
284async def test_get_album_tracks(mass: MusicAssistant) -> None:
285 """Test getting album tracks."""
286 prov = _get_yandex_provider(mass)
287 assert prov is not None
288 tracks = await prov.get_album_tracks("300")
289 assert isinstance(tracks, list)
290 assert len(tracks) >= 0
291
292
293@pytest.mark.usefixtures("yandex_music_provider")
294async def test_get_playlist_tracks(mass: MusicAssistant) -> None:
295 """Test getting playlist tracks."""
296 prov = _get_yandex_provider(mass)
297 assert prov is not None
298 tracks = await prov.get_playlist_tracks("12345:3", page=0)
299 assert isinstance(tracks, list)
300 assert len(tracks) >= 0
301
302
303@pytest.mark.usefixtures("yandex_music_provider")
304async def test_get_stream_details(mass: MusicAssistant) -> None:
305 """Test stream details retrieval."""
306 prov = _get_yandex_provider(mass)
307 assert prov is not None
308 stream_details = await prov.get_stream_details("400", MediaType.TRACK)
309 assert stream_details is not None
310 assert stream_details.stream_type == StreamType.HTTP
311 assert stream_details.path == "https://example.com/yandex_track.mp3"
312
313
314@pytest.mark.usefixtures("yandex_music_provider_lossless")
315async def test_get_stream_details_returns_flac_when_lossless_selected(
316 mass: MusicAssistant,
317) -> None:
318 """When quality=lossless and API returns MP3+FLAC, stream details use FLAC."""
319 prov = _get_yandex_provider(mass)
320 assert prov is not None
321 stream_details = await prov.get_stream_details("400", MediaType.TRACK)
322 assert stream_details is not None
323 assert stream_details.audio_format.content_type == ContentType.FLAC
324 assert stream_details.path == "https://example.com/yandex_track.flac"
325
326
327@pytest.mark.usefixtures("yandex_music_provider")
328async def test_library_items(mass: MusicAssistant) -> None:
329 """Test library artists, albums, tracks, playlists."""
330 prov = _get_yandex_provider(mass)
331 assert prov is not None
332 instance_id = prov.instance_id
333
334 artists = await mass.music.artists.library_items()
335 yandex_artists = [a for a in artists if a.provider == instance_id]
336 assert len(yandex_artists) >= 0
337
338 albums = await mass.music.albums.library_items()
339 yandex_albums = [a for a in albums if a.provider == instance_id]
340 assert len(yandex_albums) >= 0
341
342 tracks = await mass.music.tracks.library_items()
343 yandex_tracks = [t for t in tracks if t.provider == instance_id]
344 assert len(yandex_tracks) >= 0
345
346 playlists = await mass.music.playlists.library_items()
347 yandex_playlists = [p for p in playlists if p.provider == instance_id]
348 assert len(yandex_playlists) >= 0
349
350
351@pytest.mark.usefixtures("yandex_music_provider")
352async def test_browse(mass: MusicAssistant) -> None:
353 """Test browse root and subpaths."""
354 prov = _get_yandex_provider(mass)
355 assert prov is not None
356 base_path = f"{prov.instance_id}://"
357 root_items = await prov.browse(path=base_path)
358 assert root_items is not None
359 assert isinstance(root_items, (list, tuple))
360 all_names = set(BROWSE_NAMES_RU.values()) | set(BROWSE_NAMES_EN.values())
361 if root_items:
362 first_name = getattr(root_items[0], "name", None)
363 assert first_name in all_names, (
364 f"First folder name {first_name!r} should be from locale mapping"
365 )
366
367 artists_path = f"{prov.instance_id}://artists"
368 artists_items = await prov.browse(path=artists_path)
369 assert artists_items is not None
370 assert isinstance(artists_items, (list, tuple))
371
372
373# -- Playlist edge-case tests --------------------------------------------------
374
375
376@pytest.mark.usefixtures("yandex_music_provider")
377async def test_get_playlist_tracks_page_gt_zero_returns_empty(mass: MusicAssistant) -> None:
378 """Page > 0 returns empty list (Yandex returns all tracks in one call)."""
379 prov = _get_yandex_provider(mass)
380 assert prov is not None
381 # Use a different playlist ID to avoid cache collision with test_get_playlist_tracks
382 result = await prov.get_playlist_tracks("12345:99", page=1)
383 assert result == []
384
385
386@pytest.mark.usefixtures("yandex_music_provider")
387async def test_get_playlist_tracks_fetch_tracks_async_fallback(mass: MusicAssistant) -> None:
388 """When playlist.tracks is None but track_count > 0, fetch_tracks_async is used."""
389 prov = _get_yandex_provider(mass)
390 assert prov is not None
391
392 _, _, track, _ = _load_yandex_objects()
393
394 # Build a playlist object with tracks=None and track_count=5
395 track_short = type("TrackShort", (), {"track_id": 400, "id": 400})()
396 playlist_no_tracks = type(
397 "Playlist",
398 (),
399 {
400 "owner": type("Owner", (), {"uid": 12345})(),
401 "kind": 77,
402 "title": "Fallback Playlist",
403 "tracks": None,
404 "track_count": 5,
405 "fetch_tracks_async": mock.AsyncMock(return_value=[track_short]),
406 },
407 )()
408
409 prov.client.get_playlist = mock.AsyncMock(return_value=playlist_no_tracks) # type: ignore[attr-defined]
410 prov.client.get_tracks = mock.AsyncMock(return_value=[track]) # type: ignore[attr-defined]
411
412 result = await prov.get_playlist_tracks("12345:77", page=0)
413 assert isinstance(result, list)
414 assert len(result) >= 1
415 playlist_no_tracks.fetch_tracks_async.assert_awaited_once()
416
417
418@pytest.mark.usefixtures("yandex_music_provider")
419async def test_get_playlist_tracks_empty_batch_raises(mass: MusicAssistant) -> None:
420 """Empty batch result from get_tracks raises ResourceTemporarilyUnavailable."""
421 prov = _get_yandex_provider(mass)
422 assert prov is not None
423
424 # Build a playlist with tracks that have track_ids
425 track_short = type("TrackShort", (), {"track_id": 400, "id": 400})()
426 playlist_with_tracks = type(
427 "Playlist",
428 (),
429 {
430 "owner": type("Owner", (), {"uid": 12345})(),
431 "kind": 88,
432 "title": "Batch Fail Playlist",
433 "tracks": [track_short],
434 "track_count": 1,
435 },
436 )()
437
438 prov.client.get_playlist = mock.AsyncMock(return_value=playlist_with_tracks) # type: ignore[attr-defined]
439 prov.client.get_tracks = mock.AsyncMock(return_value=[]) # type: ignore[attr-defined]
440
441 with pytest.raises(ResourceTemporarilyUnavailable):
442 await prov.get_playlist_tracks("12345:88", page=0)
443