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,
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__)
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)
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:
"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
}
: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:
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
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
"""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,
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
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
--- /dev/null
+"""Tests for Yandex Music provider."""
--- /dev/null
+# 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': '',
+ })
+# ---
--- /dev/null
+"""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()
--- /dev/null
+{
+ "id": 300,
+ "title": "Test Album",
+ "available": true,
+ "artists": [],
+ "type": "album",
+ "year": 2020
+}
--- /dev/null
+{
+ "id": 100,
+ "name": "Test Artist"
+}
--- /dev/null
+{
+ "id": 200,
+ "name": "Artist With Cover",
+ "cover": {
+ "type": "from-og-image",
+ "uri": "avatars.yandex.net/get-music-content/xxx/yyy/%%"
+ }
+}
--- /dev/null
+{
+ "owner": {
+ "uid": 12345,
+ "name": "Me",
+ "login": "me"
+ },
+ "kind": 3,
+ "title": "My Playlist"
+}
--- /dev/null
+{
+ "owner": {
+ "uid": 99999,
+ "name": "Other User",
+ "login": "other_user"
+ },
+ "kind": 1,
+ "title": "Shared Playlist",
+ "description": "A shared playlist"
+}
--- /dev/null
+{
+ "id": 400,
+ "title": "Test Track",
+ "available": true,
+ "duration_ms": 180000,
+ "artists": [],
+ "albums": []
+}
--- /dev/null
+{
+ "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/%%"
+ }
+ ]
+}
--- /dev/null
+"""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))
--- /dev/null
+"""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
--- /dev/null
+"""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
projects / music-assistant-server.git / commitdiff