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