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