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