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_only_the_soloist_backend_can_declare_crossfaded_audio() -> None:
101    """
102    Librespot fetches every track on its own, so it has nothing to fade into.
103
104    With no soloist session running yet there is nothing to report either, and the
105    queue's own setting answers instead.
106    """
107    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
108    prov.backend = SoloistBackend(prov)
109    assert prov.delivers_crossfaded_audio is None
110    prov.backend = LibrespotBackend(prov)
111    assert prov.delivers_crossfaded_audio is False
112
113
114@pytest.mark.parametrize(("crossfade_ms", "expected"), [(8000, True), (0, False)])
115def test_a_running_session_reports_the_crossfade_it_was_started_with(
116    crossfade_ms: int, expected: bool
117) -> None:
118    """
119    The engine reads the crossfade once, at spawn, so the session answers for itself.
120
121    Following the current setting instead would claim a fade the running engine is
122    not applying, or deny the one it still is.
123    """
124    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
125    backend = SoloistBackend(prov)
126    prov.backend = backend
127    session = MagicMock()
128    session.usable = True
129    session.crossfade_ms = crossfade_ms
130    backend._session = session
131
132    assert backend.session_crossfades is expected
133    assert prov.delivers_crossfaded_audio is expected
134
135
136def test_turning_spotify_normalization_off_hands_it_back_to_ma() -> None:
137    """With the setting off, MA measures and normalizes as it does for any source."""
138    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
139    cast("MagicMock", prov.config).get_value = MagicMock(return_value=False)
140    prov.backend = SoloistBackend(prov)
141    assert prov.delivers_normalized_audio is False
142
143
144@pytest.mark.parametrize("normalize", [True, False])
145def test_the_engine_is_told_who_normalizes(tmp_path: Path, normalize: bool) -> None:
146    """Exactly one of the two normalizes, and the prefs say which."""
147    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
148    cast("MagicMock", prov.mass).storage_path = str(tmp_path)
149    cast("MagicMock", prov.mass).cache_path = str(tmp_path / "cache")
150    backend = SoloistBackend(prov)
151    prov.backend = backend
152    backend._prepare_data_dir(0, normalize=normalize)
153    prefs = (backend._data_dir / "settings" / "prefs").read_text(encoding="utf-8")
154    assert f"audio.normalize_v2={'true' if normalize else 'false'}" in prefs
155
156
157@pytest.mark.parametrize(
158    ("quality", "media_type", "codec", "bit_depth", "bit_rate"),
159    [
160        # only music is served losslessly
161        ("lossless", MediaType.TRACK, ContentType.FLAC, 24, None),
162        ("lossless", MediaType.PODCAST_EPISODE, ContentType.VORBIS, 16, 320),
163        ("lossless", MediaType.AUDIOBOOK, ContentType.VORBIS, 16, 320),
164        ("very_high", MediaType.TRACK, ContentType.VORBIS, 16, 320),
165        ("high", MediaType.TRACK, ContentType.VORBIS, 16, 160),
166        ("normal", MediaType.TRACK, ContentType.VORBIS, 16, 96),
167    ],
168)
169def test_the_reported_source_format_follows_the_quality_setting(
170    quality: str,
171    media_type: MediaType,
172    codec: ContentType,
173    bit_depth: int,
174    bit_rate: int | None,
175) -> None:
176    """The engine never reports what it fetched, so the configured ceiling is reported."""
177    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
178    cast("MagicMock", prov.config).get_value = MagicMock(return_value=quality)
179    fmt = SoloistBackend(prov).source_audio_format(media_type)
180    assert fmt.codec_type == codec
181    assert fmt.bit_depth == bit_depth
182    assert fmt.sample_rate == 44100
183    assert fmt.bit_rate == bit_rate
184
185
186def test_the_delivered_format_is_always_the_capture_pcm() -> None:
187    """Whatever is reported, the bytes that arrive are the capture sink's PCM."""
188    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
189    handoff = SoloistBackend(prov).handoff_audio_format
190    assert handoff is not None
191    assert handoff.content_type == ContentType.PCM_S32LE
192    assert handoff.bit_depth == 32
193    assert handoff.sample_rate == 44100
194
195
196def test_librespot_hands_over_the_source_untouched() -> None:
197    """Librespot passes Spotify's own file through, so it reports no separate handoff."""
198    backend = LibrespotBackend(_make_provider({}))
199    assert backend.handoff_audio_format is None
200    fmt = backend.source_audio_format(MediaType.TRACK)
201    assert fmt.codec_type == ContentType.VORBIS
202    assert fmt.bit_rate == 320
203
204
205def test_the_backend_streams_at_the_configured_quality() -> None:
206    """The configured tier is what reaches the engine's prefs."""
207    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
208    backend = SoloistBackend(prov)
209    # nothing chosen yet: the ceiling is stated rather than left to the engine
210    assert backend._audio_quality == AUDIO_QUALITY_LOSSLESS
211    cast("MagicMock", prov.config).get_value = MagicMock(return_value="very_high")
212    assert backend._audio_quality == "very_high"
213
214
215async def test_librespot_receives_the_translated_uri(
216    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
217) -> None:
218    """The canonical spotify:track: URI is translated to librespot's spotify:// scheme."""
219    backend = _make_librespot_backend(tmp_path)
220    captured = _install_fake_librespot_process(monkeypatch)
221    chunks = [chunk async for chunk in backend.stream_spotify_uri("spotify:track:xyz", 0)]
222    assert chunks == [b"ogg"]
223    args = captured[0]
224    assert args[args.index("--single-track") + 1] == "spotify://track:xyz"
225    assert "--start-position" not in args
226
227
228async def test_librespot_seek_adds_start_position(
229    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
230) -> None:
231    """A nonzero seek position is passed to librespot as --start-position."""
232    backend = _make_librespot_backend(tmp_path)
233    captured = _install_fake_librespot_process(monkeypatch)
234    async for _chunk in backend.stream_spotify_uri("spotify:track:xyz", 42):
235        pass
236    args = captured[0]
237    assert args[args.index("--start-position") + 1] == "42"
238
239
240def _make_provider(setup_data: dict[str, Any]) -> SpotifyProvider:
241    """Return a SpotifyProvider (bypassing __init__) with the given setup_data."""
242    prov = object.__new__(SpotifyProvider)
243    config = MagicMock(instance_id="spotify--test")
244    config.get_value = MagicMock(return_value=None)
245    config.values = {}
246    prov.config = config
247    prov.manifest = MagicMock(domain="spotify")
248    prov.logger = MagicMock()
249    prov.available = True
250    mass = MagicMock()
251    # get_setup_value reads the live setup_data blob from the store
252    mass.config.get = MagicMock(return_value=setup_data)
253    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
254    # the store keeps values encrypted; decrypt is an identity map for the test
255    mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
256    prov.mass = mass
257    return prov
258
259
260def _make_librespot_backend(tmp_path: Path) -> LibrespotBackend:
261    """Return a stream-ready LibrespotBackend with a stubbed binary and cache dir."""
262    prov = _make_provider({})
263    prov.cache_dir = str(tmp_path / "cache")
264    backend = LibrespotBackend(prov)
265    backend._librespot_bin = "/bin/librespot"
266    return backend
267
268
269def _install_fake_librespot_process(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
270    """Replace AsyncProcess in the librespot backend, returning the captured argv lists."""
271    captured: list[list[str]] = []
272
273    class _FakeProcess:
274        """AsyncProcess stand-in yielding one ogg chunk and exiting cleanly."""
275
276        def __init__(self, args: list[str], **_kwargs: Any) -> None:
277            captured.append(args)
278            self.returncode = 0
279            self.proc = None
280            self._stderr_task: asyncio.Task[None] | None = None
281
282        async def __aenter__(self) -> Self:
283            return self
284
285        async def __aexit__(self, *_exc_info: object) -> None:
286            # consume the attached stderr reader so no pending task leaks a warning
287            if self._stderr_task is not None:
288                await self._stderr_task
289
290        def attach_stderr_reader(self, task: asyncio.Task[None]) -> None:
291            self._stderr_task = task
292
293        async def iter_stderr(self) -> AsyncGenerator[str]:
294            lines: tuple[str, ...] = ()
295            for line in lines:
296                yield line
297
298        async def iter_chunked(self, _n: int = 64000) -> AsyncGenerator[bytes]:
299            yield b"ogg"
300
301    monkeypatch.setattr(
302        "music_assistant.providers.spotify.backends.librespot.AsyncProcess", _FakeProcess
303    )
304    return captured
305