/
/
/
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
10import pytest
11from music_assistant_models.enums import FlowStepType
12from ya_passport_auth import Credentials, QrSession, SecretStr
13
14from music_assistant.models.setup_flow import AbortFlow, SetupFlowContext, SetupSession
15from music_assistant.providers.yandex_ynison import setup_flow as yn_flow
16from music_assistant.providers.yandex_ynison.constants import (
17 CONF_ACCOUNT_LOGIN,
18 CONF_MASS_PLAYER_ID,
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 }
103 )
104 await _wait_for(lambda: session.finished)
105 await task
106
107 assert collected == {
108 CONF_YM_INSTANCE: "ym-a",
109 CONF_MASS_PLAYER_ID: "kitchen",
110 }
111
112
113async def test_own_mode_qr_persists_tokens_and_login() -> None:
114 """Own-mode QR login persists music token, x_token (remember on) and display login."""
115 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"), display_login="alice")
116 collected: dict[str, Any] = {}
117
118 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
119 collected.update(values)
120 return {"instance_id": "yandex_ynison--1"}
121
122 session, mass = _make_session(finish)
123 client = _FakeClient(creds)
124 with mock.patch.object(yn_flow, "PassportClient") as pc:
125 pc.create.return_value = _async_cm(client)
126 task = asyncio.create_task(yn_flow.run_setup(session))
127 await _await_user_form(session)
128 session.handle_submit(
129 {
130 CONF_YM_INSTANCE: YM_INSTANCE_OWN,
131 CONF_REMEMBER_SESSION: True,
132 CONF_MASS_PLAYER_ID: "kitchen",
133 }
134 )
135 await _wait_for(lambda: session.finished)
136 await task
137
138 assert collected == {
139 CONF_YM_INSTANCE: YM_INSTANCE_OWN,
140 CONF_TOKEN: "MT",
141 CONF_X_TOKEN: "XT",
142 CONF_ACCOUNT_LOGIN: "alice",
143 CONF_MASS_PLAYER_ID: "kitchen",
144 }
145 scan_steps = [s for s in _published_steps(mass) if s.step_id == "scan_qr"]
146 assert scan_steps
147 assert all(s.image and s.image.startswith("data:image/svg+xml") for s in scan_steps)
148
149
150async def test_own_mode_without_remember_clears_x_token() -> None:
151 """Own-mode QR login with remember off stores the music token but no x_token."""
152 creds = Credentials(x_token=SecretStr("XT"), music_token=SecretStr("MT"), display_login="bob")
153 collected: dict[str, Any] = {}
154
155 async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
156 collected.update(values)
157 return {"instance_id": "yandex_ynison--1"}
158
159 session, _mass = _make_session(finish)
160 client = _FakeClient(creds)
161 with mock.patch.object(yn_flow, "PassportClient") as pc:
162 pc.create.return_value = _async_cm(client)
163 task = asyncio.create_task(yn_flow.run_setup(session))
164 await _await_user_form(session)
165 session.handle_submit(
166 {
167 CONF_YM_INSTANCE: YM_INSTANCE_OWN,
168 CONF_REMEMBER_SESSION: False,
169 CONF_MASS_PLAYER_ID: "kitchen",
170 }
171 )
172 await _wait_for(lambda: session.finished)
173 await task
174
175 assert collected[CONF_TOKEN] == "MT"
176 assert collected[CONF_X_TOKEN] is None
177
178
179async def test_aborts_without_players() -> None:
180 """With no players registered the flow aborts with the no_players reason."""
181
182 async def finish(_s: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
183 raise AssertionError("finish must not be reached")
184
185 session, mass = _make_session(finish)
186 mass.players.all_players.return_value = []
187
188 with pytest.raises(AbortFlow) as excinfo:
189 await yn_flow.run_setup(session)
190 assert excinfo.value.reason == "no_players"
191