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
17from music_assistant_models.media_items import AudioFormat
18from music_assistant_models.streamdetails import StreamDetails
19
20from music_assistant.providers.spotify.backends.librespot import LibrespotBackend
21from music_assistant.providers.spotify.backends.soloist import SoloistBackend
22from music_assistant.providers.spotify.constants import (
23    BACKEND_LIBRESPOT,
24    BACKEND_SOLOIST,
25    CONF_AUDIO_QUALITY,
26    CONF_PLAYBACK_BACKEND,
27    CONF_SPOTIFY_NORMALIZATION,
28)
29from music_assistant.providers.spotify.provider import SpotifyProvider
30from music_assistant.providers.spotify_connect.base import AUDIO_QUALITY_LOSSLESS
31
32if TYPE_CHECKING:
33    import asyncio
34    from collections.abc import AsyncGenerator
35    from pathlib import Path
36
37
38def test_realtime_declaration_follows_the_backend() -> None:
39    """Soloist declares realtime delivery; librespot can read ahead."""
40    librespot = LibrespotBackend(_make_provider({}))
41    soloist = SoloistBackend(_make_provider({CONF_PLAYBACK_BACKEND: "soloist"}))
42    assert librespot.is_realtime is False
43    assert soloist.is_realtime is True
44
45
46def test_backend_defaults_to_librespot() -> None:
47    """A config without a stored backend choice (pre-split install) stays on librespot."""
48    prov = _make_provider({})
49    assert isinstance(prov._create_backend(), LibrespotBackend)
50
51
52def test_backend_soloist_is_selected() -> None:
53    """A stored soloist choice selects the Soloist single-track backend."""
54    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
55    assert isinstance(prov._create_backend(), SoloistBackend)
56
57
58def test_max_concurrent_streams_is_two_on_either_backend() -> None:
59    """Two source streams either way: parallel librespot fetches, or one session handover."""
60    assert _make_provider({}).max_concurrent_streams == 2
61    assert _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_LIBRESPOT}).max_concurrent_streams == 2
62    assert _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST}).max_concurrent_streams == 2
63
64
65async def test_the_quality_option_is_offered_for_soloist_only() -> None:
66    """Librespot hands over Spotify's own file untouched, so there is nothing to choose."""
67    soloist = await _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST}).get_config_entries()
68    quality = next(entry for entry in soloist if entry.key == CONF_AUDIO_QUALITY)
69    assert quality.hidden is False
70    assert quality.default_value == AUDIO_QUALITY_LOSSLESS
71    assert [option.value for option in quality.options or []] == [
72        "normal",
73        "high",
74        "very_high",
75        "lossless",
76    ]
77    librespot = await _make_provider({}).get_config_entries()
78    assert next(entry for entry in librespot if entry.key == CONF_AUDIO_QUALITY).hidden is True
79
80
81async def test_spotify_normalization_is_offered_for_soloist_only() -> None:
82    """Librespot hands over the untouched file, so it has nothing to normalize with."""
83    soloist = await _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST}).get_config_entries()
84    entry = next(e for e in soloist if e.key == CONF_SPOTIFY_NORMALIZATION)
85    assert entry.hidden is False
86    assert entry.default_value is True
87    librespot = await _make_provider({}).get_config_entries()
88    assert next(e for e in librespot if e.key == CONF_SPOTIFY_NORMALIZATION).hidden is True
89
90
91def test_only_the_soloist_backend_declares_normalized_audio() -> None:
92    """The declaration follows the backend, not just the setting."""
93    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
94    cast("MagicMock", prov.config).get_value = MagicMock(return_value=True)
95    prov.backend = SoloistBackend(prov)
96    assert prov.delivers_normalized_audio(_streamdetails()) is True
97    # the same setting on librespot declares nothing: its audio is the raw master
98    prov.backend = LibrespotBackend(prov)
99    assert prov.delivers_normalized_audio(_streamdetails()) is False
100
101
102def test_only_the_soloist_backend_can_declare_crossfaded_audio() -> None:
103    """
104    Librespot fetches every track on its own, so it has nothing to fade into.
105
106    With no soloist session running yet there is nothing to report either, and the
107    queue's own setting answers instead.
108    """
109    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
110    prov.backend = SoloistBackend(prov)
111    assert prov.delivers_crossfaded_audio(_streamdetails()) is None
112    prov.backend = LibrespotBackend(prov)
113    assert prov.delivers_crossfaded_audio(_streamdetails()) is False
114
115
116@pytest.mark.parametrize("expected", [True, False])
117def test_a_running_session_answers_for_the_boundary_it_is_asked_about(
118    expected: bool,
119) -> None:
120    """
121    The engine reads the crossfade once, at spawn, so the session answers for itself.
122
123    Following the current setting instead would claim a fade the running engine is
124    not applying, or deny the one it still is.
125    """
126    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
127    backend = SoloistBackend(prov)
128    prov.backend = backend
129    streamdetails = _streamdetails(queue_id="queue-1")
130    backend._session = _session(queue_id="queue-1", fades=expected)
131
132    assert backend.session_crossfades(streamdetails) is expected
133    assert prov.delivers_crossfaded_audio(streamdetails) is expected
134
135
136@pytest.mark.parametrize("normalizes", [True, False])
137def test_another_queues_session_answers_for_neither_hook(normalizes: bool) -> None:
138    """
139    A session serves one queue, so it says nothing about an item played on another.
140
141    Reading it anyway would hand the asking queue the other one's answers: MA
142    normalization applied on top of the engine's, or skipped when nobody applies it.
143    """
144    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
145    cast("MagicMock", prov.config).get_value = MagicMock(return_value=normalizes)
146    backend = SoloistBackend(prov)
147    prov.backend = backend
148    backend._session = _session(queue_id="queue-1", fades=True, normalizes=not normalizes)
149    other_queue = _streamdetails(queue_id="queue-2")
150
151    assert backend.session_normalizes(other_queue) is None
152    assert backend.session_crossfades(other_queue) is None
153    # so the configuration answers for normalization, and the queue's own setting
154    # (resolved by the streams core) for the crossfade
155    assert prov.delivers_normalized_audio(other_queue) is normalizes
156    assert prov.delivers_crossfaded_audio(other_queue) is None
157
158
159def test_turning_spotify_normalization_off_hands_it_back_to_ma() -> None:
160    """With the setting off, MA measures and normalizes as it does for any source."""
161    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
162    cast("MagicMock", prov.config).get_value = MagicMock(return_value=False)
163    prov.backend = SoloistBackend(prov)
164    assert prov.delivers_normalized_audio(_streamdetails()) is False
165
166
167@pytest.mark.parametrize("normalize", [True, False])
168def test_the_engine_is_told_who_normalizes(tmp_path: Path, normalize: bool) -> None:
169    """Exactly one of the two normalizes, and the prefs say which."""
170    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
171    cast("MagicMock", prov.mass).storage_path = str(tmp_path)
172    cast("MagicMock", prov.mass).cache_path = str(tmp_path / "cache")
173    backend = SoloistBackend(prov)
174    prov.backend = backend
175    backend._prepare_data_dir(0, normalize=normalize)
176    prefs = (backend._data_dir / "settings" / "prefs").read_text(encoding="utf-8")
177    assert f"audio.normalize_v2={'true' if normalize else 'false'}" in prefs
178
179
180@pytest.mark.parametrize(
181    ("quality", "media_type", "codec", "bit_depth", "bit_rate"),
182    [
183        # only music is served losslessly
184        ("lossless", MediaType.TRACK, ContentType.FLAC, 24, None),
185        ("lossless", MediaType.PODCAST_EPISODE, ContentType.VORBIS, 16, 320),
186        ("lossless", MediaType.AUDIOBOOK, ContentType.VORBIS, 16, 320),
187        ("very_high", MediaType.TRACK, ContentType.VORBIS, 16, 320),
188        ("high", MediaType.TRACK, ContentType.VORBIS, 16, 160),
189        ("normal", MediaType.TRACK, ContentType.VORBIS, 16, 96),
190    ],
191)
192def test_the_reported_source_format_follows_the_quality_setting(
193    quality: str,
194    media_type: MediaType,
195    codec: ContentType,
196    bit_depth: int,
197    bit_rate: int | None,
198) -> None:
199    """The engine never reports what it fetched, so the configured ceiling is reported."""
200    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
201    cast("MagicMock", prov.config).get_value = MagicMock(return_value=quality)
202    fmt = SoloistBackend(prov).source_audio_format(media_type)
203    assert fmt.codec_type == codec
204    assert fmt.bit_depth == bit_depth
205    assert fmt.sample_rate == 44100
206    assert fmt.bit_rate == bit_rate
207
208
209def test_the_delivered_format_is_always_the_capture_pcm() -> None:
210    """Whatever is reported, the bytes that arrive are the capture sink's PCM."""
211    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
212    handoff = SoloistBackend(prov).handoff_audio_format
213    assert handoff is not None
214    assert handoff.content_type == ContentType.PCM_S32LE
215    assert handoff.bit_depth == 32
216    assert handoff.sample_rate == 44100
217
218
219def test_librespot_hands_over_the_source_untouched() -> None:
220    """Librespot passes Spotify's own file through, so it reports no separate handoff."""
221    backend = LibrespotBackend(_make_provider({}))
222    assert backend.handoff_audio_format is None
223    fmt = backend.source_audio_format(MediaType.TRACK)
224    assert fmt.codec_type == ContentType.VORBIS
225    assert fmt.bit_rate == 320
226
227
228def test_the_backend_streams_at_the_configured_quality() -> None:
229    """The configured tier is what reaches the engine's prefs."""
230    prov = _make_provider({CONF_PLAYBACK_BACKEND: BACKEND_SOLOIST})
231    backend = SoloistBackend(prov)
232    # nothing chosen yet: the ceiling is stated rather than left to the engine
233    assert backend._audio_quality == AUDIO_QUALITY_LOSSLESS
234    cast("MagicMock", prov.config).get_value = MagicMock(return_value="very_high")
235    assert backend._audio_quality == "very_high"
236
237
238async def test_librespot_receives_the_translated_uri(
239    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
240) -> None:
241    """The canonical spotify:track: URI is translated to librespot's spotify:// scheme."""
242    backend = _make_librespot_backend(tmp_path)
243    captured = _install_fake_librespot_process(monkeypatch)
244    chunks = [chunk async for chunk in backend.stream_spotify_uri("spotify:track:xyz", 0)]
245    assert chunks == [b"ogg"]
246    args = captured[0]
247    assert args[args.index("--single-track") + 1] == "spotify://track:xyz"
248    assert "--start-position" not in args
249
250
251async def test_librespot_seek_adds_start_position(
252    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
253) -> None:
254    """A nonzero seek position is passed to librespot as --start-position."""
255    backend = _make_librespot_backend(tmp_path)
256    captured = _install_fake_librespot_process(monkeypatch)
257    async for _chunk in backend.stream_spotify_uri("spotify:track:xyz", 42):
258        pass
259    args = captured[0]
260    assert args[args.index("--start-position") + 1] == "42"
261
262
263def _streamdetails(*, queue_id: str | None = None, item_id: str = "track-1") -> StreamDetails:
264    """Return minimal stream details for the source-treatment hooks to answer about."""
265    return StreamDetails(
266        provider="spotify--test",
267        item_id=item_id,
268        audio_format=AudioFormat(content_type=ContentType.PCM_S16LE),
269        media_type=MediaType.TRACK,
270        queue_id=queue_id,
271    )
272
273
274def _session(*, queue_id: str, fades: bool, normalizes: bool = True) -> MagicMock:
275    """Return a stand-in for a running soloist session serving one queue."""
276    session = MagicMock()
277    session.usable = True
278    session.queue_id = queue_id
279    session.engine_normalizes = normalizes
280    session.fades_a_boundary_of = MagicMock(return_value=fades)
281    return session
282
283
284def _make_provider(setup_data: dict[str, Any]) -> SpotifyProvider:
285    """Return a SpotifyProvider (bypassing __init__) with the given setup_data."""
286    prov = object.__new__(SpotifyProvider)
287    config = MagicMock(instance_id="spotify--test")
288    config.get_value = MagicMock(return_value=None)
289    config.values = {}
290    prov.config = config
291    prov.manifest = MagicMock(domain="spotify")
292    prov.logger = MagicMock()
293    prov.available = True
294    mass = MagicMock()
295    # get_setup_value reads the live setup_data blob from the store
296    mass.config.get = MagicMock(return_value=setup_data)
297    mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
298    # the store keeps values encrypted; decrypt is an identity map for the test
299    mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
300    prov.mass = mass
301    return prov
302
303
304def _make_librespot_backend(tmp_path: Path) -> LibrespotBackend:
305    """Return a stream-ready LibrespotBackend with a stubbed binary and cache dir."""
306    prov = _make_provider({})
307    prov.cache_dir = str(tmp_path / "cache")
308    backend = LibrespotBackend(prov)
309    backend._librespot_bin = "/bin/librespot"
310    return backend
311
312
313def _install_fake_librespot_process(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
314    """Replace AsyncProcess in the librespot backend, returning the captured argv lists."""
315    captured: list[list[str]] = []
316
317    class _FakeProcess:
318        """AsyncProcess stand-in yielding one ogg chunk and exiting cleanly."""
319
320        def __init__(self, args: list[str], **_kwargs: Any) -> None:
321            captured.append(args)
322            self.returncode = 0
323            self.proc = None
324            self._stderr_task: asyncio.Task[None] | None = None
325
326        async def __aenter__(self) -> Self:
327            return self
328
329        async def __aexit__(self, *_exc_info: object) -> None:
330            # consume the attached stderr reader so no pending task leaks a warning
331            if self._stderr_task is not None:
332                await self._stderr_task
333
334        def attach_stderr_reader(self, task: asyncio.Task[None]) -> None:
335            self._stderr_task = task
336
337        async def iter_stderr(self) -> AsyncGenerator[str]:
338            lines: tuple[str, ...] = ()
339            for line in lines:
340                yield line
341
342        async def iter_chunked(self, _n: int = 64000) -> AsyncGenerator[bytes]:
343            yield b"ogg"
344
345    monkeypatch.setattr(
346        "music_assistant.providers.spotify.backends.librespot.AsyncProcess", _FakeProcess
347    )
348    return captured
349