music-assistant-server

1"""
2Tests for the Spotify provider's playback backend selection and wiring.
3
4The playback backend is an explicit per-instance choice stored in setup_data:
5configs predating the choice (key unset) must stay on librespot, "soloist"
6selects the single-track Soloist backend. The per-backend concurrency budget
7and librespot's URI translation are locked down here as well.
8"""
9
10from __future__ import annotations
11
12from typing import TYPE_CHECKING, Any, Self, cast
13from unittest.mock import MagicMock
14
15import pytest
16from music_assistant_models.enums import ContentType, MediaType
17
18from music_assistant.providers.spotify.backends.librespot import LibrespotBackend
19from music_assistant.providers.spotify.backends.soloist import SoloistBackend
20from music_assistant.providers.spotify.constants import (
21    BACKEND_LIBRESPOT,
22    BACKEND_SOLOIST,
23    CONF_AUDIO_QUALITY,
24    CONF_PLAYBACK_BACKEND,
25    CONF_SPOTIFY_NORMALIZATION,
26)
27from music_assistant.providers.spotify.provider import SpotifyProvider
28from music_assistant.providers.spotify_connect.base import AUDIO_QUALITY_LOSSLESS
29
30if TYPE_CHECKING:
31    import asyncio
32    from collections.abc import AsyncGenerator
33    from pathlib import Path
34
35
36def test_realtime_declaration_follows_the_backend() -> None:
37    """Soloist declares realtime delivery; librespot can read ahead."""
38    librespot = LibrespotBackend(_make_provider({}))
39    soloist = SoloistBackend(_make_provider({CONF_PLAYBACK_BACKEND: "soloist"}))
40    assert librespot.is_realtime is False
41    assert soloist.is_realtime is True
42
43
44def test_backend_defaults_to_librespot() -> None:
45    """A config without a stored backend choice (pre-split install) stays on librespot."""
46    prov = _make_provider({})
47    assert isinstance(prov._create_backend(), LibrespotBackend)
48
49
50def test_backend_soloist_is_selected() -> None:
51    """A stored soloist choice selects the Soloist single-track backend."""
52    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
53    assert isinstance(prov._create_backend(), SoloistBackend)
54
55
56def test_max_concurrent_streams_is_two_on_either_backend() -> None:
57    """Two source streams either way: parallel librespot fetches, or one session handover."""
58    assert _make_provider({}).max_concurrent_streams == 2
59    assert _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_LIBRESPOT}).max_concurrent_streams == 2
60    assert _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST}).max_concurrent_streams == 2
61
62
63async def test_the_quality_option_is_offered_for_soloist_only() -> None:
64    """Librespot hands over Spotify's own file untouched, so there is nothing to choose."""
65    soloist = await _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST}).get_config_entries()
66    quality = next(entry for entry in soloist if entry.key == CONF_AUDIO_QUALITY)
67    assert quality.hidden is False
68    assert quality.default_value == AUDIO_QUALITY_LOSSLESS
69    assert [option.value for option in quality.options or []] == [
70        "normal",
71        "high",
72        "very_high",
73        "lossless",
74    ]
75    librespot = await _make_provider({}).get_config_entries()
76    assert next(entry for entry in librespot if entry.key == CONF_AUDIO_QUALITY).hidden is True
77
78
79async def test_spotify_normalization_is_offered_for_soloist_only() -> None:
80    """Librespot hands over the untouched file, so it has nothing to normalize with."""
81    soloist = await _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST}).get_config_entries()
82    entry = next(e for e in soloist if e.key == CONF_SPOTIFY_NORMALIZATION)
83    assert entry.hidden is False
84    assert entry.default_value is True
85    librespot = await _make_provider({}).get_config_entries()
86    assert next(e for e in librespot if e.key == CONF_SPOTIFY_NORMALIZATION).hidden is True
87
88
89def test_only_the_soloist_backend_declares_normalized_audio() -> None:
90    """The declaration follows the backend, not just the setting."""
91    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
92    cast("MagicMock", prov.config).get_value = MagicMock(return_value=True)
93    prov.backend = SoloistBackend(prov)
94    assert prov.delivers_normalized_audio is True
95    # the same setting on librespot declares nothing: its audio is the raw master
96    prov.backend = LibrespotBackend(prov)
97    assert prov.delivers_normalized_audio is False
98
99
100def test_turning_spotify_normalization_off_hands_it_back_to_ma() -> None:
101    """With the setting off, MA measures and normalizes as it does for any source."""
102    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
103    cast("MagicMock", prov.config).get_value = MagicMock(return_value=False)
104    prov.backend = SoloistBackend(prov)
105    assert prov.delivers_normalized_audio is False
106
107
108@pytest.mark.parametrize("normalize", [True, False])
109def test_the_engine_is_told_who_normalizes(tmp_path: Path, normalize: bool) -> None:
110    """Exactly one of the two normalizes, and the prefs say which."""
111    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
112    cast("MagicMock", prov.mass).storage_path = str(tmp_path)
113    cast("MagicMock", prov.mass).cache_path = str(tmp_path / "cache")
114    backend = SoloistBackend(prov)
115    prov.backend = backend
116    backend._prepare_data_dir(0, normalize=normalize)
117    prefs = (backend._data_dir / "settings" / "prefs").read_text(encoding="utf-8")
118    assert f"audio.normalize_v2={'true' if normalize else 'false'}" in prefs
119
120
121@pytest.mark.parametrize(
122    ("quality", "media_type", "codec", "bit_depth", "bit_rate"),
123    [
124        # only music is served losslessly
125        ("lossless", MediaType.TRACK, ContentType.FLAC, 24, None),
126        ("lossless", MediaType.PODCAST_EPISODE, ContentType.VORBIS, 16, 320),
127        ("lossless", MediaType.AUDIOBOOK, ContentType.VORBIS, 16, 320),
128        ("very_high", MediaType.TRACK, ContentType.VORBIS, 16, 320),
129        ("high", MediaType.TRACK, ContentType.VORBIS, 16, 160),
130        ("normal", MediaType.TRACK, ContentType.VORBIS, 16, 96),
131    ],
132)
133def test_the_reported_source_format_follows_the_quality_setting(
134    quality: str,
135    media_type: MediaType,
136    codec: ContentType,
137    bit_depth: int,
138    bit_rate: int | None,
139) -> None:
140    """The engine never reports what it fetched, so the configured ceiling is reported."""
141    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
142    cast("MagicMock", prov.config).get_value = MagicMock(return_value=quality)
143    fmt = SoloistBackend(prov).source_audio_format(media_type)
144    assert fmt.codec_type == codec
145    assert fmt.bit_depth == bit_depth
146    assert fmt.sample_rate == 44100
147    assert fmt.bit_rate == bit_rate
148
149
150def test_the_delivered_format_is_always_the_capture_pcm() -> None:
151    """Whatever is reported, the bytes that arrive are the capture sink's PCM."""
152    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
153    handoff = SoloistBackend(prov).handoff_audio_format
154    assert handoff is not None
155    assert handoff.content_type == ContentType.PCM_S32LE
156    assert handoff.bit_depth == 32
157    assert handoff.sample_rate == 44100
158
159
160def test_librespot_hands_over_the_source_untouched() -> None:
161    """Librespot passes Spotify's own file through, so it reports no separate handoff."""
162    backend = LibrespotBackend(_make_provider({}))
163    assert backend.handoff_audio_format is None
164    fmt = backend.source_audio_format(MediaType.TRACK)
165    assert fmt.codec_type == ContentType.VORBIS
166    assert fmt.bit_rate == 320
167
168
169def test_the_backend_streams_at_the_configured_quality() -> None:
170    """The configured tier is what reaches the engine's prefs."""
171    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
172    backend = SoloistBackend(prov)
173    # nothing chosen yet: the ceiling is stated rather than left to the engine
174    assert backend._audio_quality == AUDIO_QUALITY_LOSSLESS
175    cast("MagicMock", prov.config).get_value = MagicMock(return_value="very_high")
176    assert backend._audio_quality == "very_high"
177
178
179async def test_librespot_receives_the_translated_uri(
180    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
181) -> None:
182    """The canonical spotify:track: URI is translated to librespot's spotify:// scheme."""
183    backend = _make_librespot_backend(tmp_path)
184    captured = _install_fake_librespot_process(monkeypatch)
185    chunks = [chunk async for chunk in backend.stream_spotify_uri("spotify:track:xyz", 0)]
186    assert chunks == [b"ogg"]
187    args = captured[0]
188    assert args[args.index("--single-track") + 1] == "spotify://track:xyz"
189    assert "--start-position" not in args
190
191
192async def test_librespot_seek_adds_start_position(
193    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
194) -> None:
195    """A nonzero seek position is passed to librespot as --start-position."""
196    backend = _make_librespot_backend(tmp_path)
197    captured = _install_fake_librespot_process(monkeypatch)
198    async for _chunk in backend.stream_spotify_uri("spotify:track:xyz", 42):
199        pass
200    args = captured[0]
201    assert args[args.index("--start-position") + 1] == "42"
202
203
204def _make_provider(setup_data: dict[str, Any]) -> SpotifyProvider:
205    """Return a SpotifyProvider (bypassing __init__) with the given setup_data."""
206    prov = object.__new__(SpotifyProvider)
207    config = MagicMock(instance_id="spotify--test")
208    config.get_value = MagicMock(return_value=None)
209    config.values = {}
210    prov.config = config
211    prov.manifest = MagicMock(domain="spotify")
212    prov.logger = MagicMock()
213    prov.available = True
214    mass = MagicMock()
215    # get_setup_value reads the live setup_data blob from the store
216    mass.config.get = MagicMock(return_value=setup_data)
217    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
218    # the store keeps values encrypted; decrypt is an identity map for the test
219    mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
220    prov.mass = mass
221    return prov
222
223
224def _make_librespot_backend(tmp_path: Path) -> LibrespotBackend:
225    """Return a stream-ready LibrespotBackend with a stubbed binary and cache dir."""
226    prov = _make_provider({})
227    prov.cache_dir = str(tmp_path / "cache")
228    backend = LibrespotBackend(prov)
229    backend._librespot_bin = "/bin/librespot"
230    return backend
231
232
233def _install_fake_librespot_process(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
234    """Replace AsyncProcess in the librespot backend, returning the captured argv lists."""
235    captured: list[list[str]] = []
236
237    class _FakeProcess:
238        """AsyncProcess stand-in yielding one ogg chunk and exiting cleanly."""
239
240        def __init__(self, args: list[str], **_kwargs: Any) -> None:
241            captured.append(args)
242            self.returncode = 0
243            self.proc = None
244            self._stderr_task: asyncio.Task[None] | None = None
245
246        async def __aenter__(self) -> Self:
247            return self
248
249        async def __aexit__(self, *_exc_info: object) -> None:
250            # consume the attached stderr reader so no pending task leaks a warning
251            if self._stderr_task is not None:
252                await self._stderr_task
253
254        def attach_stderr_reader(self, task: asyncio.Task[None]) -> None:
255            self._stderr_task = task
256
257        async def iter_stderr(self) -> AsyncGenerator[str]:
258            lines: tuple[str, ...] = ()
259            for line in lines:
260                yield line
261
262        async def iter_chunked(self, _n: int = 64000) -> AsyncGenerator[bytes]:
263            yield b"ogg"
264
265    monkeypatch.setattr(
266        "music_assistant.providers.spotify.backends.librespot.AsyncProcess", _FakeProcess
267    )
268    return captured
269