/
/
1"""
2Tests for the Spotify provider's refresh-token handling.
3
4The global refresh token is always read from the persisted setup_data, so an in-memory config
5copy that lagged a rotation can never make us refresh with a stale (revoked) token. Spotify
6rotates the refresh token on every refresh and revokes the previous one; if a newer token
7was persisted while a refresh was in flight, the stored (newer) one is kept instead of
8wiping the credentials and forcing re-auth.
9"""
10
11from __future__ import annotations
12
13import time
14from typing import cast
15from unittest.mock import AsyncMock, MagicMock
16
17import pytest
18from music_assistant_models.errors import LoginFailed
19
20from music_assistant.providers.spotify.constants import CONF_ACCOUNT_ID, CONF_REFRESH_TOKEN_GLOBAL
21from music_assistant.providers.spotify.provider import SpotifyProvider
22
23USED_TOKEN = "token_a"
24
25
26def _make_provider(stored_token: str | None) -> SpotifyProvider:
27 """Return a SpotifyProvider (bypassing __init__) with a mocked setup_data store."""
28 prov = object.__new__(SpotifyProvider)
29 # the in-memory config copy has no value for the token; the global refresh token is
30 # read from the persisted setup_data below, not from this object-local copy
31 config = MagicMock(instance_id="spotify--test")
32 config.get_value = MagicMock(return_value=None)
33 config.values = {}
34 prov.config = config
35 prov.manifest = MagicMock(domain="spotify")
36 prov.logger = MagicMock()
37 prov.available = True
38 prov._auth_info_global = None
39
40 setup_data = {CONF_REFRESH_TOKEN_GLOBAL: stored_token} if stored_token is not None else {}
41 mass = MagicMock()
42 # get_setup_value reads the live setup_data blob from the store
43 mass.config.get = MagicMock(return_value=setup_data)
44 mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
45 # the store keeps values encrypted; decrypt is an identity map for the test
46 mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
47 prov.mass = mass
48 return prov
49
50
51def test_refresh_token_superseded_no_stored_token() -> None:
52 """With no stored token there is nothing newer to protect, so it is not superseded."""
53 prov = _make_provider(stored_token=None)
54 assert prov._refresh_token_superseded(CONF_REFRESH_TOKEN_GLOBAL, USED_TOKEN) is False
55
56
57def test_stored_refresh_token_reads_from_setup_data() -> None:
58 """_stored_refresh_token returns the decrypted persisted token, or None when unset."""
59 prov = _make_provider(stored_token="token_x")
60 assert prov._stored_refresh_token(CONF_REFRESH_TOKEN_GLOBAL) == "token_x"
61 assert (
62 _make_provider(stored_token=None)._stored_refresh_token(CONF_REFRESH_TOKEN_GLOBAL) is None
63 )
64
65
66async def test_login_reads_token_from_persisted_store(monkeypatch: pytest.MonkeyPatch) -> None:
67 """The refresh token is read from the persisted store, not a stale in-memory config copy."""
68 prov = _make_provider(stored_token="fresh_token")
69 prov._sp_user = {"display_name": "tester"}
70 token_call = AsyncMock(
71 return_value={
72 "access_token": "access",
73 "refresh_token": "fresh_token",
74 "expires_at": 9999999999,
75 }
76 )
77 monkeypatch.setattr(prov, "_update_setup_data", MagicMock())
78 monkeypatch.setattr(prov, "_setup_librespot_auth", AsyncMock())
79 monkeypatch.setattr("music_assistant.providers.spotify.provider.get_spotify_token", token_call)
80 await prov.login()
81 # the token sent to Spotify must come from the persisted store
82 assert token_call.await_args is not None
83 assert token_call.await_args.args[2] == "fresh_token"
84
85
86async def test_login_keeps_token_when_rotated_in_flight(monkeypatch: pytest.MonkeyPatch) -> None:
87 """A revoked error is ignored when a newer token was persisted during the refresh."""
88 prov = _make_provider(stored_token=USED_TOKEN)
89 # the initial read uses token_a; the superseded re-check sees a newer token_b that was
90 # persisted while the refresh was in flight
91 cast("MagicMock", prov.mass.config).get = MagicMock(
92 side_effect=[
93 {CONF_REFRESH_TOKEN_GLOBAL: USED_TOKEN},
94 {CONF_REFRESH_TOKEN_GLOBAL: "token_b"},
95 ]
96 )
97 update_setup_data = MagicMock()
98 unload = MagicMock()
99 monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
100 monkeypatch.setattr(prov, "unload_with_error", unload)
101 monkeypatch.setattr(
102 "music_assistant.providers.spotify.provider.get_spotify_token",
103 AsyncMock(side_effect=LoginFailed("invalid_grant: Refresh token revoked")),
104 )
105 with pytest.raises(LoginFailed):
106 await prov.login()
107 update_setup_data.assert_not_called()
108 unload.assert_not_called()
109
110
111async def test_login_returns_cached_token_while_valid(monkeypatch: pytest.MonkeyPatch) -> None:
112 """A cached access token that is still valid is returned without contacting Spotify."""
113 prov = _make_provider(stored_token=USED_TOKEN)
114 cached = {
115 "access_token": "cached",
116 "refresh_token": USED_TOKEN,
117 "expires_at": time.time() + 3600,
118 }
119 prov._auth_info_global = cached
120 token_call = AsyncMock()
121 monkeypatch.setattr("music_assistant.providers.spotify.provider.get_spotify_token", token_call)
122 assert await prov.login() is cached
123 token_call.assert_not_awaited()
124
125
126async def test_login_refreshes_when_cached_token_expired(monkeypatch: pytest.MonkeyPatch) -> None:
127 """An expired cached access token triggers a refresh instead of being served."""
128 prov = _make_provider(stored_token=USED_TOKEN)
129 prov._sp_user = {"display_name": "tester"}
130 prov._auth_info_global = {
131 "access_token": "old",
132 "refresh_token": USED_TOKEN,
133 "expires_at": time.time() - 10,
134 }
135 token_call = AsyncMock(
136 return_value={
137 "access_token": "new",
138 "refresh_token": USED_TOKEN,
139 "expires_at": time.time() + 3600,
140 }
141 )
142 monkeypatch.setattr(prov, "_update_setup_data", MagicMock())
143 monkeypatch.setattr(prov, "_setup_librespot_auth", AsyncMock())
144 monkeypatch.setattr("music_assistant.providers.spotify.provider.get_spotify_token", token_call)
145 await prov.login()
146 token_call.assert_awaited_once()
147
148
149async def test_login_wipes_token_on_genuine_revoke(monkeypatch: pytest.MonkeyPatch) -> None:
150 """A revoked error clears the credentials when the stored token is the one we tried."""
151 prov = _make_provider(stored_token=USED_TOKEN)
152 update_setup_data = MagicMock()
153 unload = MagicMock()
154 monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
155 monkeypatch.setattr(prov, "unload_with_error", unload)
156 monkeypatch.setattr(
157 "music_assistant.providers.spotify.provider.get_spotify_token",
158 AsyncMock(side_effect=LoginFailed("invalid_grant: Refresh token revoked")),
159 )
160 with pytest.raises(LoginFailed):
161 await prov.login()
162 update_setup_data.assert_called_once_with(CONF_REFRESH_TOKEN_GLOBAL, None)
163 unload.assert_called_once()
164
165
166async def test_login_persists_rotated_token_immediately(monkeypatch: pytest.MonkeyPatch) -> None:
167 """A rotated refresh token is flushed to disk immediately so it survives a crash."""
168 prov = _make_provider(stored_token=USED_TOKEN)
169 prov._sp_user = {"display_name": "tester"} # already populated -> skip the user-info fetch
170 update_setup_data = MagicMock()
171 monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
172 monkeypatch.setattr(prov, "_setup_librespot_auth", AsyncMock())
173 monkeypatch.setattr(
174 "music_assistant.providers.spotify.provider.get_spotify_token",
175 AsyncMock(
176 return_value={
177 "access_token": "access",
178 "refresh_token": "token_rotated",
179 "expires_at": 9999999999,
180 }
181 ),
182 )
183 await prov.login()
184 update_setup_data.assert_called_once_with(
185 CONF_REFRESH_TOKEN_GLOBAL, "token_rotated", immediate=True
186 )
187
188
189async def test_login_debounces_save_when_token_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
190 """An unchanged refresh token uses the normal debounced save instead of an immediate flush."""
191 prov = _make_provider(stored_token=USED_TOKEN)
192 prov._sp_user = {"display_name": "tester"}
193 update_setup_data = MagicMock()
194 monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
195 monkeypatch.setattr(prov, "_setup_librespot_auth", AsyncMock())
196 monkeypatch.setattr(
197 "music_assistant.providers.spotify.provider.get_spotify_token",
198 AsyncMock(
199 return_value={
200 "access_token": "access",
201 "refresh_token": USED_TOKEN,
202 "expires_at": 9999999999,
203 }
204 ),
205 )
206 await prov.login()
207 update_setup_data.assert_called_once_with(
208 CONF_REFRESH_TOKEN_GLOBAL, USED_TOKEN, immediate=False
209 )
210
211
212async def test_login_records_the_account_on_a_legacy_config(
213 monkeypatch: pytest.MonkeyPatch,
214) -> None:
215 """A config predating the stored account id gets it filled in on the next login."""
216 prov = _make_provider(stored_token="fresh_token")
217 monkeypatch.setattr(
218 "music_assistant.providers.spotify.provider.get_spotify_token",
219 AsyncMock(
220 return_value={
221 "access_token": "access",
222 "refresh_token": "fresh_token",
223 "expires_at": 9999999999,
224 }
225 ),
226 )
227 monkeypatch.setattr(
228 prov, "_get_data", AsyncMock(return_value={"id": "u1", "display_name": "tester"})
229 )
230 update = MagicMock()
231 monkeypatch.setattr(prov, "_update_setup_data", update)
232 prov.mass.metadata = MagicMock()
233
234 await prov.login()
235
236 # the setup flow can now spot a duplicate account without loading this instance
237 assert (CONF_ACCOUNT_ID, "u1") in [call.args[:2] for call in update.call_args_list]
238
239
240async def test_login_leaves_a_recorded_account_alone(monkeypatch: pytest.MonkeyPatch) -> None:
241 """An account id that is already stored is not rewritten on every login."""
242 prov = _make_provider(stored_token="fresh_token")
243 prov.mass.config.get = MagicMock( # type: ignore[method-assign]
244 return_value={CONF_REFRESH_TOKEN_GLOBAL: "fresh_token", CONF_ACCOUNT_ID: "u1"}
245 )
246 monkeypatch.setattr(
247 "music_assistant.providers.spotify.provider.get_spotify_token",
248 AsyncMock(
249 return_value={
250 "access_token": "access",
251 "refresh_token": "fresh_token",
252 "expires_at": 9999999999,
253 }
254 ),
255 )
256 monkeypatch.setattr(
257 prov, "_get_data", AsyncMock(return_value={"id": "u1", "display_name": "tester"})
258 )
259 update = MagicMock()
260 monkeypatch.setattr(prov, "_update_setup_data", update)
261 prov.mass.metadata = MagicMock()
262
263 await prov.login()
264
265 assert CONF_ACCOUNT_ID not in [call.args[0] for call in update.call_args_list]
266