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