/
/
/
1"""
2Tests for the Spotify provider's playback (librespot) authorization.
3
4Spotify's login5 endpoint only accepts a stored credential minted with the same client id
5librespot presents, so the playback credential is obtained separately from the Web API tokens
6and installed into librespot's cache directory on load. An install without one cannot stream
7and must be sent back through the setup flow.
8"""
9
10from __future__ import annotations
11
12import json
13import logging
14from collections.abc import AsyncIterator
15from pathlib import Path
16from typing import Any
17from unittest.mock import AsyncMock, MagicMock
18
19import pytest
20from music_assistant_models.errors import LoginFailed
21
22from music_assistant.controllers.config.helpers import _AUTH_ERROR_CODES
23from music_assistant.helpers.oauth import authorization_code_from_url
24from music_assistant.models.setup_flow import SetupFlowError
25from music_assistant.providers.spotify.backends.librespot import LibrespotBackend
26from music_assistant.providers.spotify.constants import (
27 CONF_LIBRESPOT_CREDENTIALS,
28 CREDENTIALS_FILE,
29 PAIRING_DEVICE_NAME,
30)
31from music_assistant.providers.spotify.helpers import _log_pairing_output
32from music_assistant.providers.spotify.provider import SpotifyProvider
33
34STORED_CREDENTIALS = '{"username": "tester", "auth_type": 1, "auth_data": "blob"}'
35
36
37def _make_provider(credentials: str | None, cache_dir: str) -> SpotifyProvider:
38 """Return a SpotifyProvider (bypassing __init__) with the given stored credential."""
39 prov = object.__new__(SpotifyProvider)
40 config = MagicMock(instance_id="spotify--test")
41 config.get_value = MagicMock(return_value=None)
42 config.values = {}
43 prov.config = config
44 prov.manifest = MagicMock(domain="spotify")
45 prov.logger = MagicMock()
46 prov.available = True
47 prov.cache_dir = cache_dir
48 setup_data = {CONF_LIBRESPOT_CREDENTIALS: credentials} if credentials is not None else {}
49 mass = MagicMock()
50 # get_setup_value reads the live setup_data blob from the store
51 mass.config.get = MagicMock(return_value=setup_data)
52 mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
53 # the store keeps values encrypted; decrypt is an identity map for the test
54 mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
55 prov.mass = mass
56 return prov
57
58
59def _make_backend(prov: SpotifyProvider, monkeypatch: pytest.MonkeyPatch) -> LibrespotBackend:
60 """Return a LibrespotBackend for the given provider with a stubbed binary lookup."""
61 monkeypatch.setattr(
62 "music_assistant.providers.spotify.backends.librespot.get_librespot_binary",
63 AsyncMock(return_value="/bin/librespot"),
64 )
65 return LibrespotBackend(prov)
66
67
68async def test_stored_credential_is_installed_for_librespot(
69 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
70) -> None:
71 """The stored credential is written to librespot's cache so login5 accepts it."""
72 cache_dir = tmp_path / "cache"
73 prov = _make_provider(STORED_CREDENTIALS, str(cache_dir))
74 await _make_backend(prov, monkeypatch).setup()
75 written = json.loads((cache_dir / CREDENTIALS_FILE).read_text(encoding="utf-8"))
76 assert written["auth_data"] == "blob"
77
78
79async def test_stale_cached_credential_is_replaced(
80 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
81) -> None:
82 """A credential left in the cache from an earlier (now rejected) mint is overwritten."""
83 cache_dir = tmp_path / "cache"
84 cache_dir.mkdir()
85 credentials_file = cache_dir / CREDENTIALS_FILE
86 credentials_file.write_text('{"username": "tester", "auth_data": "stale"}', encoding="utf-8")
87 prov = _make_provider(STORED_CREDENTIALS, str(cache_dir))
88 await _make_backend(prov, monkeypatch).setup()
89 written = json.loads(credentials_file.read_text(encoding="utf-8"))
90 assert written["auth_data"] == "blob"
91
92
93async def test_missing_credential_requires_reauth(
94 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
95) -> None:
96 """Without a stored credential the provider fails with an auth error (AUTH_REQUIRED)."""
97 prov = _make_provider(None, str(tmp_path / "cache"))
98 with pytest.raises(LoginFailed) as err:
99 await _make_backend(prov, monkeypatch).setup()
100 # the error code is what actually drives the provider to AUTH_REQUIRED (and so the
101 # reconfigure prompt); the translation key is what the user reads
102 assert err.value.error_code in _AUTH_ERROR_CODES
103 assert err.value.translation_key == "playback_auth_required"
104
105
106async def test_failed_attempt_loops_back_to_the_choice(monkeypatch: pytest.MonkeyPatch) -> None:
107 """A failed attempt re-offers the choice instead of aborting the already-authorized flow."""
108 from music_assistant.providers.spotify import setup_flow # noqa: PLC0415
109
110 monkeypatch.setattr(
111 setup_flow, "get_librespot_binary", AsyncMock(return_value="/bin/librespot")
112 )
113 # first attempt fails the way a rejected token does, second one succeeds
114 pairing_mock = MagicMock(
115 side_effect=[_raising(LoginFailed("nope")), _returning(STORED_CREDENTIALS)]
116 )
117 monkeypatch.setattr(setup_flow, "librespot_credentials_via_pairing", pairing_mock)
118 session = MagicMock()
119 session.form = AsyncMock(return_value={setup_flow.CONF_PLAYBACK_AUTH_METHOD: "spotify_app"})
120 session.progress_until = AsyncMock(side_effect=_run_awaitable)
121
122 assert await setup_flow._authorize_playback(session, None) == STORED_CREDENTIALS
123 # the form was re-shown, carrying the failure reason rather than aborting the flow
124 assert session.form.await_count == 2
125 assert session.form.await_args_list[1].kwargs["errors"] == {"base": "playback_auth_failed"}
126 pairing_mock.assert_called_with("/bin/librespot", PAIRING_DEVICE_NAME)
127
128
129def test_pairing_device_name_matches_setup_text() -> None:
130 """Pairing instructions consistently identify the temporary Spotify Connect device."""
131 strings_path = Path(__file__).parents[3] / "music_assistant/providers/spotify/strings.json"
132 strings = json.loads(strings_path.read_text(encoding="utf-8"))
133
134 assert PAIRING_DEVICE_NAME == "Music Assistant Pairing"
135 assert PAIRING_DEVICE_NAME in strings["setup_flow"]["playback_auth"]["description"]
136 assert PAIRING_DEVICE_NAME in strings["setup_flow"]["playback_pairing"]["title"]
137 assert PAIRING_DEVICE_NAME in strings["setup_flow"]["playback_pairing"]["progress_text"]
138 assert PAIRING_DEVICE_NAME in strings["errors"]["pairing_not_completed"]
139
140
141async def test_pairing_output_demotes_duplicate_warnings(
142 caplog: pytest.LogCaptureFixture,
143) -> None:
144 """Repeated librespot warnings are debug logged while distinct warnings stay visible."""
145
146 async def stderr_lines() -> AsyncIterator[str]:
147 for line in (
148 "[2026-08-12T00:30:01Z WARN libmdns] No route to host",
149 "[2026-08-12T00:30:02Z WARN libmdns] No route to host",
150 "[2026-08-12T00:30:03Z WARN libmdns] Interface unavailable",
151 ):
152 yield line
153
154 process = MagicMock()
155 process.iter_stderr.return_value = stderr_lines()
156
157 with caplog.at_level(logging.DEBUG, logger="music_assistant.providers.spotify.helpers"):
158 await _log_pairing_output(process)
159
160 records = [record for record in caplog.records if "[librespot-pairing]" in record.message]
161 assert [record.levelno for record in records] == [
162 logging.WARNING,
163 logging.DEBUG,
164 logging.WARNING,
165 ]
166 assert [record.getMessage() for record in records] == [
167 "[librespot-pairing] [2026-08-12T00:30:01Z WARN libmdns] No route to host",
168 "[librespot-pairing] [2026-08-12T00:30:02Z WARN libmdns] No route to host",
169 "[librespot-pairing] [2026-08-12T00:30:03Z WARN libmdns] Interface unavailable",
170 ]
171
172
173async def _run_awaitable(awaitable: Any, **_kwargs: Any) -> Any:
174 """Stand in for session.progress_until, which awaits the work it displays progress for."""
175 return await awaitable
176
177
178async def _raising(err: Exception) -> str:
179 """Return a coroutine that raises, standing in for a failed credential attempt."""
180 raise err
181
182
183async def _returning(value: str) -> str:
184 """Return a coroutine resolving to the given credential."""
185 return value
186
187
188@pytest.mark.parametrize(
189 ("url", "expected"),
190 [
191 ("http://127.0.0.1:5588/login?code=abc123", "abc123"),
192 (" http://127.0.0.1:5588/login?code=abc123&state=x ", "abc123"),
193 ("https://127.0.0.1:5588/login?state=x&code=abc123", "abc123"),
194 ],
195)
196def test_authorization_code_from_url(url: str, expected: str) -> None:
197 """The code is recovered from the (dead) loopback URL the user pastes back."""
198 assert authorization_code_from_url(url) == expected
199
200
201@pytest.mark.parametrize(
202 "url",
203 [
204 "http://127.0.0.1:5588/login?error=access_denied",
205 "http://127.0.0.1:5588/login?code=null",
206 "not a url at all",
207 "",
208 ],
209)
210def test_authorization_code_from_url_rejects_unusable(url: str) -> None:
211 """A denied, empty or malformed paste is reported instead of silently proceeding."""
212 with pytest.raises(SetupFlowError):
213 authorization_code_from_url(url)
214
215
216@pytest.mark.parametrize(
217 ("credentials", "account_id", "differs"),
218 [
219 # the same account: the credential is accepted
220 ('{"username": "u1", "auth_data": "blob"}', "u1", False),
221 # a Spotify app logged in as someone else
222 ('{"username": "u2", "auth_data": "blob"}', "u1", True),
223 # a near miss is still another account
224 ('{"username": "u10", "auth_data": "blob"}', "u1", True),
225 # the canonical username Spotify hands librespot is the lowercased account id
226 ('{"username": "u1", "auth_data": "blob"}', "U1", False),
227 # non-ASCII usernames are stored percent-encoded; still the same account
228 ('{"username": "us%C3%A9rnam%C3%A9", "auth_data": "blob"}', "usérnamé", False),
229 # and a percent-encoded name that decodes to someone else is still spotted
230 ('{"username": "s%C3%B6meone_else", "auth_data": "blob"}', "usérnamé", True),
231 # either side unknown, or an unreadable credential: never block the setup
232 ('{"username": "u2", "auth_data": "blob"}', None, False),
233 ('{"auth_data": "blob"}', "u1", False),
234 ('{"username": null, "auth_data": "blob"}', "u1", False),
235 ('{"username": "", "auth_data": "blob"}', "u1", False),
236 ("not json", "u1", False),
237 ("[]", "u1", False),
238 ],
239)
240def test_credential_account_comparison(
241 credentials: str, account_id: str | None, differs: bool
242) -> None:
243 """A playback credential from another Spotify account is spotted, and only that."""
244 from music_assistant.providers.spotify.setup_flow import ( # noqa: PLC0415
245 _credential_account_differs,
246 )
247
248 assert _credential_account_differs(credentials, account_id) is differs
249
250
251async def test_playback_authorized_with_another_account_loops_back(
252 monkeypatch: pytest.MonkeyPatch,
253) -> None:
254 """Pairing with the wrong Spotify account re-shows the step instead of storing it."""
255 from music_assistant.providers.spotify import setup_flow # noqa: PLC0415
256
257 monkeypatch.setattr(
258 setup_flow, "get_librespot_binary", AsyncMock(return_value="/bin/librespot")
259 )
260 # first attempt pairs the wrong account, second one gets it right
261 pairing_mock = MagicMock(
262 side_effect=[
263 _returning('{"username": "someone_else", "auth_data": "blob"}'),
264 _returning('{"username": "u1", "auth_data": "blob"}'),
265 ]
266 )
267 monkeypatch.setattr(setup_flow, "librespot_credentials_via_pairing", pairing_mock)
268 session = MagicMock()
269 session.form = AsyncMock(return_value={setup_flow.CONF_PLAYBACK_AUTH_METHOD: "spotify_app"})
270 session.progress_until = AsyncMock(side_effect=_run_awaitable)
271
272 result = await setup_flow._authorize_playback(session, "u1")
273
274 assert result == '{"username": "u1", "auth_data": "blob"}'
275 # the mismatch re-showed the method step carrying the reason
276 assert session.form.await_count == 2
277 assert session.form.await_args_list[1].kwargs["errors"] == {"base": "playback_account_mismatch"}
278