/
/
1"""Tests for the Yandex Ynison interactive setup flow (run_setup)."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from typing import Any
8from unittest import mock
9
10from music_assistant_models.enums import FlowStepType
11from ya_passport_auth import Credentials, QrSession, SecretStr
12
13from music_assistant.models.setup_flow import SetupFlowContext, SetupSession
14from music_assistant.providers.yandex_ynison import setup_flow as yn_flow
15from music_assistant.providers.yandex_ynison.constants import (
16 CONF_ACCOUNT_LOGIN,
17 CONF_MASS_PLAYER_ID,
18 CONF_PUBLISH_NAME,
19 CONF_REMEMBER_SESSION,
20 CONF_TOKEN,
21 CONF_X_TOKEN,
22 CONF_YM_INSTANCE,
23 YM_INSTANCE_OWN,
24)
25
26
27class _FakeClient:
28 """Canned PassportClient that confirms a QR login."""
29
30 def __init__(self, creds: Credentials) -> None:
31 self._creds = creds
32 self.qr_starts = 0
33
34 async def start_qr_login(self) -> QrSession:
35 self.qr_starts += 1
36 return QrSession(track_id="t", csrf_token="c", qr_url="https://passport.yandex.ru/qr/abc")
37
38 async def poll_qr_until_confirmed(self, _qr: QrSession, **_kwargs: Any) -> Credentials:
39 return self._creds
40
41
42def _async_cm(client: _FakeClient) -> mock.MagicMock:
43 """Wrap a fake client as the async context manager PassportClient.create returns."""
44 ctx = mock.MagicMock()
45 ctx.__aenter__ = mock.AsyncMock(return_value=client)
46 ctx.__aexit__ = mock.AsyncMock(return_value=False)
47 return ctx
48
49
50def _make_session(
51 finish_handler: Any, providers: dict[str, Any] | None = None
52) -> tuple[SetupSession, mock.Mock]:
53 """Build a real SetupSession backed by a Mock mass listing the given YM providers."""
54 mass = mock.Mock()
55 mass.config.get = mock.Mock(return_value=providers or {})
56 player = mock.Mock()
57 player.player_id = "kitchen"
58 player.display_name = "Kitchen"
59 mass.players.all_players.return_value = [player]
60 context = SetupFlowContext(kind="setup", reason="user", domain="yandex_ynison")
61 return SetupSession(mass, "flow-test", context, finish_handler), mass
62
63
64def _published_steps(mass: mock.Mock) -> list[Any]:
65 """Return the flow steps pushed through mass.signal_event, in order."""
66 return [call.kwargs["data"] for call in mass.signal_event.call_args_list]
67
68
69async def _wait_for(predicate: Any, timeout: float = 5.0) -> Any:
70 """Wait until the predicate returns truthy (or fail the test)."""
71 deadline = time.monotonic() + timeout
72 while time.monotonic() < deadline:
73 if result := predicate():
74 return result
75 await asyncio.sleep(0.01)
76 raise AssertionError("condition not met within timeout")
77
78
79async def _await_user_form(session: SetupSession) -> None:
80 """Wait until the user form is presented."""
81 await _wait_for(lambda: session.current_step and session.current_step.type == FlowStepType.FORM)
82
83
84async def test_borrow_mode_finishes_with_instance_only() -> None:
85 """Selecting a linked Yandex Music instance persists only that instance id."""
86 collected: dict[str, Any] = {}
87
88 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
89 collected.update(values)
90 return {"instance_id": "yandex_ynison--1"}
91
92 session, _mass = _make_session(
93 finish, providers={"ym-a": {"domain": "yandex_music", "name": "Main"}}
94 )
95 task = asyncio.create_task(yn_flow.run_setup(session))
96 await _await_user_form(session)
97 session.handle_submit(
98 {
99 CONF_YM_INSTANCE: "ym-a",
100 CONF_REMEMBER_SESSION: True,
101 CONF_MASS_PLAYER_ID: "kitchen",
102 CONF_PUBLISH_NAME: "Kitchen Yandex",
103 }
104 )
105 await _wait_for(lambda: session.finished)
106 await task
107
108 assert collected == {
109 CONF_YM_INSTANCE: "ym-a",
110 CONF_MASS_PLAYER_ID: "kitchen",
111 CONF_PUBLISH_NAME: "Kitchen Yandex",
112 }
113
114
115async def test_own_mode_qr_persists_tokens_and_login() -> None:
116 """Own-mode QR login persists music token, x_token (remember on) and display login."""
117 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"), display_login="alice")
118 collected: dict[str, Any] = {}
119
120 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
121 collected.update(values)
122 return {"instance_id": "yandex_ynison--1"}
123
124 session, mass = _make_session(finish)
125 client = _FakeClient(creds)
126 with mock.patch.object(yn_flow, "PassportClient") as pc:
127 pc.create.return_value = _async_cm(client)
128 task = asyncio.create_task(yn_flow.run_setup(session))
129 await _await_user_form(session)
130 session.handle_submit(
131 {
132 CONF_YM_INSTANCE: YM_INSTANCE_OWN,
133 CONF_REMEMBER_SESSION: True,
134 CONF_MASS_PLAYER_ID: "kitchen",
135 CONF_PUBLISH_NAME: "Kitchen Yandex",
136 }
137 )
138 await _wait_for(lambda: session.finished)
139 await task
140
141 assert collected == {
142 CONF_YM_INSTANCE: YM_INSTANCE_OWN,
143 CONF_TOKEN: "MT",
144 CONF_X_TOKEN: "XT",
145 CONF_ACCOUNT_LOGIN: "alice",
146 CONF_MASS_PLAYER_ID: "kitchen",
147 CONF_PUBLISH_NAME: "Kitchen Yandex",
148 }
149 scan_steps = [s for s in _published_steps(mass) if s.step_id == "scan_qr"]
150 assert scan_steps
151 assert all(s.image and s.image.startswith("data:image/svg+xml") for s in scan_steps)
152
153
154async def test_own_mode_without_remember_clears_x_token() -> None:
155 """Own-mode QR login with remember off stores the music token but no x_token."""
156 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"), display_login="bob")
157 collected: dict[str, Any] = {}
158
159 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
160 collected.update(values)
161 return {"instance_id": "yandex_ynison--1"}
162
163 session, _mass = _make_session(finish)
164 client = _FakeClient(creds)
165 with mock.patch.object(yn_flow, "PassportClient") as pc:
166 pc.create.return_value = _async_cm(client)
167 task = asyncio.create_task(yn_flow.run_setup(session))
168 await _await_user_form(session)
169 session.handle_submit(
170 {
171 CONF_YM_INSTANCE: YM_INSTANCE_OWN,
172 CONF_REMEMBER_SESSION: False,
173 CONF_MASS_PLAYER_ID: "kitchen",
174 CONF_PUBLISH_NAME: "Kitchen Yandex",
175 }
176 )
177 await _wait_for(lambda: session.finished)
178 await task
179
180 assert collected[CONF_TOKEN] == "MT"
181 assert collected[CONF_X_TOKEN] is None
182