/
/
/
1"""Unit tests for Yandex Music 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 NetworkError as PassportNetworkError
12
13from music_assistant.providers.yandex_music.auth import (
14 refresh_credentials_via_passport,
15 refresh_music_token,
16 validate_x_token,
17)
18
19
20def _make_credentials(
21 x_token: str = "test_x_token", # noqa: S107
22 music_token: str | None = "test_music_token", # noqa: S107
23 refresh_token: str | None = "test_refresh_token", # noqa: S107
24) -> Credentials:
25 """Build credentials for token-maintenance tests."""
26 return Credentials(
27 x_token=SecretStr(x_token),
28 music_token=SecretStr(music_token) if music_token else None,
29 refresh_token=SecretStr(refresh_token) if refresh_token else None,
30 )
31
32
33async def test_refresh_music_token_success() -> None:
34 """Successful refresh returns a SecretStr."""
35 mock_client = mock.AsyncMock()
36 mock_client.refresh_music_token.return_value = SecretStr("new_music_token")
37
38 with mock.patch(
39 "music_assistant.providers.yandex_music.auth.PassportClient.create",
40 ) as mock_create:
41 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
42 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
43
44 result = await refresh_music_token(SecretStr("my_x_token"))
45
46 assert result.get_secret() == "new_music_token"
47 mock_client.refresh_music_token.assert_awaited_once()
48
49
50async def test_refresh_music_token_auth_error_raises_login_failed() -> None:
51 """Auth failure during refresh is mapped to LoginFailed."""
52 mock_client = mock.AsyncMock()
53 mock_client.refresh_music_token.side_effect = InvalidCredentialsError("bad token")
54
55 with mock.patch(
56 "music_assistant.providers.yandex_music.auth.PassportClient.create",
57 ) as mock_create:
58 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
59 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
60
61 with pytest.raises(LoginFailed, match="Music token refresh was rejected"):
62 await refresh_music_token(SecretStr("bad_x_token"))
63
64
65@pytest.mark.parametrize(
66 "exc",
67 [PassportNetworkError("offline"), RateLimitedError("429")],
68 ids=["network", "rate_limited"],
69)
70async def test_refresh_music_token_transient_error_raises_temporarily_unavailable(
71 exc: Exception,
72) -> None:
73 """Transient Passport failures don't masquerade as LoginFailed."""
74 mock_client = mock.AsyncMock()
75 mock_client.refresh_music_token.side_effect = exc
76
77 with mock.patch(
78 "music_assistant.providers.yandex_music.auth.PassportClient.create",
79 ) as mock_create:
80 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
81 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
82
83 with pytest.raises(ResourceTemporarilyUnavailable, match="temporarily unavailable"):
84 await refresh_music_token(SecretStr("my_x_token"))
85
86
87async def test_validate_x_token_valid() -> None:
88 """Valid x_token returns True."""
89 mock_client = mock.AsyncMock()
90 mock_client.validate_x_token.return_value = True
91
92 with mock.patch(
93 "music_assistant.providers.yandex_music.auth.PassportClient.create",
94 ) as mock_create:
95 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
96 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
97
98 result = await validate_x_token(SecretStr("good_token"))
99
100 assert result is True
101
102
103async def test_validate_x_token_invalid_returns_false() -> None:
104 """A terminal credential error returns False."""
105 mock_client = mock.AsyncMock()
106 mock_client.validate_x_token.side_effect = InvalidCredentialsError("token rejected")
107
108 with mock.patch(
109 "music_assistant.providers.yandex_music.auth.PassportClient.create",
110 ) as mock_create:
111 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
112 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
113
114 result = await validate_x_token(SecretStr("some_token"))
115
116 assert result is False
117
118
119@pytest.mark.parametrize(
120 "exc",
121 [PassportNetworkError("offline"), RateLimitedError("429")],
122 ids=["network", "rate_limited"],
123)
124async def test_validate_x_token_transient_error_propagates(exc: Exception) -> None:
125 """Transient failures are distinguishable from invalid credentials."""
126 mock_client = mock.AsyncMock()
127 mock_client.validate_x_token.side_effect = exc
128
129 with mock.patch(
130 "music_assistant.providers.yandex_music.auth.PassportClient.create",
131 ) as mock_create:
132 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
133 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
134
135 with pytest.raises((PassportNetworkError, RateLimitedError)):
136 await validate_x_token(SecretStr("some_token"))
137
138
139async def test_refresh_credentials_via_passport_success() -> None:
140 """Successful refresh returns the full credential triple."""
141 new_creds = _make_credentials(
142 x_token="new_x",
143 music_token="new_music",
144 refresh_token="new_refresh",
145 )
146 mock_client = mock.AsyncMock()
147 mock_client.refresh_credentials.return_value = new_creds
148
149 with mock.patch(
150 "music_assistant.providers.yandex_music.auth.PassportClient.create",
151 ) as mock_create:
152 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
153 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
154
155 result = await refresh_credentials_via_passport(
156 SecretStr("old_x"), SecretStr("old_refresh")
157 )
158
159 assert result.x_token.get_secret() == "new_x"
160 assert result.music_token is not None
161 assert result.music_token.get_secret() == "new_music"
162 assert result.refresh_token is not None
163 assert result.refresh_token.get_secret() == "new_refresh"
164 mock_client.refresh_credentials.assert_awaited_once()
165
166
167async def test_refresh_credentials_via_passport_error_raises_login_failed() -> None:
168 """Auth failure during credential refresh is mapped to LoginFailed."""
169 mock_client = mock.AsyncMock()
170 mock_client.refresh_credentials.side_effect = InvalidCredentialsError("dead")
171
172 with mock.patch(
173 "music_assistant.providers.yandex_music.auth.PassportClient.create",
174 ) as mock_create:
175 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
176 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
177
178 with pytest.raises(LoginFailed, match="Credential refresh was rejected"):
179 await refresh_credentials_via_passport(SecretStr("bad_x"), SecretStr("bad_refresh"))
180
181
182@pytest.mark.parametrize(
183 "exc",
184 [PassportNetworkError("offline"), RateLimitedError("429")],
185 ids=["network", "rate_limited"],
186)
187async def test_refresh_credentials_via_passport_transient_error_is_temporary(
188 exc: Exception,
189) -> None:
190 """Transient credential refresh failures remain retryable."""
191 mock_client = mock.AsyncMock()
192 mock_client.refresh_credentials.side_effect = exc
193
194 with mock.patch(
195 "music_assistant.providers.yandex_music.auth.PassportClient.create",
196 ) as mock_create:
197 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
198 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
199
200 with pytest.raises(ResourceTemporarilyUnavailable, match="temporarily unavailable"):
201 await refresh_credentials_via_passport(SecretStr("x"), SecretStr("refresh"))
202
203
204_SECRET_PAYLOAD = "token=ABC_TOKEN_LEAK&csrf=xyz"
205
206
207@pytest.mark.parametrize(
208 ("exc", "expected_exc_type"),
209 [
210 (PassportNetworkError(_SECRET_PAYLOAD), ResourceTemporarilyUnavailable),
211 (RateLimitedError(_SECRET_PAYLOAD), ResourceTemporarilyUnavailable),
212 (InvalidCredentialsError(_SECRET_PAYLOAD), LoginFailed),
213 ],
214 ids=["network", "rate_limited", "invalid_credentials"],
215)
216async def test_refresh_music_token_error_does_not_leak_library_payload(
217 exc: Exception, expected_exc_type: type[Exception]
218) -> None:
219 """Music-token refresh errors redact upstream payloads."""
220 mock_client = mock.AsyncMock()
221 mock_client.refresh_music_token.side_effect = exc
222
223 with mock.patch(
224 "music_assistant.providers.yandex_music.auth.PassportClient.create",
225 ) as mock_create:
226 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
227 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
228
229 with pytest.raises(expected_exc_type) as exc_info:
230 await refresh_music_token(SecretStr("my_x_token"))
231
232 assert _SECRET_PAYLOAD not in str(exc_info.value)
233 assert "ABC_TOKEN_LEAK" not in str(exc_info.value)
234
235
236@pytest.mark.parametrize(
237 ("exc", "expected_exc_type"),
238 [
239 (PassportNetworkError(_SECRET_PAYLOAD), ResourceTemporarilyUnavailable),
240 (RateLimitedError(_SECRET_PAYLOAD), ResourceTemporarilyUnavailable),
241 (InvalidCredentialsError(_SECRET_PAYLOAD), LoginFailed),
242 ],
243 ids=["network", "rate_limited", "invalid_credentials"],
244)
245async def test_refresh_credentials_error_does_not_leak_library_payload(
246 exc: Exception, expected_exc_type: type[Exception]
247) -> None:
248 """Credential refresh errors redact upstream payloads."""
249 mock_client = mock.AsyncMock()
250 mock_client.refresh_credentials.side_effect = exc
251
252 with mock.patch(
253 "music_assistant.providers.yandex_music.auth.PassportClient.create",
254 ) as mock_create:
255 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
256 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
257
258 with pytest.raises(expected_exc_type) as exc_info:
259 await refresh_credentials_via_passport(SecretStr("x"), SecretStr("refresh"))
260
261 assert _SECRET_PAYLOAD not in str(exc_info.value)
262 assert "ABC_TOKEN_LEAK" not in str(exc_info.value)
263