/
/
/
1"""Unit tests for auth.py (ya-passport-auth cookie login + token maintenance)."""
2
3from __future__ import annotations
4
5import json
6from unittest import mock
7
8import pytest
9from music_assistant_models.errors import (
10 InvalidDataError,
11 LoginFailed,
12 ResourceTemporarilyUnavailable,
13)
14from ya_passport_auth import Credentials, SecretStr
15from ya_passport_auth.exceptions import (
16 InvalidCredentialsError,
17 RateLimitedError,
18 YaPassportError,
19)
20from ya_passport_auth.exceptions import (
21 NetworkError as PassportNetworkError,
22)
23
24# Import via the namespace set up by conftest.py (avoids relative-import issues)
25from music_assistant.providers.yandex_station.auth import (
26 login_with_cookies,
27 refresh_credentials_via_passport,
28 refresh_music_token,
29 validate_x_token,
30)
31
32# mock target prefix: the module as seen in sys.modules
33_MOD = "music_assistant.providers.yandex_station.auth"
34
35
36# -- helpers -------------------------------------------------------------------
37
38
39def _make_credentials(
40 x_token: str = "test_x_token", # noqa: S107
41 music_token: str | None = "test_music_token", # noqa: S107
42 refresh_token: str | None = "test_refresh_token", # noqa: S107
43) -> Credentials:
44 """Build a Credentials dataclass for testing."""
45 return Credentials(
46 x_token=SecretStr(x_token),
47 music_token=SecretStr(music_token) if music_token else None,
48 refresh_token=SecretStr(refresh_token) if refresh_token else None,
49 )
50
51
52# -- refresh_music_token -------------------------------------------------------
53
54
55async def test_refresh_music_token_success() -> None:
56 """Successful refresh returns a SecretStr."""
57 mock_client = mock.AsyncMock()
58 mock_client.refresh_music_token.return_value = SecretStr("new_music_token")
59
60 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
61 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
62 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
63
64 result = await refresh_music_token(SecretStr("my_x_token"))
65
66 assert result.get_secret() == "new_music_token"
67 mock_client.refresh_music_token.assert_awaited_once()
68
69
70async def test_refresh_music_token_auth_error_raises_login_failed() -> None:
71 """Auth failure during refresh is mapped to LoginFailed."""
72 mock_client = mock.AsyncMock()
73 mock_client.refresh_music_token.side_effect = InvalidCredentialsError("bad token")
74
75 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
76 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
77 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
78
79 with pytest.raises(LoginFailed, match="Music token refresh was rejected"):
80 await refresh_music_token(SecretStr("bad_x_token"))
81
82
83@pytest.mark.parametrize(
84 "transient_err",
85 [PassportNetworkError("socket reset"), RateLimitedError("429")],
86)
87async def test_refresh_music_token_transient_raises_provider_unavailable(
88 transient_err: Exception,
89) -> None:
90 """Network/rate-limit failures must NOT be mapped to LoginFailed (would wipe creds)."""
91 mock_client = mock.AsyncMock()
92 mock_client.refresh_music_token.side_effect = transient_err
93
94 with mock.patch(f"{_MOD}.PassportClient.create") 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 with pytest.raises(ResourceTemporarilyUnavailable):
98 await refresh_music_token(SecretStr("x_token"))
99
100
101# -- validate_x_token ----------------------------------------------------------
102
103
104async def test_validate_x_token_valid() -> None:
105 """Valid x_token returns True."""
106 mock_client = mock.AsyncMock()
107 mock_client.validate_x_token.return_value = True
108
109 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
110 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
111 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
112
113 result = await validate_x_token(SecretStr("good_token"))
114
115 assert result is True
116
117
118async def test_validate_x_token_error_returns_false() -> None:
119 """A terminal YaPassportError returns False; transient errors re-raise."""
120 mock_client = mock.AsyncMock()
121 mock_client.validate_x_token.side_effect = YaPassportError("rejected")
122
123 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
124 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
125 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
126
127 result = await validate_x_token(SecretStr("some_token"))
128
129 assert result is False
130
131
132# -- refresh_credentials_via_passport ------------------------------------------
133
134
135async def test_refresh_credentials_via_passport_success() -> None:
136 """Successful refresh returns full Credentials triple."""
137 new_creds = _make_credentials(
138 x_token="new_x",
139 music_token="new_music",
140 refresh_token="new_refresh",
141 )
142 mock_client = mock.AsyncMock()
143 mock_client.refresh_credentials.return_value = new_creds
144
145 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
146 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
147 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
148
149 result = await refresh_credentials_via_passport(
150 SecretStr("old_x"), SecretStr("old_refresh")
151 )
152
153 assert result.x_token.get_secret() == "new_x"
154 assert result.music_token is not None
155 assert result.music_token.get_secret() == "new_music"
156 assert result.refresh_token is not None
157 assert result.refresh_token.get_secret() == "new_refresh"
158 mock_client.refresh_credentials.assert_awaited_once()
159
160
161async def test_refresh_credentials_via_passport_error_raises_login_failed() -> None:
162 """Auth failure during credential refresh is mapped to LoginFailed."""
163 mock_client = mock.AsyncMock()
164 mock_client.refresh_credentials.side_effect = InvalidCredentialsError("dead")
165
166 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
167 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
168 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
169
170 with pytest.raises(LoginFailed, match="Credential refresh was rejected"):
171 await refresh_credentials_via_passport(SecretStr("bad_x"), SecretStr("bad_refresh"))
172
173
174@pytest.mark.parametrize(
175 "transient_err",
176 [PassportNetworkError("socket reset"), RateLimitedError("429")],
177)
178async def test_refresh_credentials_via_passport_transient_raises_provider_unavailable(
179 transient_err: Exception,
180) -> None:
181 """Network/rate-limit failures must NOT be mapped to LoginFailed (would wipe creds)."""
182 mock_client = mock.AsyncMock()
183 mock_client.refresh_credentials.side_effect = transient_err
184
185 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
186 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
187 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
188 with pytest.raises(ResourceTemporarilyUnavailable):
189 await refresh_credentials_via_passport(SecretStr("x"), SecretStr("r"))
190
191
192# -- login_with_cookies --------------------------------------------------------
193
194
195async def test_login_with_cookies_raw_string() -> None:
196 """Raw cookie string auth returns (x_token, music_token)."""
197 creds = _make_credentials(x_token="cookie_x_token", music_token="cookie_music_token")
198
199 mock_client = mock.AsyncMock()
200 mock_client.login_cookies.return_value = creds
201
202 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
203 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
204 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
205
206 x_token, music_token = await login_with_cookies("Session_id=abc123; yandexuid=456")
207
208 assert x_token == "cookie_x_token"
209 assert music_token == "cookie_music_token"
210 mock_client.login_cookies.assert_awaited_once_with("Session_id=abc123; yandexuid=456")
211
212
213async def test_login_with_cookies_json_format() -> None:
214 """JSON cookie array is converted to semicolon string and passed to library."""
215 cookies_json = json.dumps(
216 [
217 {"name": "Session_id", "value": "abc123", "domain": ".yandex.ru"},
218 {"name": "yandexuid", "value": "456", "domain": ".yandex.ru"},
219 ]
220 )
221
222 creds = _make_credentials(x_token="json_x_token", music_token="json_music_token")
223
224 mock_client = mock.AsyncMock()
225 mock_client.login_cookies.return_value = creds
226
227 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
228 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
229 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
230
231 x_token, music_token = await login_with_cookies(cookies_json)
232
233 assert x_token == "json_x_token"
234 assert music_token == "json_music_token"
235 mock_client.login_cookies.assert_awaited_once_with("Session_id=abc123; yandexuid=456")
236
237
238async def test_login_with_cookies_empty_raises() -> None:
239 """Empty cookie string raises InvalidDataError (validation failure)."""
240 with pytest.raises(InvalidDataError, match="Empty cookies"):
241 await login_with_cookies("")
242
243
244async def test_login_with_cookies_invalid_format_raises() -> None:
245 """Cookie string without '=' raises InvalidDataError (validation failure)."""
246 with pytest.raises(InvalidDataError, match="Invalid cookie format"):
247 await login_with_cookies("no_equals_sign_here")
248
249
250async def test_login_with_cookies_auth_error_raises_login_failed() -> None:
251 """InvalidCredentialsError from library is mapped to LoginFailed."""
252 mock_client = mock.AsyncMock()
253 mock_client.login_cookies.side_effect = InvalidCredentialsError("bad cookies")
254
255 with mock.patch(f"{_MOD}.PassportClient.create") as mock_create:
256 mock_create.return_value.__aenter__ = mock.AsyncMock(return_value=mock_client)
257 mock_create.return_value.__aexit__ = mock.AsyncMock(return_value=False)
258
259 with pytest.raises(LoginFailed, match="Cookie authentication"):
260 await login_with_cookies("Session_id=expired; yandexuid=456")
261