/
/
/
1"""Tests for the Pandora setup flow and the options surface it took over from."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from typing import Any
8from unittest.mock import Mock
9
10from music_assistant_models.enums import ConfigEntryType, FlowStepType
11
12from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME
13from music_assistant.models.setup_flow import SetupFlowContext, SetupFlowError, SetupSession
14from music_assistant.providers.pandora.constants import (
15 CONF_QUALITY,
16 CONF_TAKEOVER_ACTION,
17 QUALITY_STANDARD,
18)
19from music_assistant.providers.pandora.provider import PandoraProvider
20from music_assistant.providers.pandora.setup_flow import run_setup
21
22
23def _make_session(
24 finish_handler: Any, setup_data: dict[str, Any] | None = None
25) -> tuple[SetupSession, Mock]:
26 """Build a SetupSession backed by a Mock mass for driving run_setup directly."""
27 mass = Mock()
28 context = SetupFlowContext(
29 kind="setup", reason="user", domain="pandora", setup_data=setup_data or {}
30 )
31 session = SetupSession(mass, "flow-test", context, finish_handler)
32 return session, mass
33
34
35async def _wait_for(predicate: Any, timeout: float = 5.0) -> Any:
36 """Wait until the predicate returns truthy (or fail the test)."""
37 deadline = time.monotonic() + timeout
38 while time.monotonic() < deadline:
39 if result := predicate():
40 return result
41 await asyncio.sleep(0.01)
42 raise AssertionError("condition not met within timeout")
43
44
45async def _wait_for_form(session: SetupSession, with_errors: bool = False) -> Any:
46 """Wait until the flow publishes a FORM step (optionally one carrying errors)."""
47 return await _wait_for(
48 lambda: (
49 session.current_step
50 if session.current_step
51 and session.current_step.type == FlowStepType.FORM
52 and (session.current_step.errors if with_errors else True)
53 else None
54 )
55 )
56
57
58async def test_get_config_entries_excludes_credentials() -> None:
59 """The credentials are flow-owned now and must not appear on the options surface."""
60 provider = PandoraProvider.__new__(PandoraProvider)
61 entries = await provider.get_config_entries()
62 keys = {entry.key for entry in entries}
63 assert CONF_USERNAME not in keys
64 assert CONF_PASSWORD not in keys
65 # the genuine options (and the takeover action) stay
66 assert {CONF_QUALITY, CONF_TAKEOVER_ACTION} <= keys
67 # every remaining options entry must resolve without user input, since the instance
68 # is created with empty values and its config is validated right after construction
69 assert all(
70 entry.default_value is not None or not entry.required
71 for entry in entries
72 if entry.type != ConfigEntryType.ACTION
73 )
74
75
76async def test_run_setup_collects_credentials() -> None:
77 """The flow shows a username/password form and persists both as setup data."""
78 collected: dict[str, Any] = {}
79
80 async def finish_handler(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
81 collected.update(values)
82 return {"instance_id": "pandora--test"}
83
84 session, _mass = _make_session(finish_handler)
85 task = asyncio.create_task(run_setup(session))
86 step = await _wait_for_form(session)
87
88 assert step.step_id == "user"
89 assert step.last_step is True
90 entries = {entry.key: entry for entry in step.entries}
91 assert set(entries) == {CONF_USERNAME, CONF_PASSWORD}
92 assert entries[CONF_USERNAME].type == ConfigEntryType.STRING
93 assert entries[CONF_PASSWORD].type == ConfigEntryType.SECURE_STRING
94 assert all(entry.required for entry in entries.values())
95
96 session.handle_submit({CONF_USERNAME: "[email protected]", CONF_PASSWORD: "hunter2"})
97 await _wait_for(lambda: session.finished)
98 await task
99
100 assert collected == {CONF_USERNAME: "[email protected]", CONF_PASSWORD: "hunter2"}
101
102
103async def test_run_setup_prefills_username_but_not_password() -> None:
104 """A reconfigure re-shows the stored username; the secret is never echoed back."""
105
106 async def finish_handler(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
107 return {"instance_id": "pandora--test"}
108
109 session, _mass = _make_session(
110 finish_handler,
111 setup_data={CONF_USERNAME: "[email protected]", CONF_PASSWORD: "old-secret"},
112 )
113 task = asyncio.create_task(run_setup(session))
114 step = await _wait_for_form(session)
115
116 entries = {entry.key: entry for entry in step.entries}
117 assert entries[CONF_USERNAME].value == "[email protected]"
118 assert entries[CONF_PASSWORD].value is None
119
120 session.handle_submit({CONF_USERNAME: "[email protected]", CONF_PASSWORD: "new-secret"})
121 await _wait_for(lambda: session.finished)
122 await task
123
124
125async def test_run_setup_retries_form_on_login_failure() -> None:
126 """A failed login re-renders the form with the error, then succeeds on retry."""
127 attempts: list[dict[str, Any]] = []
128
129 async def finish_handler(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
130 # snapshot: the flow carries (and mutates) one setup_data dict across retries
131 attempts.append(dict(values))
132 if len(attempts) == 1:
133 raise SetupFlowError("Authentication failed", translation_key="login_failed")
134 return {"instance_id": "pandora--test"}
135
136 session, _mass = _make_session(finish_handler)
137 task = asyncio.create_task(run_setup(session))
138 await _wait_for_form(session)
139 session.handle_submit({CONF_USERNAME: "listener", CONF_PASSWORD: "wrong"})
140
141 error_form = await _wait_for_form(session, with_errors=True)
142 assert error_form.errors == {"base": "login_failed"}
143 # the rejected username is prefilled again so only the password has to be retyped
144 entries = {entry.key: entry for entry in error_form.entries}
145 assert entries[CONF_USERNAME].value == "listener"
146 assert entries[CONF_PASSWORD].value is None
147
148 session.handle_submit({CONF_USERNAME: "listener", CONF_PASSWORD: "right"})
149 await _wait_for(lambda: session.finished)
150 await task
151
152 assert [attempt[CONF_PASSWORD] for attempt in attempts] == ["wrong", "right"]
153
154
155async def test_default_quality_is_standard() -> None:
156 """The quality option keeps a default so the instance loads without user input."""
157 provider = PandoraProvider.__new__(PandoraProvider)
158 entries = {entry.key: entry for entry in await provider.get_config_entries()}
159 assert entries[CONF_QUALITY].default_value == QUALITY_STANDARD
160