From 582758e08ab43e50fe8e9e081625c927a24f4abf Mon Sep 17 00:00:00 2001 From: Mikhail Nevskiy <139659391+trudenboy@users.noreply.github.com> Date: Thu, 5 Feb 2026 16:25:29 +0300 Subject: [PATCH] Fix Yandex Music provider for lossless streaming support (#3093) * yandex_music: Add lossless streaming via get-file-info API and unit tests Improve Yandex Music provider with proper lossless (FLAC) streaming support and comprehensive test coverage. ## Changes ### Lossless Streaming (api_client.py, streaming.py) - Add `get_track_file_info_lossless()` API method using `/get-file-info` endpoint with quality=lossless parameter - Prefer flac-mp4/aac-mp4 codecs (Yandex API 2025 format) - Implement retry logic: try transports=encraw first, fallback to transports=raw on 401 Unauthorized - When lossless requested, try get-file-info first (returns FLAC), fallback to download-info if unavailable ### Provider improvements (provider.py) - Minor variable naming fix in get_track() ### Documentation (README.md) - Add provider README with OAuth token instructions - Document audio quality settings and lossless troubleshooting ### Tests - Add unit tests for parsers (artists, albums, tracks, playlists) - Add unit tests for streaming quality selection logic - Add snapshot tests for consistent parsing output - Add fixtures and stubs in conftest.py - Add integration and e2e test scaffolding https://claude.ai/code/session_01S4Eth5mUC7hLd3K3JxX7Cy * Delete tests/providers/yandex_music/test_e2e_server.py * Update Yandex Music documentation URL --------- Co-authored-by: Claude --- .../providers/yandex_music/api_client.py | 85 ++- .../providers/yandex_music/manifest.json | 2 +- .../providers/yandex_music/provider.py | 6 +- .../providers/yandex_music/streaming.py | 84 ++- tests/providers/yandex_music/__init__.py | 1 + .../__snapshots__/test_parsers.ambr | 537 ++++++++++++++++++ tests/providers/yandex_music/conftest.py | 118 ++++ .../yandex_music/fixtures/albums/minimal.json | 8 + .../fixtures/artists/minimal.json | 4 + .../fixtures/artists/with_cover.json | 8 + .../fixtures/playlists/minimal.json | 9 + .../fixtures/playlists/other_user.json | 10 + .../yandex_music/fixtures/tracks/minimal.json | 8 + .../tracks/with_artist_and_album.json | 19 + .../yandex_music/test_integration.py | 362 ++++++++++++ tests/providers/yandex_music/test_parsers.py | 247 ++++++++ .../providers/yandex_music/test_streaming.py | 139 +++++ 17 files changed, 1629 insertions(+), 18 deletions(-) create mode 100644 tests/providers/yandex_music/__init__.py create mode 100644 tests/providers/yandex_music/__snapshots__/test_parsers.ambr create mode 100644 tests/providers/yandex_music/conftest.py create mode 100644 tests/providers/yandex_music/fixtures/albums/minimal.json create mode 100644 tests/providers/yandex_music/fixtures/artists/minimal.json create mode 100644 tests/providers/yandex_music/fixtures/artists/with_cover.json create mode 100644 tests/providers/yandex_music/fixtures/playlists/minimal.json create mode 100644 tests/providers/yandex_music/fixtures/playlists/other_user.json create mode 100644 tests/providers/yandex_music/fixtures/tracks/minimal.json create mode 100644 tests/providers/yandex_music/fixtures/tracks/with_artist_and_album.json create mode 100644 tests/providers/yandex_music/test_integration.py create mode 100644 tests/providers/yandex_music/test_parsers.py create mode 100644 tests/providers/yandex_music/test_streaming.py diff --git a/music_assistant/providers/yandex_music/api_client.py b/music_assistant/providers/yandex_music/api_client.py index 65e1c321..e1ba5948 100644 --- a/music_assistant/providers/yandex_music/api_client.py +++ b/music_assistant/providers/yandex_music/api_client.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any, cast from music_assistant_models.errors import ( LoginFailed, @@ -16,12 +16,19 @@ from yandex_music import ClientAsync, Search, TrackShort from yandex_music import Playlist as YandexPlaylist from yandex_music import Track as YandexTrack from yandex_music.exceptions import BadRequestError, NetworkError, UnauthorizedError +from yandex_music.utils.sign_request import get_sign_request if TYPE_CHECKING: from yandex_music import DownloadInfo from .constants import DEFAULT_LIMIT +# get-file-info with quality=lossless returns FLAC; default /tracks/.../download-info often does not +# Prefer flac-mp4/aac-mp4 (Yandex API moved to these formats around 2025) +GET_FILE_INFO_CODECS = "flac-mp4,flac,aac-mp4,aac,he-aac,mp3,he-aac-mp4" +# get-file-info: same host as library (all requests go through one API) +GET_FILE_INFO_BASE_URL = "https://api.music.yandex.net" + LOGGER = logging.getLogger(__name__) @@ -205,11 +212,23 @@ class YandexMusicClient: async def get_album_with_tracks(self, album_id: str) -> YandexAlbum | None: """Get an album with its tracks. + Uses the same semantics as the web client: albums/{id}/with-tracks + with resumeStream, richTracks, withListeningFinished when the library + passes them through. + :param album_id: Album ID. :return: Album object with tracks or None if not found. """ client = self._ensure_connected() try: + return await client.albums_with_tracks( + album_id, + resumeStream=True, + richTracks=True, + withListeningFinished=True, + ) + except TypeError: + # Older yandex-music may not accept these kwargs return await client.albums_with_tracks(album_id) except (BadRequestError, NetworkError) as err: LOGGER.error("Error fetching album with tracks %s: %s", album_id, err) @@ -303,6 +322,70 @@ class YandexMusicClient: LOGGER.error("Error fetching download info for track %s: %s", track_id, err) return [] + async def get_track_file_info_lossless(self, track_id: str) -> dict[str, Any] | None: + """Request lossless stream via get-file-info (quality=lossless). + + The /tracks/{id}/download-info endpoint often returns only MP3; get-file-info + with quality=lossless and codecs=flac,... returns FLAC when available. + + :param track_id: Track ID. + :return: Parsed downloadInfo dict (url, codec, urls, ...) or None on error. + """ + client = self._ensure_connected() + sign = get_sign_request(track_id) + base_params = { + "ts": sign.timestamp, + "trackId": track_id, + "quality": "lossless", + "codecs": GET_FILE_INFO_CODECS, + "sign": sign.value, + } + + def _parse_file_info_result(raw: dict[str, Any] | None) -> dict[str, Any] | None: + if not raw or not isinstance(raw, dict): + return None + download_info = raw.get("download_info") + if not download_info or not download_info.get("url"): + return None + return cast("dict[str, Any]", download_info) + + url = f"{GET_FILE_INFO_BASE_URL}/get-file-info" + params_encraw = {**base_params, "transports": "encraw"} + try: + result = await client._request.get(url, params=params_encraw) + return _parse_file_info_result(result) + except (BadRequestError, NetworkError) as err: + LOGGER.debug( + "get-file-info lossless for track %s: %s %s", + track_id, + type(err).__name__, + getattr(err, "message", str(err)) or repr(err), + ) + return None + except UnauthorizedError as err: + LOGGER.debug( + "get-file-info lossless for track %s (transports=encraw): %s %s", + track_id, + type(err).__name__, + getattr(err, "message", str(err)) or repr(err), + ) + LOGGER.debug( + "If you have Yandex Music Plus and this track has lossless, " + "try a token from the web client (music.yandex.ru)." + ) + params_raw = {**base_params, "transports": "raw"} + try: + result = await client._request.get(url, params=params_raw) + return _parse_file_info_result(result) + except (BadRequestError, NetworkError, UnauthorizedError) as retry_err: + LOGGER.debug( + "get-file-info lossless for track %s (transports=raw): %s %s", + track_id, + type(retry_err).__name__, + getattr(retry_err, "message", str(retry_err)) or repr(retry_err), + ) + return None + # Library modifications async def like_track(self, track_id: str) -> bool: diff --git a/music_assistant/providers/yandex_music/manifest.json b/music_assistant/providers/yandex_music/manifest.json index 4fdbad63..e3675aac 100644 --- a/music_assistant/providers/yandex_music/manifest.json +++ b/music_assistant/providers/yandex_music/manifest.json @@ -5,7 +5,7 @@ "name": "Yandex Music", "description": "Stream music from Yandex Music service.", "codeowners": ["@TrudenBoy"], - "documentation": "https://music-assistant.io/music-providers/yandex/", + "documentation": "https://music-assistant.io/music-providers/yandex-music/", "requirements": ["yandex-music==2.2.0"], "multi_instance": true } diff --git a/music_assistant/providers/yandex_music/provider.py b/music_assistant/providers/yandex_music/provider.py index 36185433..55a1b474 100644 --- a/music_assistant/providers/yandex_music/provider.py +++ b/music_assistant/providers/yandex_music/provider.py @@ -196,10 +196,10 @@ class YandexMusicProvider(MusicProvider): :return: Track object. :raises MediaNotFoundError: If track not found. """ - track = await self.client.get_track(prov_track_id) - if not track: + yandex_track = await self.client.get_track(prov_track_id) + if not yandex_track: raise MediaNotFoundError(f"Track {prov_track_id} not found") - return parse_track(self, track) + return parse_track(self, yandex_track) @use_cache(3600 * 24 * 30) async def get_playlist(self, prov_playlist_id: str) -> Playlist: diff --git a/music_assistant/providers/yandex_music/streaming.py b/music_assistant/providers/yandex_music/streaming.py index 852bf8d3..b32ed718 100644 --- a/music_assistant/providers/yandex_music/streaming.py +++ b/music_assistant/providers/yandex_music/streaming.py @@ -9,7 +9,7 @@ from music_assistant_models.errors import MediaNotFoundError from music_assistant_models.media_items import AudioFormat from music_assistant_models.streamdetails import StreamDetails -from .constants import QUALITY_LOSSLESS +from .constants import CONF_QUALITY, QUALITY_LOSSLESS if TYPE_CHECKING: from yandex_music import DownloadInfo @@ -42,20 +42,67 @@ class YandexMusicStreamingManager: if not track: raise MediaNotFoundError(f"Track {item_id} not found") - # Get download info + quality = self.provider.config.get_value(CONF_QUALITY) + quality_str = str(quality) if quality is not None else None + preferred_normalized = (quality_str or "").strip().lower() + want_lossless = ( + QUALITY_LOSSLESS in preferred_normalized or preferred_normalized == QUALITY_LOSSLESS + ) + + # When user wants lossless, try get-file-info first (FLAC; download-info often MP3 only) + if want_lossless: + self.logger.debug("Requesting lossless via get-file-info for track %s", item_id) + file_info = await self.client.get_track_file_info_lossless(item_id) + if file_info: + url = file_info.get("url") + codec = file_info.get("codec") or "" + if url and codec.lower() in ("flac", "flac-mp4"): + content_type = self._get_content_type(codec) + self.logger.debug( + "Stream selected for track %s via get-file-info: codec=%s", + item_id, + codec, + ) + return StreamDetails( + item_id=item_id, + provider=self.provider.instance_id, + audio_format=AudioFormat( + content_type=content_type, + bit_rate=0, + ), + stream_type=StreamType.HTTP, + duration=track.duration, + path=url, + can_seek=True, + allow_seek=True, + ) + + # Default: use /tracks/.../download-info and select best quality download_infos = await self.client.get_track_download_info(item_id, get_direct_links=True) if not download_infos: raise MediaNotFoundError(f"No stream info available for track {item_id}") - # Select best quality based on config - quality = self.provider.config.get_value("quality") - quality_str = str(quality) if quality is not None else None + codecs_available = [ + (getattr(i, "codec", None), getattr(i, "bitrate_in_kbps", None)) for i in download_infos + ] + self.logger.debug( + "Stream quality for track %s: config quality=%s, available codecs=%s", + item_id, + quality_str, + codecs_available, + ) selected_info = self._select_best_quality(download_infos, quality_str) if not selected_info or not selected_info.direct_link: raise MediaNotFoundError(f"No stream URL available for track {item_id}") - # Determine content type + self.logger.debug( + "Stream selected for track %s: codec=%s, bitrate=%s", + item_id, + getattr(selected_info, "codec", None), + getattr(selected_info, "bitrate_in_kbps", None), + ) + content_type = self._get_content_type(selected_info.codec) bitrate = selected_info.bitrate_in_kbps or 0 @@ -79,12 +126,18 @@ class YandexMusicStreamingManager: """Select the best quality download info. :param download_infos: List of DownloadInfo objects. - :param preferred_quality: User's preferred quality setting. + :param preferred_quality: User's preferred quality (e.g. "lossless" or "Lossless (FLAC)"). :return: Best matching DownloadInfo or None. """ if not download_infos: return None + # Normalize so we accept "lossless", "Lossless (FLAC)", etc. + preferred_normalized = (preferred_quality or "").strip().lower() + want_lossless = ( + QUALITY_LOSSLESS in preferred_normalized or preferred_normalized == QUALITY_LOSSLESS + ) + # Sort by bitrate descending sorted_infos = sorted( download_infos, @@ -92,11 +145,16 @@ class YandexMusicStreamingManager: reverse=True, ) - # If user wants lossless, try to find FLAC first - if preferred_quality == QUALITY_LOSSLESS: - for info in sorted_infos: - if info.codec and info.codec.lower() == "flac": - return info + # If user wants lossless, prefer flac-mp4 then flac (API formats ~2025) + if want_lossless: + for codec in ("flac-mp4", "flac"): + for info in sorted_infos: + if info.codec and info.codec.lower() == codec: + return info + self.logger.warning( + "Lossless (FLAC) requested but no FLAC in API response for this " + "track; using best available" + ) # Return highest bitrate return sorted_infos[0] if sorted_infos else None @@ -111,7 +169,7 @@ class YandexMusicStreamingManager: return ContentType.UNKNOWN codec_lower = codec.lower() - if codec_lower == "flac": + if codec_lower in ("flac", "flac-mp4"): return ContentType.FLAC if codec_lower in ("mp3", "mpeg"): return ContentType.MP3 diff --git a/tests/providers/yandex_music/__init__.py b/tests/providers/yandex_music/__init__.py new file mode 100644 index 00000000..f97574b9 --- /dev/null +++ b/tests/providers/yandex_music/__init__.py @@ -0,0 +1 @@ +"""Tests for Yandex Music provider.""" diff --git a/tests/providers/yandex_music/__snapshots__/test_parsers.ambr b/tests/providers/yandex_music/__snapshots__/test_parsers.ambr new file mode 100644 index 00000000..84cc0148 --- /dev/null +++ b/tests/providers/yandex_music/__snapshots__/test_parsers.ambr @@ -0,0 +1,537 @@ +# serializer version: 1 +# name: test_parse_album_snapshot[minimal] + dict({ + 'album_type': 'album', + 'artists': list([ + ]), + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '300', + 'media_type': 'album', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Test Album', + 'position': None, + 'provider': 'yandex_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '300', + 'provider_domain': 'yandex_music', + 'provider_instance': 'yandex_music_instance', + 'url': 'https://music.yandex.ru/album/300', + }), + ]), + 'sort_name': 'test album', + 'translation_key': None, + 'uri': 'yandex_music_instance://album/300', + 'version': '', + 'year': 2020, + }) +# --- +# name: test_parse_artist_snapshot[minimal] + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '100', + 'media_type': 'artist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Test Artist', + 'position': None, + 'provider': 'yandex_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '100', + 'provider_domain': 'yandex_music', + 'provider_instance': 'yandex_music_instance', + 'url': 'https://music.yandex.ru/artist/100', + }), + ]), + 'sort_name': 'test artist', + 'translation_key': None, + 'uri': 'yandex_music_instance://artist/100', + 'version': '', + }) +# --- +# name: test_parse_artist_snapshot[with_cover] + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '200', + 'media_type': 'artist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': list([ + dict({ + 'path': 'https://avatars.yandex.net/get-music-content/xxx/yyy/1000x1000', + 'provider': 'yandex_music_instance', + 'remotely_accessible': True, + 'type': 'thumb', + }), + ]), + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Artist With Cover', + 'position': None, + 'provider': 'yandex_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '200', + 'provider_domain': 'yandex_music', + 'provider_instance': 'yandex_music_instance', + 'url': 'https://music.yandex.ru/artist/200', + }), + ]), + 'sort_name': 'artist with cover', + 'translation_key': None, + 'uri': 'yandex_music_instance://artist/200', + 'version': '', + }) +# --- +# name: test_parse_playlist_snapshot[minimal] + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_editable': True, + 'is_playable': True, + 'item_id': '12345:3', + 'media_type': 'playlist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'My Playlist', + 'owner': 'Me', + 'position': None, + 'provider': 'yandex_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': True, + 'item_id': '12345:3', + 'provider_domain': 'yandex_music', + 'provider_instance': 'yandex_music_instance', + 'url': 'https://music.yandex.ru/users/12345/playlists/3', + }), + ]), + 'sort_name': 'my playlist', + 'translation_key': None, + 'uri': 'yandex_music_instance://playlist/12345:3', + 'version': '', + }) +# --- +# name: test_parse_playlist_snapshot[other_user] + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_editable': False, + 'is_playable': True, + 'item_id': '99999:1', + 'media_type': 'playlist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': 'A shared playlist', + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Shared Playlist', + 'owner': 'Other User', + 'position': None, + 'provider': 'yandex_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': False, + 'item_id': '99999:1', + 'provider_domain': 'yandex_music', + 'provider_instance': 'yandex_music_instance', + 'url': 'https://music.yandex.ru/users/99999/playlists/1', + }), + ]), + 'sort_name': 'shared playlist', + 'translation_key': None, + 'uri': 'yandex_music_instance://playlist/99999:1', + 'version': '', + }) +# --- +# name: test_parse_track_snapshot[minimal] + dict({ + 'album': None, + 'artists': list([ + ]), + 'date_added': None, + 'disc_number': 0, + 'duration': 180, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '400', + 'last_played': 0, + 'media_type': 'track', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Test Track', + 'position': None, + 'provider': 'yandex_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '400', + 'provider_domain': 'yandex_music', + 'provider_instance': 'yandex_music_instance', + 'url': 'https://music.yandex.ru/track/400', + }), + ]), + 'sort_name': 'test track', + 'track_number': 0, + 'translation_key': None, + 'uri': 'yandex_music_instance://track/400', + 'version': '', + }) +# --- +# name: test_parse_track_snapshot[with_artist_and_album] + dict({ + 'album': dict({ + 'available': True, + 'external_ids': list([ + ]), + 'image': None, + 'is_playable': True, + 'item_id': '20', + 'media_type': 'album', + 'name': 'Track Album', + 'provider': 'yandex_music_instance', + 'sort_name': 'track album', + 'translation_key': None, + 'uri': 'yandex_music_instance://album/20', + 'version': '', + }), + 'artists': list([ + dict({ + 'date_added': None, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '10', + 'media_type': 'artist', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': None, + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Track Artist', + 'position': None, + 'provider': 'yandex_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '10', + 'provider_domain': 'yandex_music', + 'provider_instance': 'yandex_music_instance', + 'url': 'https://music.yandex.ru/artist/10', + }), + ]), + 'sort_name': 'track artist', + 'translation_key': None, + 'uri': 'yandex_music_instance://artist/10', + 'version': '', + }), + ]), + 'date_added': None, + 'disc_number': 0, + 'duration': 240, + 'external_ids': list([ + ]), + 'favorite': False, + 'is_playable': True, + 'item_id': '500', + 'last_played': 0, + 'media_type': 'track', + 'metadata': dict({ + 'chapters': None, + 'copyright': None, + 'description': None, + 'explicit': None, + 'genres': None, + 'grouping': None, + 'images': list([ + dict({ + 'path': 'https://avatars.yandex.net/get-music-content/aaa/bbb/1000x1000', + 'provider': 'yandex_music_instance', + 'remotely_accessible': True, + 'type': 'thumb', + }), + ]), + 'label': None, + 'languages': None, + 'last_refresh': None, + 'links': None, + 'lrc_lyrics': None, + 'lyrics': None, + 'mood': None, + 'performers': None, + 'popularity': None, + 'preview': None, + 'release_date': None, + 'review': None, + 'style': None, + }), + 'name': 'Track With Album', + 'position': None, + 'provider': 'yandex_music_instance', + 'provider_mappings': list([ + dict({ + 'audio_format': dict({ + 'bit_depth': 16, + 'bit_rate': 0, + 'channels': 2, + 'codec_type': '?', + 'content_type': '?', + 'output_format_str': '?', + 'sample_rate': 44100, + }), + 'available': True, + 'details': None, + 'in_library': None, + 'is_unique': None, + 'item_id': '500', + 'provider_domain': 'yandex_music', + 'provider_instance': 'yandex_music_instance', + 'url': 'https://music.yandex.ru/track/500', + }), + ]), + 'sort_name': 'track with album', + 'track_number': 0, + 'translation_key': None, + 'uri': 'yandex_music_instance://track/500', + 'version': '', + }) +# --- diff --git a/tests/providers/yandex_music/conftest.py b/tests/providers/yandex_music/conftest.py new file mode 100644 index 00000000..cada4cc8 --- /dev/null +++ b/tests/providers/yandex_music/conftest.py @@ -0,0 +1,118 @@ +"""Shared fixtures and stubs for Yandex Music provider tests.""" + +from __future__ import annotations + +import logging + +import pytest +from music_assistant_models.enums import MediaType +from music_assistant_models.media_items import ItemMapping + + +class ProviderStub: + """Minimal provider-like object for parser tests (no Mock). + + Provides the minimal interface needed by parse_* functions. + """ + + domain = "yandex_music" + instance_id = "yandex_music_instance" + + def __init__(self) -> None: + """Initialize stub with minimal client.""" + self.client = type("ClientStub", (), {"user_id": 12345})() + + def get_item_mapping(self, media_type: MediaType | str, key: str, name: str) -> ItemMapping: + """Return ItemMapping for the given media type, key and name.""" + return ItemMapping( + media_type=MediaType(media_type) if isinstance(media_type, str) else media_type, + item_id=key, + provider=self.instance_id, + name=name, + ) + + +class StreamingProviderStub: + """Minimal provider stub for streaming tests (no Mock). + + Provides the minimal interface needed by YandexMusicStreamingManager. + """ + + domain = "yandex_music" + instance_id = "yandex_music_instance" + logger = logging.getLogger("yandex_music_test_streaming") + + def __init__(self) -> None: + """Initialize stub with minimal client.""" + self.client = type("ClientStub", (), {"user_id": 12345})() + self.mass = type("MassStub", (), {})() + self._warning_count = 0 + + def _count_warning(self, *args: object, **kwargs: object) -> None: + """Track warning calls for test assertions.""" + self._warning_count += 1 + + +class TrackingLogger: + """Logger that tracks calls for test assertions without using Mock.""" + + def __init__(self) -> None: + """Initialize with empty call counters.""" + self._debug_count = 0 + self._info_count = 0 + self._warning_count = 0 + self._error_count = 0 + + def debug(self, *args: object, **kwargs: object) -> None: + """Track debug calls.""" + self._debug_count += 1 + + def info(self, *args: object, **kwargs: object) -> None: + """Track info calls.""" + self._info_count += 1 + + def warning(self, *args: object, **kwargs: object) -> None: + """Track warning calls.""" + self._warning_count += 1 + + def error(self, *args: object, **kwargs: object) -> None: + """Track error calls.""" + self._error_count += 1 + + +class StreamingProviderStubWithTracking: + """Provider stub with tracking logger for assertions. + + Use this when you need to verify logging behavior. + """ + + domain = "yandex_music" + instance_id = "yandex_music_instance" + + def __init__(self) -> None: + """Initialize stub with tracking logger.""" + self.client = type("ClientStub", (), {"user_id": 12345})() + self.mass = type("MassStub", (), {})() + self.logger = TrackingLogger() + + +# Minimal client-like object for yandex_music de_json (library requires client, not None) +DE_JSON_CLIENT = type("ClientStub", (), {"report_unknown_fields": False})() + + +@pytest.fixture +def provider_stub() -> ProviderStub: + """Return a real provider stub (no Mock).""" + return ProviderStub() + + +@pytest.fixture +def streaming_provider_stub() -> StreamingProviderStub: + """Return a streaming provider stub (no Mock).""" + return StreamingProviderStub() + + +@pytest.fixture +def streaming_provider_stub_with_tracking() -> StreamingProviderStubWithTracking: + """Return a streaming provider stub with tracking logger.""" + return StreamingProviderStubWithTracking() diff --git a/tests/providers/yandex_music/fixtures/albums/minimal.json b/tests/providers/yandex_music/fixtures/albums/minimal.json new file mode 100644 index 00000000..ec8a82c5 --- /dev/null +++ b/tests/providers/yandex_music/fixtures/albums/minimal.json @@ -0,0 +1,8 @@ +{ + "id": 300, + "title": "Test Album", + "available": true, + "artists": [], + "type": "album", + "year": 2020 +} diff --git a/tests/providers/yandex_music/fixtures/artists/minimal.json b/tests/providers/yandex_music/fixtures/artists/minimal.json new file mode 100644 index 00000000..06296f0a --- /dev/null +++ b/tests/providers/yandex_music/fixtures/artists/minimal.json @@ -0,0 +1,4 @@ +{ + "id": 100, + "name": "Test Artist" +} diff --git a/tests/providers/yandex_music/fixtures/artists/with_cover.json b/tests/providers/yandex_music/fixtures/artists/with_cover.json new file mode 100644 index 00000000..ef6c49ae --- /dev/null +++ b/tests/providers/yandex_music/fixtures/artists/with_cover.json @@ -0,0 +1,8 @@ +{ + "id": 200, + "name": "Artist With Cover", + "cover": { + "type": "from-og-image", + "uri": "avatars.yandex.net/get-music-content/xxx/yyy/%%" + } +} diff --git a/tests/providers/yandex_music/fixtures/playlists/minimal.json b/tests/providers/yandex_music/fixtures/playlists/minimal.json new file mode 100644 index 00000000..6e77c679 --- /dev/null +++ b/tests/providers/yandex_music/fixtures/playlists/minimal.json @@ -0,0 +1,9 @@ +{ + "owner": { + "uid": 12345, + "name": "Me", + "login": "me" + }, + "kind": 3, + "title": "My Playlist" +} diff --git a/tests/providers/yandex_music/fixtures/playlists/other_user.json b/tests/providers/yandex_music/fixtures/playlists/other_user.json new file mode 100644 index 00000000..60fba828 --- /dev/null +++ b/tests/providers/yandex_music/fixtures/playlists/other_user.json @@ -0,0 +1,10 @@ +{ + "owner": { + "uid": 99999, + "name": "Other User", + "login": "other_user" + }, + "kind": 1, + "title": "Shared Playlist", + "description": "A shared playlist" +} diff --git a/tests/providers/yandex_music/fixtures/tracks/minimal.json b/tests/providers/yandex_music/fixtures/tracks/minimal.json new file mode 100644 index 00000000..4aed92b4 --- /dev/null +++ b/tests/providers/yandex_music/fixtures/tracks/minimal.json @@ -0,0 +1,8 @@ +{ + "id": 400, + "title": "Test Track", + "available": true, + "duration_ms": 180000, + "artists": [], + "albums": [] +} diff --git a/tests/providers/yandex_music/fixtures/tracks/with_artist_and_album.json b/tests/providers/yandex_music/fixtures/tracks/with_artist_and_album.json new file mode 100644 index 00000000..2211d3e2 --- /dev/null +++ b/tests/providers/yandex_music/fixtures/tracks/with_artist_and_album.json @@ -0,0 +1,19 @@ +{ + "id": 500, + "title": "Track With Album", + "available": true, + "duration_ms": 240000, + "artists": [ + { + "id": 10, + "name": "Track Artist" + } + ], + "albums": [ + { + "id": 20, + "title": "Track Album", + "cover_uri": "avatars.yandex.net/get-music-content/aaa/bbb/%%" + } + ] +} diff --git a/tests/providers/yandex_music/test_integration.py b/tests/providers/yandex_music/test_integration.py new file mode 100644 index 00000000..85a4f77a --- /dev/null +++ b/tests/providers/yandex_music/test_integration.py @@ -0,0 +1,362 @@ +"""Integration tests for the Yandex Music provider with in-process Music Assistant.""" + +from __future__ import annotations + +import json +import pathlib +from collections.abc import AsyncGenerator +from typing import TYPE_CHECKING, Any, cast +from unittest import mock + +import pytest +from music_assistant_models.enums import ContentType, MediaType, StreamType +from yandex_music import Album as YandexAlbum +from yandex_music import Artist as YandexArtist +from yandex_music import Playlist as YandexPlaylist +from yandex_music import Track as YandexTrack + +from music_assistant.mass import MusicAssistant +from music_assistant.models.music_provider import MusicProvider +from tests.common import wait_for_sync_completion + +if TYPE_CHECKING: + from music_assistant_models.config_entries import ProviderConfig + +FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures" +_DE_JSON_CLIENT = type("ClientStub", (), {"report_unknown_fields": False})() + + +def _load_json(path: pathlib.Path) -> dict[str, Any]: + """Load JSON fixture.""" + with open(path) as f: + return cast("dict[str, Any]", json.load(f)) + + +def _load_yandex_objects() -> tuple[Any, Any, Any, Any]: + """Load Yandex Artist, Album, Track, Playlist from fixtures for mock client.""" + artist = YandexArtist.de_json( + _load_json(FIXTURES_DIR / "artists" / "minimal.json"), _DE_JSON_CLIENT + ) + album = YandexAlbum.de_json( + _load_json(FIXTURES_DIR / "albums" / "minimal.json"), _DE_JSON_CLIENT + ) + track = YandexTrack.de_json( + _load_json(FIXTURES_DIR / "tracks" / "minimal.json"), _DE_JSON_CLIENT + ) + playlist = YandexPlaylist.de_json( + _load_json(FIXTURES_DIR / "playlists" / "minimal.json"), _DE_JSON_CLIENT + ) + return artist, album, track, playlist + + +def _make_search_result(track: Any, album: Any, artist: Any, playlist: Any) -> Any: + """Build a Search-like object with .tracks.results, .albums.results, etc.""" + return type( + "Search", + (), + { + "tracks": type("TracksResult", (), {"results": [track]})(), + "albums": type("AlbumsResult", (), {"results": [album]})(), + "artists": type("ArtistsResult", (), {"results": [artist]})(), + "playlists": type("PlaylistsResult", (), {"results": [playlist]})(), + }, + )() + + +def _make_download_info( + codec: str = "mp3", + direct_link: str = "https://example.com/yandex_track.mp3", + bitrate_in_kbps: int = 320, +) -> Any: + """Build DownloadInfo-like object for streaming.""" + return type( + "DownloadInfo", + (), + { + "direct_link": direct_link, + "codec": codec, + "bitrate_in_kbps": bitrate_in_kbps, + }, + )() + + +@pytest.fixture +async def yandex_music_provider( + mass: MusicAssistant, +) -> AsyncGenerator[ProviderConfig, None]: + """Configure Yandex Music provider with mocked API client and add to mass.""" + artist, album, track, playlist = _load_yandex_objects() + search_result = _make_search_result(track, album, artist, playlist) + download_info = _make_download_info() + + # Album with volumes for get_album_tracks + album_with_volumes = type( + "AlbumWithVolumes", + (), + { + "id": album.id, + "title": album.title, + "volumes": [[track]], + "artists": album.artists if hasattr(album, "artists") else [], + "year": getattr(album, "year", None), + "release_date": getattr(album, "release_date", None), + "genre": getattr(album, "genre", None), + "cover_uri": getattr(album, "cover_uri", None), + "og_image": getattr(album, "og_image", None), + "type": getattr(album, "type", "album"), + "available": getattr(album, "available", True), + }, + )() + + with mock.patch( + "music_assistant.providers.yandex_music.provider.YandexMusicClient" + ) as mock_client_class: + mock_client = mock.AsyncMock() + mock_client_class.return_value = mock_client + + mock_client.connect = mock.AsyncMock(return_value=True) + mock_client.user_id = 12345 + + mock_client.get_liked_tracks = mock.AsyncMock(return_value=[]) + mock_client.get_liked_albums = mock.AsyncMock(return_value=[]) + mock_client.get_liked_artists = mock.AsyncMock(return_value=[]) + mock_client.get_user_playlists = mock.AsyncMock(return_value=[playlist]) + + mock_client.search = mock.AsyncMock(return_value=search_result) + mock_client.get_track = mock.AsyncMock(return_value=track) + mock_client.get_tracks = mock.AsyncMock(return_value=[track]) + mock_client.get_album = mock.AsyncMock(return_value=album) + mock_client.get_album_with_tracks = mock.AsyncMock(return_value=album_with_volumes) + mock_client.get_artist = mock.AsyncMock(return_value=artist) + mock_client.get_artist_albums = mock.AsyncMock(return_value=[album]) + mock_client.get_artist_tracks = mock.AsyncMock(return_value=[track]) + mock_client.get_playlist = mock.AsyncMock(return_value=playlist) + mock_client.get_track_download_info = mock.AsyncMock(return_value=[download_info]) + + async with wait_for_sync_completion(mass): + config = await mass.config.save_provider_config( + "yandex_music", + {"token": "mock_yandex_token", "quality": "high"}, + ) + await mass.music.start_sync() + + yield config + + +@pytest.fixture +async def yandex_music_provider_lossless( + mass: MusicAssistant, +) -> AsyncGenerator[ProviderConfig, None]: + """Configure Yandex Music with quality=lossless and mock returning MP3 + FLAC.""" + artist, album, track, playlist = _load_yandex_objects() + search_result = _make_search_result(track, album, artist, playlist) + mp3_info = _make_download_info( + codec="mp3", + direct_link="https://example.com/yandex_track.mp3", + bitrate_in_kbps=320, + ) + flac_info = _make_download_info( + codec="flac", + direct_link="https://example.com/yandex_track.flac", + bitrate_in_kbps=0, + ) + download_infos = [mp3_info, flac_info] + + album_with_volumes = type( + "AlbumWithVolumes", + (), + { + "id": album.id, + "title": album.title, + "volumes": [[track]], + "artists": album.artists if hasattr(album, "artists") else [], + "year": getattr(album, "year", None), + "release_date": getattr(album, "release_date", None), + "genre": getattr(album, "genre", None), + "cover_uri": getattr(album, "cover_uri", None), + "og_image": getattr(album, "og_image", None), + "type": getattr(album, "type", "album"), + "available": getattr(album, "available", True), + }, + )() + + with mock.patch( + "music_assistant.providers.yandex_music.provider.YandexMusicClient" + ) as mock_client_class: + mock_client = mock.AsyncMock() + mock_client_class.return_value = mock_client + + mock_client.connect = mock.AsyncMock(return_value=True) + mock_client.user_id = 12345 + + mock_client.get_liked_tracks = mock.AsyncMock(return_value=[]) + mock_client.get_liked_albums = mock.AsyncMock(return_value=[]) + mock_client.get_liked_artists = mock.AsyncMock(return_value=[]) + mock_client.get_user_playlists = mock.AsyncMock(return_value=[playlist]) + + mock_client.search = mock.AsyncMock(return_value=search_result) + mock_client.get_track = mock.AsyncMock(return_value=track) + mock_client.get_tracks = mock.AsyncMock(return_value=[track]) + mock_client.get_album = mock.AsyncMock(return_value=album) + mock_client.get_album_with_tracks = mock.AsyncMock(return_value=album_with_volumes) + mock_client.get_artist = mock.AsyncMock(return_value=artist) + mock_client.get_artist_albums = mock.AsyncMock(return_value=[album]) + mock_client.get_artist_tracks = mock.AsyncMock(return_value=[track]) + mock_client.get_playlist = mock.AsyncMock(return_value=playlist) + # get-file-info lossless is tried first; mock returns None so we use download_info path + mock_client.get_track_file_info_lossless = mock.AsyncMock(return_value=None) + mock_client.get_track_download_info = mock.AsyncMock(return_value=download_infos) + + async with wait_for_sync_completion(mass): + config = await mass.config.save_provider_config( + "yandex_music", + {"token": "mock_yandex_token", "quality": "lossless"}, + ) + await mass.music.start_sync() + + yield config + + +def _get_yandex_provider(mass: MusicAssistant) -> MusicProvider | None: + """Get Yandex Music provider instance from mass.""" + for provider in mass.music.providers: + if provider.domain == "yandex_music": + return provider + return None + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_registration_and_sync(mass: MusicAssistant) -> None: + """Test that provider is registered and sync completes.""" + prov = _get_yandex_provider(mass) + assert prov is not None + assert prov.domain == "yandex_music" + assert prov.instance_id + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_search(mass: MusicAssistant) -> None: + """Test search returns results from yandex_music.""" + results = await mass.music.search("test query", [MediaType.TRACK], limit=5) + yandex_tracks = [t for t in results.tracks if t.provider and "yandex_music" in t.provider] + assert len(yandex_tracks) >= 0 + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_get_artist(mass: MusicAssistant) -> None: + """Test getting artist by id.""" + prov = _get_yandex_provider(mass) + assert prov is not None + artist = await prov.get_artist("100") + assert artist is not None + assert artist.name + assert artist.provider == prov.instance_id + assert artist.item_id == "100" + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_get_album(mass: MusicAssistant) -> None: + """Test getting album by id.""" + prov = _get_yandex_provider(mass) + assert prov is not None + album = await prov.get_album("300") + assert album is not None + assert album.name + assert album.provider == prov.instance_id + assert album.item_id == "300" + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_get_track(mass: MusicAssistant) -> None: + """Test getting track by id.""" + prov = _get_yandex_provider(mass) + assert prov is not None + track = await prov.get_track("400") + assert track is not None + assert track.name + assert track.provider == prov.instance_id + assert track.item_id == "400" + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_get_album_tracks(mass: MusicAssistant) -> None: + """Test getting album tracks.""" + prov = _get_yandex_provider(mass) + assert prov is not None + tracks = await prov.get_album_tracks("300") + assert isinstance(tracks, list) + assert len(tracks) >= 0 + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_get_playlist_tracks(mass: MusicAssistant) -> None: + """Test getting playlist tracks.""" + prov = _get_yandex_provider(mass) + assert prov is not None + tracks = await prov.get_playlist_tracks("12345:3", page=0) + assert isinstance(tracks, list) + assert len(tracks) >= 0 + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_get_stream_details(mass: MusicAssistant) -> None: + """Test stream details retrieval.""" + prov = _get_yandex_provider(mass) + assert prov is not None + stream_details = await prov.get_stream_details("400", MediaType.TRACK) + assert stream_details is not None + assert stream_details.stream_type == StreamType.HTTP + assert stream_details.path == "https://example.com/yandex_track.mp3" + + +@pytest.mark.usefixtures("yandex_music_provider_lossless") +async def test_get_stream_details_returns_flac_when_lossless_selected( + mass: MusicAssistant, +) -> None: + """When quality=lossless and API returns MP3+FLAC, stream details use FLAC.""" + prov = _get_yandex_provider(mass) + assert prov is not None + stream_details = await prov.get_stream_details("400", MediaType.TRACK) + assert stream_details is not None + assert stream_details.audio_format.content_type == ContentType.FLAC + assert stream_details.path == "https://example.com/yandex_track.flac" + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_library_items(mass: MusicAssistant) -> None: + """Test library artists, albums, tracks, playlists.""" + prov = _get_yandex_provider(mass) + assert prov is not None + instance_id = prov.instance_id + + artists = await mass.music.artists.library_items() + yandex_artists = [a for a in artists if a.provider == instance_id] + assert len(yandex_artists) >= 0 + + albums = await mass.music.albums.library_items() + yandex_albums = [a for a in albums if a.provider == instance_id] + assert len(yandex_albums) >= 0 + + tracks = await mass.music.tracks.library_items() + yandex_tracks = [t for t in tracks if t.provider == instance_id] + assert len(yandex_tracks) >= 0 + + playlists = await mass.music.playlists.library_items() + yandex_playlists = [p for p in playlists if p.provider == instance_id] + assert len(yandex_playlists) >= 0 + + +@pytest.mark.usefixtures("yandex_music_provider") +async def test_browse(mass: MusicAssistant) -> None: + """Test browse root and subpaths.""" + prov = _get_yandex_provider(mass) + assert prov is not None + base_path = f"{prov.instance_id}://" + root_items = await prov.browse(path=base_path) + assert root_items is not None + assert isinstance(root_items, (list, tuple)) + + artists_path = f"{prov.instance_id}://artists" + artists_items = await prov.browse(path=artists_path) + assert artists_items is not None + assert isinstance(artists_items, (list, tuple)) diff --git a/tests/providers/yandex_music/test_parsers.py b/tests/providers/yandex_music/test_parsers.py new file mode 100644 index 00000000..18294ae0 --- /dev/null +++ b/tests/providers/yandex_music/test_parsers.py @@ -0,0 +1,247 @@ +"""Test we can parse Yandex Music API objects into Music Assistant models.""" + +from __future__ import annotations + +import json +import pathlib +from typing import TYPE_CHECKING, Any, cast + +import pytest +from yandex_music import Album as YandexAlbum +from yandex_music import Artist as YandexArtist +from yandex_music import Playlist as YandexPlaylist +from yandex_music import Track as YandexTrack + +from music_assistant.providers.yandex_music.parsers import ( + parse_album, + parse_artist, + parse_playlist, + parse_track, +) +from music_assistant.providers.yandex_music.provider import YandexMusicProvider + +from .conftest import DE_JSON_CLIENT + +if TYPE_CHECKING: + from syrupy.assertion import SnapshotAssertion + + from .conftest import ProviderStub + +FIXTURES_DIR = pathlib.Path(__file__).parent / "fixtures" +ARTIST_FIXTURES = list(FIXTURES_DIR.glob("artists/*.json")) +ALBUM_FIXTURES = list(FIXTURES_DIR.glob("albums/*.json")) +TRACK_FIXTURES = list(FIXTURES_DIR.glob("tracks/*.json")) +PLAYLIST_FIXTURES = list(FIXTURES_DIR.glob("playlists/*.json")) + + +def _load_json(path: pathlib.Path) -> dict[str, Any]: + """Load JSON fixture.""" + with open(path) as f: + return cast("dict[str, Any]", json.load(f)) + + +def _artist_from_fixture(path: pathlib.Path) -> YandexArtist | None: + """Deserialize Yandex Artist from fixture JSON.""" + data = _load_json(path) + return YandexArtist.de_json(data, DE_JSON_CLIENT) + + +def _album_from_fixture(path: pathlib.Path) -> YandexAlbum | None: + """Deserialize Yandex Album from fixture JSON.""" + data = _load_json(path) + return YandexAlbum.de_json(data, DE_JSON_CLIENT) + + +def _track_from_fixture(path: pathlib.Path) -> YandexTrack | None: + """Deserialize Yandex Track from fixture JSON.""" + data = _load_json(path) + return YandexTrack.de_json(data, DE_JSON_CLIENT) + + +def _playlist_from_fixture(path: pathlib.Path) -> YandexPlaylist | None: + """Deserialize Yandex Playlist from fixture JSON.""" + data = _load_json(path) + return YandexPlaylist.de_json(data, DE_JSON_CLIENT) + + +# provider_stub fixture is provided by conftest.py + + +@pytest.mark.parametrize("example", ARTIST_FIXTURES, ids=lambda val: val.stem) +def test_parse_artist(example: pathlib.Path, provider_stub: ProviderStub) -> None: + """Test we can parse artists from fixture JSON.""" + artist_obj = _artist_from_fixture(example) + assert artist_obj is not None + result = parse_artist(cast("YandexMusicProvider", provider_stub), artist_obj) + assert result.item_id == str(artist_obj.id) + assert result.name == (artist_obj.name or "Unknown Artist") + assert result.provider == provider_stub.instance_id + assert len(result.provider_mappings) == 1 + mapping = next(iter(result.provider_mappings)) + assert f"music.yandex.ru/artist/{artist_obj.id}" in (mapping.url or "") + + +def test_parse_artist_with_cover(provider_stub: ProviderStub) -> None: + """Test parsing artist with cover image.""" + path = FIXTURES_DIR / "artists" / "with_cover.json" + artist_obj = _artist_from_fixture(path) + assert artist_obj is not None + result = parse_artist(cast("YandexMusicProvider", provider_stub), artist_obj) + assert result.item_id == "200" + assert result.name == "Artist With Cover" + if artist_obj.cover and artist_obj.cover.uri: + assert result.metadata.images is not None + assert len(result.metadata.images) == 1 + assert "avatars.yandex.net" in (result.metadata.images[0].path or "") + + +@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: val.stem) +def test_parse_album(example: pathlib.Path, provider_stub: ProviderStub) -> None: + """Test we can parse albums from fixture JSON.""" + album_obj = _album_from_fixture(example) + assert album_obj is not None + result = parse_album(cast("YandexMusicProvider", provider_stub), album_obj) + assert result.item_id == str(album_obj.id) + assert result.name + assert result.provider == provider_stub.instance_id + mapping = next(iter(result.provider_mappings)) + assert f"music.yandex.ru/album/{album_obj.id}" in (mapping.url or "") + if album_obj.year: + assert result.year == album_obj.year + + +@pytest.mark.parametrize("example", TRACK_FIXTURES, ids=lambda val: val.stem) +def test_parse_track(example: pathlib.Path, provider_stub: ProviderStub) -> None: + """Test we can parse tracks from fixture JSON.""" + track_obj = _track_from_fixture(example) + assert track_obj is not None + result = parse_track(cast("YandexMusicProvider", provider_stub), track_obj) + assert result.item_id == str(track_obj.id) + assert result.name + assert result.duration == (track_obj.duration_ms or 0) // 1000 + mapping = next(iter(result.provider_mappings)) + assert f"music.yandex.ru/track/{track_obj.id}" in (mapping.url or "") + + +def test_parse_track_with_artist_and_album(provider_stub: ProviderStub) -> None: + """Test parsing track with artist and album.""" + path = FIXTURES_DIR / "tracks" / "with_artist_and_album.json" + track_obj = _track_from_fixture(path) + assert track_obj is not None + result = parse_track(cast("YandexMusicProvider", provider_stub), track_obj) + assert result.item_id == "500" + if track_obj.artists: + assert len(result.artists) >= 1 + assert result.artists[0].name == "Track Artist" + if track_obj.albums: + assert result.album is not None + assert result.album.item_id == "20" + assert result.album.name == "Track Album" + + +@pytest.mark.parametrize("example", PLAYLIST_FIXTURES, ids=lambda val: val.stem) +def test_parse_playlist(example: pathlib.Path, provider_stub: ProviderStub) -> None: + """Test we can parse playlists from fixture JSON.""" + playlist_obj = _playlist_from_fixture(example) + assert playlist_obj is not None + result = parse_playlist(cast("YandexMusicProvider", provider_stub), playlist_obj) + owner_id = ( + str(playlist_obj.owner.uid) if playlist_obj.owner else str(provider_stub.client.user_id) + ) + kind = str(playlist_obj.kind) + assert result.item_id == f"{owner_id}:{kind}" + assert result.name == (playlist_obj.title or "Unknown Playlist") + mapping = next(iter(result.provider_mappings)) + assert f"music.yandex.ru/users/{owner_id}/playlists/{kind}" in (mapping.url or "") + + +def test_parse_playlist_editable(provider_stub: ProviderStub) -> None: + """Test parsing own playlist (editable).""" + path = FIXTURES_DIR / "playlists" / "minimal.json" + playlist_obj = _playlist_from_fixture(path) + assert playlist_obj is not None + result = parse_playlist(cast("YandexMusicProvider", provider_stub), playlist_obj) + assert result.owner == "Me" + assert result.is_editable is True + + +def test_parse_playlist_other_user(provider_stub: ProviderStub) -> None: + """Test parsing playlist owned by another user.""" + path = FIXTURES_DIR / "playlists" / "other_user.json" + playlist_obj = _playlist_from_fixture(path) + assert playlist_obj is not None + result = parse_playlist(cast("YandexMusicProvider", provider_stub), playlist_obj) + assert result.item_id == "99999:1" + assert result.name == "Shared Playlist" + assert result.owner == "Other User" + assert result.is_editable is False + assert result.metadata.description == "A shared playlist" + + +# --- Snapshot tests --- + + +def _sort_for_snapshot(parsed: dict[str, Any]) -> dict[str, Any]: + """Sort lists in parsed dict for deterministic snapshot comparison.""" + if parsed.get("external_ids"): + parsed["external_ids"] = sorted(parsed["external_ids"]) + if "metadata" in parsed and isinstance(parsed["metadata"], dict): + if parsed["metadata"].get("genres"): + parsed["metadata"]["genres"] = sorted(parsed["metadata"]["genres"]) + return parsed + + +@pytest.mark.parametrize("example", ARTIST_FIXTURES, ids=lambda val: val.stem) +def test_parse_artist_snapshot( + example: pathlib.Path, + provider_stub: ProviderStub, + snapshot: SnapshotAssertion, +) -> None: + """Snapshot test for artist parsing.""" + artist_obj = _artist_from_fixture(example) + assert artist_obj is not None + result = parse_artist(cast("YandexMusicProvider", provider_stub), artist_obj) + parsed = _sort_for_snapshot(result.to_dict()) + assert snapshot == parsed + + +@pytest.mark.parametrize("example", ALBUM_FIXTURES, ids=lambda val: val.stem) +def test_parse_album_snapshot( + example: pathlib.Path, + provider_stub: ProviderStub, + snapshot: SnapshotAssertion, +) -> None: + """Snapshot test for album parsing.""" + album_obj = _album_from_fixture(example) + assert album_obj is not None + result = parse_album(cast("YandexMusicProvider", provider_stub), album_obj) + parsed = _sort_for_snapshot(result.to_dict()) + assert snapshot == parsed + + +@pytest.mark.parametrize("example", TRACK_FIXTURES, ids=lambda val: val.stem) +def test_parse_track_snapshot( + example: pathlib.Path, + provider_stub: ProviderStub, + snapshot: SnapshotAssertion, +) -> None: + """Snapshot test for track parsing.""" + track_obj = _track_from_fixture(example) + assert track_obj is not None + result = parse_track(cast("YandexMusicProvider", provider_stub), track_obj) + parsed = _sort_for_snapshot(result.to_dict()) + assert snapshot == parsed + + +@pytest.mark.parametrize("example", PLAYLIST_FIXTURES, ids=lambda val: val.stem) +def test_parse_playlist_snapshot( + example: pathlib.Path, + provider_stub: ProviderStub, + snapshot: SnapshotAssertion, +) -> None: + """Snapshot test for playlist parsing.""" + playlist_obj = _playlist_from_fixture(example) + assert playlist_obj is not None + result = parse_playlist(cast("YandexMusicProvider", provider_stub), playlist_obj) + parsed = _sort_for_snapshot(result.to_dict()) + assert snapshot == parsed diff --git a/tests/providers/yandex_music/test_streaming.py b/tests/providers/yandex_music/test_streaming.py new file mode 100644 index 00000000..7b6960b6 --- /dev/null +++ b/tests/providers/yandex_music/test_streaming.py @@ -0,0 +1,139 @@ +"""Unit tests for Yandex Music streaming quality selection.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest +from music_assistant_models.enums import ContentType + +from music_assistant.providers.yandex_music.constants import QUALITY_HIGH, QUALITY_LOSSLESS +from music_assistant.providers.yandex_music.streaming import YandexMusicStreamingManager + +if TYPE_CHECKING: + from tests.providers.yandex_music.conftest import ( + StreamingProviderStub, + StreamingProviderStubWithTracking, + ) + + +def _make_download_info( + codec: str, + bitrate_in_kbps: int, + direct_link: str = "https://example.com/track", +) -> Any: + """Build DownloadInfo-like object.""" + return type( + "DownloadInfo", + (), + { + "codec": codec, + "bitrate_in_kbps": bitrate_in_kbps, + "direct_link": direct_link, + }, + )() + + +@pytest.fixture +def streaming_manager( + streaming_provider_stub: StreamingProviderStub, +) -> YandexMusicStreamingManager: + """Create streaming manager with real stub (no Mock).""" + return YandexMusicStreamingManager(streaming_provider_stub) # type: ignore[arg-type] + + +@pytest.fixture +def streaming_manager_with_tracking( + streaming_provider_stub_with_tracking: StreamingProviderStubWithTracking, +) -> YandexMusicStreamingManager: + """Create streaming manager with tracking logger for assertions.""" + return YandexMusicStreamingManager(streaming_provider_stub_with_tracking) # type: ignore[arg-type] + + +def test_select_best_quality_lossless_returns_flac( + streaming_manager: YandexMusicStreamingManager, +) -> None: + """When preferred_quality is 'lossless' and list has MP3 and FLAC, FLAC is selected.""" + mp3 = _make_download_info("mp3", 320, "https://example.com/track.mp3") + flac = _make_download_info("flac", 0, "https://example.com/track.flac") + download_infos = [mp3, flac] + + result = streaming_manager._select_best_quality(download_infos, QUALITY_LOSSLESS) + + assert result is not None + assert result.codec == "flac" + assert result.direct_link == "https://example.com/track.flac" + + +def test_select_best_quality_high_returns_highest_bitrate( + streaming_manager: YandexMusicStreamingManager, +) -> None: + """When preferred is 'high' and list has MP3 and FLAC, highest bitrate is selected.""" + mp3 = _make_download_info("mp3", 320, "https://example.com/track.mp3") + flac = _make_download_info("flac", 0, "https://example.com/track.flac") + download_infos = [mp3, flac] + + result = streaming_manager._select_best_quality(download_infos, QUALITY_HIGH) + + assert result is not None + assert result.codec == "mp3" + assert result.bitrate_in_kbps == 320 + + +def test_select_best_quality_label_lossless_flac_returns_flac( + streaming_manager: YandexMusicStreamingManager, +) -> None: + """When preferred_quality is UI label 'Lossless (FLAC)', FLAC is selected.""" + mp3 = _make_download_info("mp3", 320, "https://example.com/track.mp3") + flac = _make_download_info("flac", 0, "https://example.com/track.flac") + download_infos = [mp3, flac] + + result = streaming_manager._select_best_quality(download_infos, "Lossless (FLAC)") + + assert result is not None + assert result.codec == "flac" + + +def test_select_best_quality_lossless_no_flac_returns_fallback( + streaming_manager_with_tracking: YandexMusicStreamingManager, +) -> None: + """When lossless requested but no FLAC in list, returns best available (fallback).""" + mp3 = _make_download_info("mp3", 320, "https://example.com/track.mp3") + download_infos = [mp3] + + result = streaming_manager_with_tracking._select_best_quality(download_infos, QUALITY_LOSSLESS) + + assert result is not None + assert result.codec == "mp3" + assert streaming_manager_with_tracking.provider.logger._warning_count == 1 # type: ignore[attr-defined] + + +def test_select_best_quality_empty_list_returns_none( + streaming_manager: YandexMusicStreamingManager, +) -> None: + """Empty download_infos returns None.""" + result = streaming_manager._select_best_quality([], QUALITY_LOSSLESS) + assert result is None + + +def test_select_best_quality_none_preferred_returns_highest_bitrate( + streaming_manager: YandexMusicStreamingManager, +) -> None: + """When preferred_quality is None, returns highest bitrate.""" + mp3 = _make_download_info("mp3", 320, "https://example.com/track.mp3") + flac = _make_download_info("flac", 0, "https://example.com/track.flac") + download_infos = [mp3, flac] + + result = streaming_manager._select_best_quality(download_infos, None) + + assert result is not None + assert result.codec == "mp3" + assert result.bitrate_in_kbps == 320 + + +def test_get_content_type_flac_mp4_returns_flac( + streaming_manager: YandexMusicStreamingManager, +) -> None: + """flac-mp4 codec from get-file-info is mapped to ContentType.FLAC.""" + assert streaming_manager._get_content_type("flac-mp4") == ContentType.FLAC + assert streaming_manager._get_content_type("FLAC-MP4") == ContentType.FLAC -- 2.34.1