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