/
/
/
1"""Tests for the Yandex Music interactive setup flow (run_setup)."""
2
3from __future__ import annotations
4
5import asyncio
6import base64
7import time
8from collections.abc import Awaitable, Callable
9from contextlib import suppress
10from types import TracebackType
11from typing import Any
12from unittest import mock
13from urllib.parse import unquote
14
15import pytest
16from music_assistant_models.enums import ConfigEntryType, FlowStepType
17from ya_passport_auth import Credentials, DeviceCodeSession, QrSession, SecretStr
18from ya_passport_auth.exceptions import DeviceCodeTimeoutError, QRTimeoutError
19
20from music_assistant.models.setup_flow import SetupFlowContext, SetupSession, StepExpiredError
21from music_assistant.providers.yandex_music import setup_flow as ym_flow
22from music_assistant.providers.yandex_music.constants import (
23 CONF_REFRESH_TOKEN,
24 CONF_REMEMBER_SESSION,
25 CONF_TOKEN,
26 CONF_X_TOKEN,
27)
28
29
30class _FakeClient:
31 """Canned PassportClient that confirms a QR/device login (optionally after one expiry)."""
32
33 def __init__(
34 self,
35 creds: Credentials,
36 *,
37 qr_fail_first: bool = False,
38 device_fail_first: bool = False,
39 ) -> None:
40 self._creds = creds
41 self.qr_starts = 0
42 self._qr_polls = 0
43 self._qr_fail_first = qr_fail_first
44 self.device_starts = 0
45 self._device_polls = 0
46 self._device_fail_first = device_fail_first
47
48 async def start_qr_login(self) -> QrSession:
49 self.qr_starts += 1
50 return QrSession(track_id="t", csrf_token="c", qr_url="https://passport.yandex.ru/qr/abc")
51
52 async def poll_qr_until_confirmed(self, _qr: QrSession, **_kwargs: Any) -> Credentials:
53 self._qr_polls += 1
54 if self._qr_fail_first and self._qr_polls == 1:
55 raise QRTimeoutError("expired")
56 return self._creds
57
58 async def start_device_login(self, **_kwargs: Any) -> DeviceCodeSession:
59 self.device_starts += 1
60 return DeviceCodeSession(
61 device_code=SecretStr("dc"),
62 user_code="ABCD-1234",
63 verification_url="https://ya.ru/device",
64 expires_in=300,
65 interval=5,
66 )
67
68 async def poll_device_until_confirmed(
69 self, _session: DeviceCodeSession, **_kwargs: Any
70 ) -> Credentials:
71 self._device_polls += 1
72 if self._device_fail_first and self._device_polls == 1:
73 raise DeviceCodeTimeoutError("expired")
74 return self._creds
75
76
77class _HangingClient(_FakeClient):
78 """Passport client whose confirmation polls never finish on their own."""
79
80 async def poll_qr_until_confirmed(self, _qr: QrSession, **_kwargs: Any) -> Credentials:
81 await asyncio.Event().wait()
82 raise AssertionError("unreachable")
83
84 async def poll_device_until_confirmed(
85 self, _session: DeviceCodeSession, **_kwargs: Any
86 ) -> Credentials:
87 await asyncio.Event().wait()
88 raise AssertionError("unreachable")
89
90
91class _ControlledTimeout:
92 """Timeout context that expires when explicitly triggered by the test."""
93
94 def __init__(self) -> None:
95 self._task: asyncio.Task[Any] | None = None
96
97 async def __aenter__(self) -> None:
98 self._task = asyncio.current_task()
99 assert self._task is not None
100
101 async def __aexit__(
102 self,
103 exc_type: type[BaseException] | None,
104 _exc: BaseException | None,
105 _traceback: TracebackType | None,
106 ) -> bool:
107 if exc_type is not asyncio.CancelledError:
108 return False
109 assert self._task is not None
110 self._task.uncancel()
111 raise TimeoutError
112
113 def expire(self) -> None:
114 """Expire the context at the next suspension point."""
115 assert self._task is not None
116 self._task.cancel()
117
118
119def _async_cm(client: Any) -> mock.MagicMock:
120 """Wrap a fake client as the async context manager PassportClient.create returns."""
121 ctx = mock.MagicMock()
122 ctx.__aenter__ = mock.AsyncMock(return_value=client)
123 ctx.__aexit__ = mock.AsyncMock(return_value=False)
124 return ctx
125
126
127def _make_session(finish_handler: Any) -> tuple[SetupSession, mock.Mock]:
128 """Build a real SetupSession backed by a Mock mass for driving run_setup directly."""
129 mass = mock.Mock()
130 context = SetupFlowContext(kind="setup", reason="user", domain="yandex_music")
131 return SetupSession(mass, "flow-test", context, finish_handler), mass
132
133
134def _published_steps(mass: mock.Mock) -> list[Any]:
135 """Return the flow steps pushed through mass.signal_event, in order."""
136 return [call.kwargs["data"] for call in mass.signal_event.call_args_list]
137
138
139async def _wait_for(predicate: Any, timeout: float = 5.0) -> Any:
140 """Wait until the predicate returns truthy (or fail the test)."""
141 deadline = time.monotonic() + timeout
142 while time.monotonic() < deadline:
143 if result := predicate():
144 return result
145 await asyncio.sleep(0.01)
146 raise AssertionError("condition not met within timeout")
147
148
149async def _drive(session: SetupSession, submit: dict[str, Any]) -> None:
150 """Wait for the user form, submit the given values, then wait for finish."""
151 await _wait_for(lambda: session.current_step and session.current_step.type == FlowStepType.FORM)
152 session.handle_submit(submit)
153 await _wait_for(lambda: session.finished)
154
155
156async def _assert_login_has_hard_timeout(
157 login: Callable[[SetupSession], Awaitable[Credentials]],
158) -> None:
159 """Assert an abandoned login expires even while Yandex polling remains pending."""
160 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"))
161 client = _HangingClient(creds)
162 session = mock.Mock(spec=SetupSession)
163 loop = mock.Mock()
164 loop.time.side_effect = [100.0, 100.0]
165 timeout = _ControlledTimeout()
166 provider_asyncio = mock.Mock(wraps=asyncio)
167 provider_asyncio.get_running_loop.return_value = loop
168 provider_asyncio.timeout_at.return_value = timeout
169
170 async def progress_until(awaitable: Awaitable[Credentials], **_kwargs: Any) -> Credentials:
171 timeout.expire()
172 return await awaitable
173
174 session.progress_until = mock.AsyncMock(side_effect=progress_until)
175 with (
176 mock.patch.object(ym_flow, "_AUTH_FLOW_TIMEOUT_SECONDS", 10),
177 mock.patch.object(ym_flow, "PassportClient") as passport_client,
178 mock.patch.object(ym_flow, "asyncio", provider_asyncio),
179 ):
180 passport_client.create.return_value = _async_cm(client)
181 with pytest.raises(StepExpiredError):
182 await login(session)
183
184 provider_asyncio.timeout_at.assert_called_once_with(110)
185 assert client.qr_starts + client.device_starts == 1
186
187
188def test_qr_image_has_opaque_white_quiet_zone() -> None:
189 """The QR remains high-contrast against Music Assistant's dark theme."""
190 image = ym_flow._qr_image("https://passport.yandex.ru/qr/test")
191 svg = unquote(image.split(",", 1)[1])
192
193 assert "<path fill='#fff' d='M0 0h37v37h-37z'/>" in svg
194 assert "<path class='qrline' stroke='#000' d='M4 4.5" in svg
195
196
197async def test_qr_login_has_hard_timeout() -> None:
198 """An abandoned QR login cannot refresh codes forever."""
199 await _assert_login_has_hard_timeout(ym_flow._qr_login)
200
201
202async def test_device_login_has_hard_timeout() -> None:
203 """An abandoned Device Code login cannot refresh codes forever."""
204 await _assert_login_has_hard_timeout(ym_flow._device_login)
205
206
207async def test_device_countdown_respects_hard_timeout() -> None:
208 """The displayed Device Code lifetime cannot exceed the whole flow lifetime."""
209 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"))
210 client = _FakeClient(creds)
211 session = mock.Mock(spec=SetupSession)
212 shown_expiry: float | None = None
213 loop = mock.Mock()
214 loop.time.side_effect = [100.0, 100.0]
215 provider_asyncio = mock.Mock(wraps=asyncio)
216 provider_asyncio.get_running_loop.return_value = loop
217 provider_asyncio.timeout_at.return_value = _async_cm(None)
218
219 async def progress_until(
220 awaitable: Awaitable[Credentials], *, expires_in: float, **_kwargs: Any
221 ) -> Credentials:
222 nonlocal shown_expiry
223 shown_expiry = expires_in
224 return await awaitable
225
226 session.progress_until = mock.AsyncMock(side_effect=progress_until)
227 with (
228 mock.patch.object(ym_flow, "_AUTH_FLOW_TIMEOUT_SECONDS", 10),
229 mock.patch.object(ym_flow, "PassportClient") as passport_client,
230 mock.patch.object(ym_flow, "asyncio", provider_asyncio),
231 ):
232 passport_client.create.return_value = _async_cm(client)
233 await ym_flow._device_login(session)
234
235 assert shown_expiry == 10
236
237
238def test_device_image_makes_verification_address_prominent() -> None:
239 """The non-clickable fallback clearly tells users where to enter the code."""
240 image = ym_flow._device_image("ABCD-1234", "https://ya.ru/device")
241 svg = base64.b64decode(image.split(",", 1)[1]).decode("utf-8")
242
243 assert "Open this address in a browser" in svg
244 assert ">ya.ru/device</text>" in svg
245 assert "https://ya.ru/device" not in svg
246 address = svg.split(">ya.ru/device</text>", 1)[0].rsplit("<text", 1)[1]
247 assert 'font-size="24"' in address
248
249
250async def test_manual_token_is_only_shown_after_selecting_its_method() -> None:
251 """QR and Device Code never render the secure token input on their method form."""
252
253 async def finish(_s: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
254 raise AssertionError("form inspection must not finish the flow")
255
256 session, _mass = _make_session(finish)
257 task = asyncio.create_task(ym_flow.run_setup(session))
258 try:
259 form = await _wait_for(
260 lambda: (
261 session.current_step
262 if session.current_step and session.current_step.type == FlowStepType.FORM
263 else None
264 )
265 )
266 method_entry = next(entry for entry in form.entries if entry.key == ym_flow.CONF_METHOD)
267 assert {option.value for option in method_entry.options} >= {"qr", "device", "token"}
268 assert method_entry.default_value == ym_flow.METHOD_QR
269 assert CONF_TOKEN not in {entry.key for entry in form.entries}
270
271 session.handle_submit({ym_flow.CONF_METHOD: "token", CONF_REMEMBER_SESSION: True})
272 token_form = await _wait_for(
273 lambda: (
274 session.current_step
275 if session.current_step
276 and session.current_step.type == FlowStepType.FORM
277 and session.current_step.step_id == "token_login"
278 else None
279 ),
280 timeout=0.5,
281 )
282 token_entry = next(entry for entry in token_form.entries if entry.key == CONF_TOKEN)
283 assert token_entry.type == ConfigEntryType.SECURE_STRING
284 assert token_entry.required is True
285 assert {entry.key for entry in token_form.entries} == {CONF_TOKEN}
286 finally:
287 task.cancel()
288 with suppress(BaseException):
289 await task
290
291
292async def test_manual_token_login_persists_only_submitted_token() -> None:
293 """Manual login bypasses Passport and clears credentials from a previous session."""
294 creds = Credentials(x_token=SecretStr("unused-XT"), music_token=SecretStr("unused-MT"))
295 client = _FakeClient(creds)
296 collected: dict[str, Any] = {}
297
298 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
299 collected.update(values)
300 return {"instance_id": "yandex_music--1"}
301
302 session, _mass = _make_session(finish)
303 with mock.patch.object(ym_flow, "PassportClient") as passport_client:
304 passport_client.create.return_value = _async_cm(client)
305 task = asyncio.create_task(ym_flow.run_setup(session))
306 await _wait_for(lambda: session.current_step and session.current_step.step_id == "user")
307 session.handle_submit({ym_flow.CONF_METHOD: "token", CONF_REMEMBER_SESSION: True})
308 await _wait_for(
309 lambda: session.current_step and session.current_step.step_id == "token_login",
310 timeout=0.5,
311 )
312 session.handle_submit({CONF_TOKEN: "manual-token"})
313 await _wait_for(lambda: session.finished)
314 await task
315
316 assert collected == {
317 CONF_TOKEN: "manual-token",
318 CONF_X_TOKEN: None,
319 CONF_REFRESH_TOKEN: None,
320 }
321 passport_client.create.assert_not_called()
322
323
324async def test_manual_token_login_rejects_empty_token() -> None:
325 """Selecting manual login without a token re-renders the form with a field error."""
326
327 async def finish(_s: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
328 raise AssertionError("an empty manual token must not finish the flow")
329
330 session, _mass = _make_session(finish)
331 task = asyncio.create_task(ym_flow.run_setup(session))
332 try:
333 await _wait_for(lambda: session.current_step and session.current_step.step_id == "user")
334 session.handle_submit({ym_flow.CONF_METHOD: "token", CONF_REMEMBER_SESSION: True})
335 await _wait_for(
336 lambda: session.current_step and session.current_step.step_id == "token_login",
337 timeout=0.5,
338 )
339 session.handle_submit({CONF_TOKEN: ""})
340 await _wait_for(
341 lambda: (
342 session.current_step and session.current_step.errors.get(CONF_TOKEN) == "required"
343 ),
344 timeout=0.5,
345 )
346 assert session.current_step is not None
347 assert session.current_step.errors == {CONF_TOKEN: "required"}
348 assert not task.done()
349 finally:
350 task.cancel()
351 with suppress(BaseException):
352 await task
353
354
355async def test_device_login_remember_persists_full_triple() -> None:
356 """Device login with remember on persists music + x + refresh tokens."""
357 creds = Credentials(
358 x_token=SecretStr("XT"),
359 music_token=SecretStr("MT"),
360 refresh_token=SecretStr("RT"),
361 display_login="alice",
362 uid=1,
363 )
364 collected: dict[str, Any] = {}
365
366 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
367 collected.update(values)
368 return {"instance_id": "yandex_music--1"}
369
370 session, mass = _make_session(finish)
371 client = _FakeClient(creds)
372 with mock.patch.object(ym_flow, "PassportClient") as pc:
373 pc.create.return_value = _async_cm(client)
374 task = asyncio.create_task(ym_flow.run_setup(session))
375 form = await _wait_for(
376 lambda: (
377 session.current_step
378 if session.current_step and session.current_step.type == FlowStepType.FORM
379 else None
380 )
381 )
382 method_entry = next(entry for entry in form.entries if entry.key == ym_flow.CONF_METHOD)
383 assert method_entry.default_value == ym_flow.METHOD_QR
384 await _drive(
385 session, {ym_flow.CONF_METHOD: ym_flow.METHOD_DEVICE, CONF_REMEMBER_SESSION: True}
386 )
387 await task
388
389 assert collected == {CONF_TOKEN: "MT", CONF_X_TOKEN: "XT", CONF_REFRESH_TOKEN: "RT"}
390 progress = [s for s in _published_steps(mass) if s.type == FlowStepType.PROGRESS]
391 assert progress
392 assert progress[0].step_id == "device_login"
393 assert progress[0].image is not None
394 assert progress[0].image.startswith("data:image/svg+xml")
395
396
397async def test_qr_login_without_remember_stores_music_token_only() -> None:
398 """QR login with remember off stores only the music token (x/refresh cleared)."""
399 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"), display_login="bob")
400 collected: dict[str, Any] = {}
401
402 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
403 collected.update(values)
404 return {"instance_id": "yandex_music--1"}
405
406 session, mass = _make_session(finish)
407 client = _FakeClient(creds)
408 with mock.patch.object(ym_flow, "PassportClient") as pc:
409 pc.create.return_value = _async_cm(client)
410 task = asyncio.create_task(ym_flow.run_setup(session))
411 await _drive(
412 session, {ym_flow.CONF_METHOD: ym_flow.METHOD_QR, CONF_REMEMBER_SESSION: False}
413 )
414 await task
415
416 assert collected == {CONF_TOKEN: "MT", CONF_X_TOKEN: None, CONF_REFRESH_TOKEN: None}
417 scan_steps = [s for s in _published_steps(mass) if s.step_id == "scan_qr"]
418 assert scan_steps
419 assert all(s.image and s.image.startswith("data:image/svg+xml") for s in scan_steps)
420
421
422async def test_qr_login_refreshes_expired_code() -> None:
423 """An expired QR code is minted afresh and the login still completes."""
424 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"))
425 collected: dict[str, Any] = {}
426
427 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
428 collected.update(values)
429 return {"instance_id": "yandex_music--1"}
430
431 session, _mass = _make_session(finish)
432 client = _FakeClient(creds, qr_fail_first=True)
433 with mock.patch.object(ym_flow, "PassportClient") as pc:
434 pc.create.return_value = _async_cm(client)
435 task = asyncio.create_task(ym_flow.run_setup(session))
436 await _drive(session, {ym_flow.CONF_METHOD: ym_flow.METHOD_QR, CONF_REMEMBER_SESSION: True})
437 await task
438
439 assert collected[CONF_TOKEN] == "MT"
440 # the expired code triggered a second start_qr_login (refresh loop)
441 assert client.qr_starts == 2
442
443
444async def test_device_login_refreshes_expired_code() -> None:
445 """An expired Device Code is minted afresh and the login still completes."""
446 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"))
447 collected: dict[str, Any] = {}
448
449 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
450 collected.update(values)
451 return {"instance_id": "yandex_music--1"}
452
453 session, _mass = _make_session(finish)
454 client = _FakeClient(creds, device_fail_first=True)
455 with mock.patch.object(ym_flow, "PassportClient") as passport_client:
456 passport_client.create.return_value = _async_cm(client)
457 task = asyncio.create_task(ym_flow.run_setup(session))
458 await _drive(
459 session, {ym_flow.CONF_METHOD: ym_flow.METHOD_DEVICE, CONF_REMEMBER_SESSION: True}
460 )
461 await task
462
463 assert collected[CONF_TOKEN] == "MT"
464 assert client.device_starts == 2
465