/
/
/
1"""Unit tests for auth.py (ya-passport-auth token maintenance)."""
2
3from __future__ import annotations
4
5from unittest import mock
6
7import pytest
8from music_assistant_models.errors import LoginFailed, ResourceTemporarilyUnavailable
9from ya_passport_auth import Credentials, SecretStr
10from ya_passport_auth.exceptions import InvalidCredentialsError, RateLimitedError
11from ya_passport_auth.exceptions import (
12 NetworkError as PassportNetworkError,
13)
14
15from music_assistant.providers.yandex_music.auth import (
16 refresh_credentials_via_passport,
17 refresh_music_token,
18 validate_x_token,
19)
20
21# -- helpers -------------------------------------------------------------------
22
23
24def _make_credentials(
25 x_token: str = "test_x_token", # noqa: S107
26 music_token: str | None = "test_music_token", # noqa: S107
27 refresh_token: str | None = "test_refresh_token", # noqa: S107
28) -> Credentials:
29 """Build a Credentials dataclass for testing."""
30 return Credentials(
31 x_token=SecretStr(x_token),
32 music_token=SecretStr(music_token) if music_token else None,
33 refresh_token=SecretStr(refresh_token) if refresh_token else None,
34 )
35
36
37# -- refresh_music_token -------------------------------------------------------
38
39
40async def test_refresh_music_token_success() -> None:
41 """Successful refresh returns a SecretStr."""
42 mock_client = mock.AsyncMock()
43 mock_client.refresh_music_token.return_value = SecretStr("new_music_token")
44
45 with mock.patch(
46 "music_assistant.providers.yandex_music.auth.PassportClient.create",
47 ) as mock_create:
48 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
49 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
50
51 result = await refresh_music_token(SecretStr("my_x_token"))
52
53 assert result.get_secret() == "new_music_token"
54 mock_client.refresh_music_token.assert_awaited_once()
55
56
57async def test_refresh_music_token_auth_error_raises_login_failed() -> None:
58 """Auth failure during refresh is mapped to LoginFailed."""
59 mock_client = mock.AsyncMock()
60 mock_client.refresh_music_token.side_effect = InvalidCredentialsError("bad token")
61
62 with mock.patch(
63 "music_assistant.providers.yandex_music.auth.PassportClient.create",
64 ) as mock_create:
65 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
66 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
67
68 with pytest.raises(LoginFailed, match="Music token refresh was rejected"):
69 await refresh_music_token(SecretStr("bad_x_token"))
70
71
72@pytest.mark.parametrize(
73 "exc",
74 [PassportNetworkError("offline"), RateLimitedError("429")],
75 ids=["network", "rate_limited"],
76)
77async def test_refresh_music_token_transient_error_raises_temporarily_unavailable(
78 exc: Exception,
79) -> None:
80 """Transient Passport failures don't masquerade as LoginFailed."""
81 mock_client = mock.AsyncMock()
82 mock_client.refresh_music_token.side_effect = exc
83
84 with mock.patch(
85 "music_assistant.providers.yandex_music.auth.PassportClient.create",
86 ) as mock_create:
87 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
88 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
89
90 with pytest.raises(ResourceTemporarilyUnavailable, match="temporarily unavailable"):
91 await refresh_music_token(SecretStr("my_x_token"))
92
93
94# -- validate_x_token ----------------------------------------------------------
95
96
97async def test_validate_x_token_valid() -> None:
98 """Valid x_token returns True."""
99 mock_client = mock.AsyncMock()
100 mock_client.validate_x_token.return_value = True
101
102 with mock.patch(
103 "music_assistant.providers.yandex_music.auth.PassportClient.create",
104 ) as mock_create:
105 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
106 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
107
108 result = await validate_x_token(SecretStr("good_token"))
109
110 assert result is True
111
112
113async def test_validate_x_token_invalid_returns_false() -> None:
114 """A terminal credential error returns False (token rejected by Passport)."""
115 mock_client = mock.AsyncMock()
116 mock_client.validate_x_token.side_effect = InvalidCredentialsError("token rejected")
117
118 with mock.patch(
119 "music_assistant.providers.yandex_music.auth.PassportClient.create",
120 ) as mock_create:
121 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
122 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
123
124 result = await validate_x_token(SecretStr("some_token"))
125
126 assert result is False
127
128
129@pytest.mark.parametrize(
130 "exc",
131 [PassportNetworkError("offline"), RateLimitedError("429")],
132 ids=["network", "rate_limited"],
133)
134async def test_validate_x_token_transient_error_propagates(exc: Exception) -> None:
135 """
136 Transient Passport failures must not masquerade as "token invalid".
137
138 A network blip or 429 should not cause callers to clear the stored
139 credential â re-raise so the caller can distinguish the two.
140 """
141 mock_client = mock.AsyncMock()
142 mock_client.validate_x_token.side_effect = exc
143
144 with mock.patch(
145 "music_assistant.providers.yandex_music.auth.PassportClient.create",
146 ) as mock_create:
147 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
148 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
149
150 with pytest.raises((PassportNetworkError, RateLimitedError)):
151 await validate_x_token(SecretStr("some_token"))
152
153
154# -- refresh_credentials_via_passport ------------------------------------------
155
156
157async def test_refresh_credentials_via_passport_success() -> None:
158 """Successful refresh returns full Credentials triple."""
159 new_creds = _make_credentials(
160 x_token="new_x",
161 music_token="new_music",
162 refresh_token="new_refresh",
163 )
164 mock_client = mock.AsyncMock()
165 mock_client.refresh_credentials.return_value = new_creds
166
167 with mock.patch(
168 "music_assistant.providers.yandex_music.auth.PassportClient.create",
169 ) as mock_create:
170 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
171 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
172
173 result = await refresh_credentials_via_passport(
174 SecretStr("old_x"), SecretStr("old_refresh")
175 )
176
177 assert result.x_token.get_secret() == "new_x"
178 assert result.music_token is not None
179 assert result.music_token.get_secret() == "new_music"
180 assert result.refresh_token is not None
181 assert result.refresh_token.get_secret() == "new_refresh"
182 mock_client.refresh_credentials.assert_awaited_once()
183
184
185async def test_refresh_credentials_via_passport_error_raises_login_failed() -> None:
186 """Auth failure during credential refresh is mapped to LoginFailed."""
187 mock_client = mock.AsyncMock()
188 mock_client.refresh_credentials.side_effect = InvalidCredentialsError("dead")
189
190 with mock.patch(
191 "music_assistant.providers.yandex_music.auth.PassportClient.create",
192 ) as mock_create:
193 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
194 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
195
196 with pytest.raises(LoginFailed, match="Credential refresh was rejected"):
197 await refresh_credentials_via_passport(SecretStr("bad_x"), SecretStr("bad_refresh"))
198
199
200@pytest.mark.parametrize(
201 "exc",
202 [PassportNetworkError("offline"), RateLimitedError("429")],
203 ids=["network", "rate_limited"],
204)
205async def test_refresh_credentials_via_passport_transient_error_raises_temporarily_unavailable(
206 exc: Exception,
207) -> None:
208 """Transient Passport failures don't masquerade as LoginFailed."""
209 mock_client = mock.AsyncMock()
210 mock_client.refresh_credentials.side_effect = exc
211
212 with mock.patch(
213 "music_assistant.providers.yandex_music.auth.PassportClient.create",
214 ) as mock_create:
215 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
216 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
217
218 with pytest.raises(ResourceTemporarilyUnavailable, match="temporarily unavailable"):
219 await refresh_credentials_via_passport(SecretStr("x"), SecretStr("refresh"))
220
221
222# -- exception-message redaction ----------------------------------------------
223#
224# These tests guard against leaking the upstream ``ya-passport-auth`` exception
225# payload into our own ``LoginFailed`` / ``ResourceTemporarilyUnavailable``
226# messages. The library may include token fragments, device codes, or raw
227# response bodies in its exception text â none of which should reach MA logs
228# or the frontend.
229
230_SECRET_PAYLOAD = "token=ABC_TOKEN_LEAK&csrf=xyz"
231
232
233@pytest.mark.parametrize(
234 ("exc", "expected_exc_type"),
235 [
236 (PassportNetworkError(_SECRET_PAYLOAD), ResourceTemporarilyUnavailable),
237 (RateLimitedError(_SECRET_PAYLOAD), ResourceTemporarilyUnavailable),
238 (InvalidCredentialsError(_SECRET_PAYLOAD), LoginFailed),
239 ],
240 ids=["network", "rate_limited", "invalid_credentials"],
241)
242async def test_refresh_music_token_error_does_not_leak_library_payload(
243 exc: Exception, expected_exc_type: type[Exception]
244) -> None:
245 """``refresh_music_token`` exceptions must not include library str()."""
246 mock_client = mock.AsyncMock()
247 mock_client.refresh_music_token.side_effect = exc
248
249 with mock.patch(
250 "music_assistant.providers.yandex_music.auth.PassportClient.create",
251 ) as mock_create:
252 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
253 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
254
255 with pytest.raises(expected_exc_type) as exc_info:
256 await refresh_music_token(SecretStr("my_x_token"))
257
258 assert _SECRET_PAYLOAD not in str(exc_info.value)
259 assert "ABC_TOKEN_LEAK" not in str(exc_info.value)
260
261
262@pytest.mark.parametrize(
263 ("exc", "expected_exc_type"),
264 [
265 (PassportNetworkError(_SECRET_PAYLOAD), ResourceTemporarilyUnavailable),
266 (RateLimitedError(_SECRET_PAYLOAD), ResourceTemporarilyUnavailable),
267 (InvalidCredentialsError(_SECRET_PAYLOAD), LoginFailed),
268 ],
269 ids=["network", "rate_limited", "invalid_credentials"],
270)
271async def test_refresh_credentials_via_passport_error_does_not_leak_library_payload(
272 exc: Exception, expected_exc_type: type[Exception]
273) -> None:
274 """``refresh_credentials_via_passport`` exceptions must not include library str()."""
275 mock_client = mock.AsyncMock()
276 mock_client.refresh_credentials.side_effect = exc
277
278 with mock.patch(
279 "music_assistant.providers.yandex_music.auth.PassportClient.create",
280 ) as mock_create:
281 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
282 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
283
284 with pytest.raises(expected_exc_type) as exc_info:
285 await refresh_credentials_via_passport(SecretStr("x"), SecretStr("refresh"))
286
287 assert _SECRET_PAYLOAD not in str(exc_info.value)
288 assert "ABC_TOKEN_LEAK" not in str(exc_info.value)
289