/
/
/
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
13from pathlib import Path
14from typing import Any
15from unittest.mock import AsyncMock, MagicMock
16
17import pytest
18from music_assistant_models.errors import LoginFailed
19
20from music_assistant.controllers.config.helpers import _AUTH_ERROR_CODES
21from music_assistant.helpers.oauth import authorization_code_from_url
22from music_assistant.models.setup_flow import SetupFlowError
23from music_assistant.providers.spotify.constants import (
24 CONF_LIBRESPOT_CREDENTIALS,
25 CREDENTIALS_FILE,
26)
27from music_assistant.providers.spotify.provider import SpotifyProvider
28
29STORED_CREDENTIALS = '{"username": "tester", "auth_type": 1, "auth_data": "blob"}'
30
31
32def _make_provider(credentials: str | None, cache_dir: str) -> SpotifyProvider:
33 """Return a SpotifyProvider (bypassing __init__) with the given stored credential."""
34 prov = object.__new__(SpotifyProvider)
35 config = MagicMock(instance_id="spotify--test")
36 config.get_value = MagicMock(return_value=None)
37 config.values = {}
38 prov.config = config
39 prov.manifest = MagicMock(domain="spotify")
40 prov.logger = MagicMock()
41 prov.available = True
42 prov.cache_dir = cache_dir
43 prov._librespot_bin = "/bin/librespot"
44 setup_data = {CONF_LIBRESPOT_CREDENTIALS: credentials} if credentials is not None else {}
45 mass = MagicMock()
46 # get_setup_value reads the live setup_data blob from the store
47 mass.config.get = MagicMock(return_value=setup_data)
48 mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
49 # the store keeps values encrypted; decrypt is an identity map for the test
50 mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
51 prov.mass = mass
52 return prov
53
54
55async def test_stored_credential_is_installed_for_librespot(
56 tmp_path: Path,
57) -> None:
58 """The stored credential is written to librespot's cache so login5 accepts it."""
59 cache_dir = tmp_path / "cache"
60 prov = _make_provider(STORED_CREDENTIALS, str(cache_dir))
61 await prov._setup_librespot_auth()
62 written = json.loads((cache_dir / CREDENTIALS_FILE).read_text(encoding="utf-8"))
63 assert written["auth_data"] == "blob"
64
65
66async def test_stale_cached_credential_is_replaced(tmp_path: Path) -> None:
67 """A credential left in the cache from an earlier (now rejected) mint is overwritten."""
68 cache_dir = tmp_path / "cache"
69 cache_dir.mkdir()
70 credentials_file = cache_dir / CREDENTIALS_FILE
71 credentials_file.write_text('{"username": "tester", "auth_data": "stale"}', encoding="utf-8")
72 prov = _make_provider(STORED_CREDENTIALS, str(cache_dir))
73 await prov._setup_librespot_auth()
74 written = json.loads(credentials_file.read_text(encoding="utf-8"))
75 assert written["auth_data"] == "blob"
76
77
78async def test_missing_credential_requires_reauth(tmp_path: Path) -> None:
79 """Without a stored credential the provider fails with an auth error (AUTH_REQUIRED)."""
80 prov = _make_provider(None, str(tmp_path / "cache"))
81 with pytest.raises(LoginFailed) as err:
82 await prov._setup_librespot_auth()
83 # the error code is what actually drives the provider to AUTH_REQUIRED (and so the
84 # reconfigure prompt); the translation key is what the user reads
85 assert err.value.error_code in _AUTH_ERROR_CODES
86 assert err.value.translation_key == "playback_auth_required"
87
88
89async def test_failed_attempt_loops_back_to_the_choice(monkeypatch: pytest.MonkeyPatch) -> None:
90 """A failed attempt re-offers the choice instead of aborting the already-authorized flow."""
91 from music_assistant.providers.spotify import setup_flow # noqa: PLC0415
92
93 monkeypatch.setattr(
94 setup_flow, "get_librespot_binary", AsyncMock(return_value="/bin/librespot")
95 )
96 # first attempt fails the way a rejected token does, second one succeeds
97 monkeypatch.setattr(
98 setup_flow,
99 "librespot_credentials_via_pairing",
100 MagicMock(side_effect=[_raising(LoginFailed("nope")), _returning(STORED_CREDENTIALS)]),
101 )
102 session = MagicMock()
103 session.form = AsyncMock(return_value={setup_flow.CONF_PLAYBACK_AUTH_METHOD: "spotify_app"})
104 session.progress_until = AsyncMock(side_effect=_run_awaitable)
105
106 assert await setup_flow._authorize_playback(session) == STORED_CREDENTIALS
107 # the form was re-shown, carrying the failure reason rather than aborting the flow
108 assert session.form.await_count == 2
109 assert session.form.await_args_list[1].kwargs["errors"] == {"base": "playback_auth_failed"}
110
111
112async def _run_awaitable(awaitable: Any, **_kwargs: Any) -> Any:
113 """Stand in for session.progress_until, which awaits the work it displays progress for."""
114 return await awaitable
115
116
117async def _raising(err: Exception) -> str:
118 """Return a coroutine that raises, standing in for a failed credential attempt."""
119 raise err
120
121
122async def _returning(value: str) -> str:
123 """Return a coroutine resolving to the given credential."""
124 return value
125
126
127@pytest.mark.parametrize(
128 ("url", "expected"),
129 [
130 ("http://127.0.0.1:5588/login?code=abc123", "abc123"),
131 (" http://127.0.0.1:5588/login?code=abc123&state=x ", "abc123"),
132 ("https://127.0.0.1:5588/login?state=x&code=abc123", "abc123"),
133 ],
134)
135def test_authorization_code_from_url(url: str, expected: str) -> None:
136 """The code is recovered from the (dead) loopback URL the user pastes back."""
137 assert authorization_code_from_url(url) == expected
138
139
140@pytest.mark.parametrize(
141 "url",
142 [
143 "http://127.0.0.1:5588/login?error=access_denied",
144 "http://127.0.0.1:5588/login?code=null",
145 "not a url at all",
146 "",
147 ],
148)
149def test_authorization_code_from_url_rejects_unusable(url: str) -> None:
150 """A denied, empty or malformed paste is reported instead of silently proceeding."""
151 with pytest.raises(SetupFlowError):
152 authorization_code_from_url(url)
153