/
/
/
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("music_assistant.providers.spotify.provider.get_spotify_token", token_call)
79 await prov.login()
80 # the token sent to Spotify must come from the persisted store
81 assert token_call.await_args is not None
82 assert token_call.await_args.args[2] == "fresh_token"
83
84
85async def test_login_keeps_token_when_rotated_in_flight(monkeypatch: pytest.MonkeyPatch) -> None:
86 """A revoked error is ignored when a newer token was persisted during the refresh."""
87 prov = _make_provider(stored_token=USED_TOKEN)
88 # the initial read uses token_a; the superseded re-check sees a newer token_b that was
89 # persisted while the refresh was in flight
90 cast("MagicMock", prov.mass.config).get = MagicMock(
91 side_effect=[
92 {CONF_REFRESH_TOKEN_GLOBAL: USED_TOKEN},
93 {CONF_REFRESH_TOKEN_GLOBAL: "token_b"},
94 ]
95 )
96 update_setup_data = MagicMock()
97 unload = MagicMock()
98 monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
99 monkeypatch.setattr(prov, "unload_with_error", unload)
100 monkeypatch.setattr(
101 "music_assistant.providers.spotify.provider.get_spotify_token",
102 AsyncMock(side_effect=LoginFailed("invalid_grant: Refresh token revoked")),
103 )
104 with pytest.raises(LoginFailed):
105 await prov.login()
106 update_setup_data.assert_not_called()
107 unload.assert_not_called()
108
109
110async def test_login_returns_cached_token_while_valid(monkeypatch: pytest.MonkeyPatch) -> None:
111 """A cached access token that is still valid is returned without contacting Spotify."""
112 prov = _make_provider(stored_token=USED_TOKEN)
113 cached = {
114 "access_token": "cached",
115 "refresh_token": USED_TOKEN,
116 "expires_at": time.time() + 3600,
117 }
118 prov._auth_info_global = cached
119 token_call = AsyncMock()
120 monkeypatch.setattr("music_assistant.providers.spotify.provider.get_spotify_token", token_call)
121 assert await prov.login() is cached
122 token_call.assert_not_awaited()
123
124
125async def test_login_refreshes_when_cached_token_expired(monkeypatch: pytest.MonkeyPatch) -> None:
126 """An expired cached access token triggers a refresh instead of being served."""
127 prov = _make_provider(stored_token=USED_TOKEN)
128 prov._sp_user = {"display_name": "tester"}
129 prov._auth_info_global = {
130 "access_token": "old",
131 "refresh_token": USED_TOKEN,
132 "expires_at": time.time() - 10,
133 }
134 token_call = AsyncMock(
135 return_value={
136 "access_token": "new",
137 "refresh_token": USED_TOKEN,
138 "expires_at": time.time() + 3600,
139 }
140 )
141 monkeypatch.setattr(prov, "_update_setup_data", MagicMock())
142 monkeypatch.setattr("music_assistant.providers.spotify.provider.get_spotify_token", token_call)
143 await prov.login()
144 token_call.assert_awaited_once()
145
146
147async def test_login_wipes_token_on_genuine_revoke(monkeypatch: pytest.MonkeyPatch) -> None:
148 """A revoked error clears the credentials when the stored token is the one we tried."""
149 prov = _make_provider(stored_token=USED_TOKEN)
150 update_setup_data = MagicMock()
151 unload = MagicMock()
152 monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
153 monkeypatch.setattr(prov, "unload_with_error", unload)
154 monkeypatch.setattr(
155 "music_assistant.providers.spotify.provider.get_spotify_token",
156 AsyncMock(side_effect=LoginFailed("invalid_grant: Refresh token revoked")),
157 )
158 with pytest.raises(LoginFailed):
159 await prov.login()
160 update_setup_data.assert_called_once_with(CONF_REFRESH_TOKEN_GLOBAL, None)
161 unload.assert_called_once()
162
163
164async def test_login_persists_rotated_token_immediately(monkeypatch: pytest.MonkeyPatch) -> None:
165 """A rotated refresh token is flushed to disk immediately so it survives a crash."""
166 prov = _make_provider(stored_token=USED_TOKEN)
167 prov._sp_user = {"display_name": "tester"} # already populated -> skip the user-info fetch
168 update_setup_data = MagicMock()
169 monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
170 monkeypatch.setattr(
171 "music_assistant.providers.spotify.provider.get_spotify_token",
172 AsyncMock(
173 return_value={
174 "access_token": "access",
175 "refresh_token": "token_rotated",
176 "expires_at": 9999999999,
177 }
178 ),
179 )
180 await prov.login()
181 update_setup_data.assert_called_once_with(
182 CONF_REFRESH_TOKEN_GLOBAL, "token_rotated", immediate=True
183 )
184
185
186async def test_login_debounces_save_when_token_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
187 """An unchanged refresh token uses the normal debounced save instead of an immediate flush."""
188 prov = _make_provider(stored_token=USED_TOKEN)
189 prov._sp_user = {"display_name": "tester"}
190 update_setup_data = MagicMock()
191 monkeypatch.setattr(prov, "_update_setup_data", update_setup_data)
192 monkeypatch.setattr(
193 "music_assistant.providers.spotify.provider.get_spotify_token",
194 AsyncMock(
195 return_value={
196 "access_token": "access",
197 "refresh_token": USED_TOKEN,
198 "expires_at": 9999999999,
199 }
200 ),
201 )
202 await prov.login()
203 update_setup_data.assert_called_once_with(
204 CONF_REFRESH_TOKEN_GLOBAL, USED_TOKEN, immediate=False
205 )
206
207
208async def test_login_records_the_account_on_a_legacy_config(
209 monkeypatch: pytest.MonkeyPatch,
210) -> None:
211 """A config predating the stored account id gets it filled in on the next login."""
212 prov = _make_provider(stored_token="fresh_token")
213 monkeypatch.setattr(
214 "music_assistant.providers.spotify.provider.get_spotify_token",
215 AsyncMock(
216 return_value={
217 "access_token": "access",
218 "refresh_token": "fresh_token",
219 "expires_at": 9999999999,
220 }
221 ),
222 )
223 monkeypatch.setattr(
224 prov, "_get_data", AsyncMock(return_value={"id": "u1", "display_name": "tester"})
225 )
226 update = MagicMock()
227 monkeypatch.setattr(prov, "_update_setup_data", update)
228 prov.mass.metadata = MagicMock()
229
230 await prov.login()
231
232 # the setup flow can now spot a duplicate account without loading this instance
233 assert (CONF_ACCOUNT_ID, "u1") in [call.args[:2] for call in update.call_args_list]
234
235
236async def test_login_leaves_a_recorded_account_alone(monkeypatch: pytest.MonkeyPatch) -> None:
237 """An account id that is already stored is not rewritten on every login."""
238 prov = _make_provider(stored_token="fresh_token")
239 prov.mass.config.get = MagicMock( # type: ignore[method-assign]
240 return_value={CONF_REFRESH_TOKEN_GLOBAL: "fresh_token", CONF_ACCOUNT_ID: "u1"}
241 )
242 monkeypatch.setattr(
243 "music_assistant.providers.spotify.provider.get_spotify_token",
244 AsyncMock(
245 return_value={
246 "access_token": "access",
247 "refresh_token": "fresh_token",
248 "expires_at": 9999999999,
249 }
250 ),
251 )
252 monkeypatch.setattr(
253 prov, "_get_data", AsyncMock(return_value={"id": "u1", "display_name": "tester"})
254 )
255 update = MagicMock()
256 monkeypatch.setattr(prov, "_update_setup_data", update)
257 prov.mass.metadata = MagicMock()
258
259 await prov.login()
260
261 assert CONF_ACCOUNT_ID not in [call.args[0] for call in update.call_args_list]
262