/
/
/
1"""Tests for the NetEase Cloud Music interactive setup flow."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from typing import Any
8from unittest.mock import Mock, patch
9
10from music_assistant_models.enums import FlowStepType
11
12from music_assistant.models.setup_flow import (
13 SetupFlowContext,
14 SetupFlowError,
15 SetupSession,
16)
17from music_assistant.providers.neteasecloudmusic import setup_flow as ncm_flow
18from music_assistant.providers.neteasecloudmusic.constants import (
19 CONF_API_BASE_URL,
20 CONF_COOKIE,
21 CONF_UID,
22)
23
24_QR_IMAGE = "data:image/png;base64,QR-IMAGE-DATA"
25
26
27class _FakeClient:
28 """Canned NeteaseCloudMusicApi client that walks a QR login to confirmation."""
29
30 def __init__(self, *_args: Any, check_codes: list[int] | None = None) -> None:
31 # default: one expiry (800) then confirmation (803), exercising the refresh loop
32 self._check_codes = check_codes if check_codes is not None else [800, 803]
33 self._check_idx = 0
34 self.calls: list[str] = []
35
36 async def get(self, path: str, **_kwargs: Any) -> dict[str, Any]:
37 """Return a canned payload keyed by request path."""
38 self.calls.append(path)
39 if path == "/login/qr/key":
40 return {"code": 200, "data": {"unikey": "unikey-123"}}
41 if path == "/login/qr/create":
42 return {"code": 200, "data": {"qrimg": _QR_IMAGE}}
43 if path == "/login/qr/check":
44 code = self._check_codes[min(self._check_idx, len(self._check_codes) - 1)]
45 self._check_idx += 1
46 if code == 803:
47 return {"code": 803, "cookie": "MUSIC_U=secret-cookie"}
48 return {"code": code}
49 if path == "/login/status":
50 return {"code": 200, "data": {"profile": {"userId": 42}}}
51 raise AssertionError(f"unexpected path {path}")
52
53
54def _make_session(finish_handler: Any) -> tuple[SetupSession, Mock]:
55 """Build a SetupSession backed by a Mock mass for driving run_setup directly."""
56 mass = Mock()
57 context = SetupFlowContext(kind="setup", reason="user", domain="neteasecloudmusic")
58 session = SetupSession(mass, "flow-test", context, finish_handler)
59 return session, mass
60
61
62def _published_steps(mass: Mock) -> list[Any]:
63 """Return the flow steps pushed through mass.signal_event, in order."""
64 return [call.kwargs["data"] for call in mass.signal_event.call_args_list]
65
66
67async def _wait_for(predicate: Any, timeout: float = 5.0) -> Any:
68 """Wait until the predicate returns truthy (or fail the test)."""
69 deadline = time.monotonic() + timeout
70 while time.monotonic() < deadline:
71 if result := predicate():
72 return result
73 await asyncio.sleep(0.01)
74 raise AssertionError("condition not met within timeout")
75
76
77async def test_run_setup_qr_refresh_and_finish() -> None:
78 """The QR flow collects the backend, refreshes an expired QR, then stores cookie/uid."""
79 collected: dict[str, Any] = {}
80
81 async def finish_handler(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
82 collected.update(values)
83 return {"instance_id": "neteasecloudmusic--test"}
84
85 session, mass = _make_session(finish_handler)
86 fake_client = _FakeClient()
87
88 with patch.object(ncm_flow, "NcmApiClient", return_value=fake_client):
89 task = asyncio.create_task(ncm_flow.run_setup(session))
90 # first step is the backend-url form
91 user_step = await _wait_for(
92 lambda: (
93 session.current_step
94 if session.current_step and session.current_step.type == FlowStepType.FORM
95 else None
96 )
97 )
98 assert user_step.step_id == "user"
99 session.handle_submit({CONF_API_BASE_URL: "http://127.0.0.1:3000"})
100 await _wait_for(lambda: session.finished)
101 await task
102
103 # cookie + uid resolved and the chosen backend url are persisted together
104 assert collected == {
105 CONF_API_BASE_URL: "http://127.0.0.1:3000",
106 CONF_COOKIE: "MUSIC_U=secret-cookie",
107 CONF_UID: "42",
108 }
109 # the expired code triggered a second key/create/check round (refresh loop)
110 assert fake_client.calls.count("/login/qr/key") == 2
111 assert fake_client.calls.count("/login/qr/check") == 2
112 # the QR image is emitted verbatim on the scan_qr progress step(s)
113 progress_steps = [step for step in _published_steps(mass) if step.type == FlowStepType.PROGRESS]
114 assert progress_steps
115 assert all(step.step_id == "scan_qr" for step in progress_steps)
116 assert all(step.image == _QR_IMAGE for step in progress_steps)
117
118
119async def test_run_setup_retries_form_on_finish_error() -> None:
120 """A finish failure re-renders the user form with the error, then succeeds on retry."""
121 attempts = {"count": 0}
122
123 async def finish_handler(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
124 attempts["count"] += 1
125 if attempts["count"] == 1:
126 raise SetupFlowError("bad backend", translation_key="login_failed")
127 return {"instance_id": "neteasecloudmusic--test"}
128
129 session, _mass = _make_session(finish_handler)
130
131 # a fresh client per instantiation so each attempt confirms immediately
132 with patch.object(
133 ncm_flow, "NcmApiClient", side_effect=lambda *_a, **_k: _FakeClient(check_codes=[803])
134 ):
135 task = asyncio.create_task(ncm_flow.run_setup(session))
136 await _wait_for(
137 lambda: session.current_step and session.current_step.type == FlowStepType.FORM
138 )
139 session.handle_submit({CONF_API_BASE_URL: "http://127.0.0.1:3000"})
140 # after the failed finish the flow loops back to the user form with the error
141 error_form = await _wait_for(
142 lambda: (
143 session.current_step
144 if session.current_step
145 and session.current_step.type == FlowStepType.FORM
146 and session.current_step.errors
147 else None
148 )
149 )
150 assert error_form.errors == {"base": "login_failed"}
151 session.handle_submit({CONF_API_BASE_URL: "http://127.0.0.1:3000"})
152 await _wait_for(lambda: session.finished)
153 await task
154
155 assert attempts["count"] == 2
156