/
/
/
1"""Integration tests for the KION 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_kion_objects() -> tuple[Any, Any, Any, Any]:
36 """Load 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/kion_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 kion_music_provider(
85 mass: MusicAssistant,
86) -> AsyncGenerator[ProviderConfig, None]:
87 """Configure KION Music provider with mocked API client and add to mass."""
88 artist, album, track, playlist = _load_kion_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.kion_music.provider.KionMusicClient"
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 "kion_music",
139 {"token": "mock_kion_token", "quality": "high"},
140 )
141 await mass.music.start_sync()
142
143 yield config
144
145
146@pytest.fixture
147async def kion_music_provider_lossless(
148 mass: MusicAssistant,
149) -> AsyncGenerator[ProviderConfig, None]:
150 """Configure KION Music with quality=lossless and mock returning MP3 + FLAC."""
151 artist, album, track, playlist = _load_kion_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/kion_track.mp3",
156 bitrate_in_kbps=320,
157 )
158 flac_info = _make_download_info(
159 codec="flac",
160 direct_link="https://example.com/kion_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.kion_music.provider.KionMusicClient"
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 mock_client.get_track_file_info_lossless = mock.AsyncMock(return_value=None)
207 mock_client.get_track_download_info = mock.AsyncMock(return_value=download_infos)
208
209 async with wait_for_sync_completion(mass):
210 config = await mass.config.save_provider_config(
211 "kion_music",
212 {"token": "mock_kion_token", "quality": "lossless"},
213 )
214 await mass.music.start_sync()
215
216 yield config
217
218
219def _get_kion_provider(mass: MusicAssistant) -> MusicProvider | None:
220 """Get KION Music provider instance from mass."""
221 for provider in mass.music.providers:
222 if provider.domain == "kion_music":
223 return provider
224 return None
225
226
227@pytest.mark.usefixtures("kion_music_provider")
228async def test_registration_and_sync(mass: MusicAssistant) -> None:
229 """Test that provider is registered and sync completes."""
230 prov = _get_kion_provider(mass)
231 assert prov is not None
232 assert prov.domain == "kion_music"
233 assert prov.instance_id
234
235
236@pytest.mark.usefixtures("kion_music_provider")
237async def test_search(mass: MusicAssistant) -> None:
238 """Test search returns results from kion_music."""
239 results = await mass.music.search("test query", [MediaType.TRACK], limit=5)
240 kion_tracks = [t for t in results.tracks if t.provider and "kion_music" in t.provider]
241 assert len(kion_tracks) >= 0
242
243
244@pytest.mark.usefixtures("kion_music_provider")
245async def test_get_artist(mass: MusicAssistant) -> None:
246 """Test getting artist by id."""
247 prov = _get_kion_provider(mass)
248 assert prov is not None
249 artist = await prov.get_artist("100")
250 assert artist is not None
251 assert artist.name
252 assert artist.provider == prov.instance_id
253 assert artist.item_id == "100"
254
255
256@pytest.mark.usefixtures("kion_music_provider")
257async def test_get_album(mass: MusicAssistant) -> None:
258 """Test getting album by id."""
259 prov = _get_kion_provider(mass)
260 assert prov is not None
261 album = await prov.get_album("300")
262 assert album is not None
263 assert album.name
264 assert album.provider == prov.instance_id
265 assert album.item_id == "300"
266
267
268@pytest.mark.usefixtures("kion_music_provider")
269async def test_get_track(mass: MusicAssistant) -> None:
270 """Test getting track by id."""
271 prov = _get_kion_provider(mass)
272 assert prov is not None
273 track = await prov.get_track("400")
274 assert track is not None
275 assert track.name
276 assert track.provider == prov.instance_id
277 assert track.item_id == "400"
278
279
280@pytest.mark.usefixtures("kion_music_provider")
281async def test_get_album_tracks(mass: MusicAssistant) -> None:
282 """Test getting album tracks."""
283 prov = _get_kion_provider(mass)
284 assert prov is not None
285 tracks = await prov.get_album_tracks("300")
286 assert isinstance(tracks, list)
287 assert len(tracks) >= 0
288
289
290@pytest.mark.usefixtures("kion_music_provider")
291async def test_get_playlist_tracks(mass: MusicAssistant) -> None:
292 """Test getting playlist tracks."""
293 prov = _get_kion_provider(mass)
294 assert prov is not None
295 tracks = await prov.get_playlist_tracks("12345:3", page=0)
296 assert isinstance(tracks, list)
297 assert len(tracks) >= 0
298
299
300@pytest.mark.usefixtures("kion_music_provider")
301async def test_get_playlist_tracks_page_gt_zero_returns_empty(mass: MusicAssistant) -> None:
302 """Test that page > 0 returns empty list (no server-side pagination)."""
303 prov = _get_kion_provider(mass)
304 assert prov is not None
305 tracks = await prov.get_playlist_tracks("12345:3", page=1)
306 assert tracks == []
307
308
309@pytest.mark.usefixtures("kion_music_provider")
310async def test_get_stream_details(mass: MusicAssistant) -> None:
311 """Test stream details retrieval."""
312 prov = _get_kion_provider(mass)
313 assert prov is not None
314 stream_details = await prov.get_stream_details("400", MediaType.TRACK)
315 assert stream_details is not None
316 assert stream_details.stream_type == StreamType.HTTP
317 assert stream_details.path == "https://example.com/kion_track.mp3"
318
319
320@pytest.mark.usefixtures("kion_music_provider_lossless")
321async def test_get_stream_details_returns_flac_when_lossless_selected(
322 mass: MusicAssistant,
323) -> None:
324 """When quality=lossless and API returns MP3+FLAC, stream details use FLAC."""
325 prov = _get_kion_provider(mass)
326 assert prov is not None
327 stream_details = await prov.get_stream_details("400", MediaType.TRACK)
328 assert stream_details is not None
329 assert stream_details.audio_format.content_type == ContentType.FLAC
330 assert stream_details.path == "https://example.com/kion_track.flac"
331
332
333@pytest.mark.usefixtures("kion_music_provider")
334async def test_library_items(mass: MusicAssistant) -> None:
335 """Test library artists, albums, tracks, playlists."""
336 prov = _get_kion_provider(mass)
337 assert prov is not None
338 instance_id = prov.instance_id
339
340 artists = await mass.music.artists.library_items()
341 kion_artists = [a for a in artists if a.provider == instance_id]
342 assert len(kion_artists) >= 0
343
344 albums = await mass.music.albums.library_items()
345 kion_albums = [a for a in albums if a.provider == instance_id]
346 assert len(kion_albums) >= 0
347
348 tracks = await mass.music.tracks.library_items()
349 kion_tracks = [t for t in tracks if t.provider == instance_id]
350 assert len(kion_tracks) >= 0
351
352 playlists = await mass.music.playlists.library_items()
353 kion_playlists = [p for p in playlists if p.provider == instance_id]
354 assert len(kion_playlists) >= 0
355