/
/
/
1"""Tests for the Audible setup flow."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from pathlib import Path
8from typing import Any
9from unittest.mock import AsyncMock, MagicMock, Mock, patch
10
11from music_assistant_models.enums import FlowStepType
12
13from music_assistant.models.setup_flow import (
14 SetupFlowContext,
15 SetupFlowError,
16 SetupSession,
17 StepExpiredError,
18)
19from music_assistant.providers.audible import CONF_AUTH_FILE, CONF_LOCALE
20from music_assistant.providers.audible.setup_flow import CONF_POST_LOGIN_URL, run_setup
21
22
23def _make_session(
24 finish_handler: Any, tmp_path: Path, setup_data: dict[str, Any] | None = None
25) -> SetupSession:
26 """Build a SetupSession backed by a Mock mass for driving run_setup directly."""
27 mass = Mock()
28 mass.storage_path = str(tmp_path)
29 context = SetupFlowContext(
30 kind="reconfigure" if setup_data else "setup",
31 reason="user",
32 domain="audible",
33 setup_data=setup_data or {},
34 )
35 session = SetupSession(mass, "flow-test", context, finish_handler)
36 # skip the 30s "open the login URL" wait; the flow suppresses the expiry
37 session.external_until = AsyncMock(side_effect=StepExpiredError) # type: ignore[method-assign]
38 return session
39
40
41def _mock_registered_auth() -> MagicMock:
42 """Return a mock Authenticator whose to_file writes a placeholder file."""
43 auth = MagicMock()
44 auth.adp_token = "adp_token"
45 auth.device_private_key = "private_key"
46 auth.to_file.side_effect = lambda path: Path(path).write_text("{}")
47 return auth
48
49
50async def _wait_for(predicate: Any, timeout: float = 5.0) -> Any:
51 """Wait until the predicate returns truthy (or fail the test)."""
52 deadline = time.monotonic() + timeout
53 while time.monotonic() < deadline:
54 if result := predicate():
55 return result
56 await asyncio.sleep(0.01)
57 raise AssertionError("condition not met within timeout")
58
59
60async def _wait_for_form(session: SetupSession, step_id: str, with_errors: bool = False) -> Any:
61 """Wait until the flow publishes the FORM step with the given step_id."""
62 return await _wait_for(
63 lambda: (
64 session.current_step
65 if session.current_step
66 and session.current_step.type == FlowStepType.FORM
67 and session.current_step.step_id == step_id
68 and (session.current_step.errors if with_errors else True)
69 else None
70 )
71 )
72
73
74async def _drive_to_finish(session: SetupSession) -> None:
75 """Submit the locale and post-login-url forms so the flow reaches finish()."""
76 await _wait_for_form(session, "user")
77 session.handle_submit({CONF_LOCALE: "de"})
78 await _wait_for_form(session, "authenticate")
79 session.handle_submit({CONF_POST_LOGIN_URL: "https://example.com/?code=dummy"})
80
81
82async def test_reauth_retires_previous_registration(tmp_path: Path) -> None:
83 """A successful re-auth deregisters the old device and removes its token file."""
84 old_file = tmp_path / "audible_auth_old.json"
85 old_file.write_text("{}")
86 collected: dict[str, Any] = {}
87
88 async def finish_handler(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
89 collected.update(values)
90 return {"instance_id": "audible--test"}
91
92 session = _make_session(
93 finish_handler, tmp_path, {CONF_LOCALE: "us", CONF_AUTH_FILE: str(old_file)}
94 )
95 prefix = "music_assistant.providers.audible.setup_flow"
96 with (
97 patch(f"{prefix}.audible_get_auth_info", return_value=("verifier", "http://l", "serial")),
98 patch(f"{prefix}.audible_custom_login", return_value=_mock_registered_auth()),
99 patch(f"{prefix}.deregister_auth_file") as deregister_mock,
100 ):
101 task = asyncio.create_task(run_setup(session))
102 await _drive_to_finish(session)
103 await _wait_for(lambda: session.finished)
104 await task
105
106 assert collected[CONF_LOCALE] == "de"
107 assert collected[CONF_AUTH_FILE] != str(old_file)
108 deregister_mock.assert_called_once_with(str(old_file))
109 assert not old_file.exists()
110 assert Path(collected[CONF_AUTH_FILE]).exists()
111
112
113async def test_failed_finish_keeps_previous_registration(tmp_path: Path) -> None:
114 """A failed finish drops the new token file and leaves the previous one alone."""
115 old_file = tmp_path / "audible_auth_old.json"
116 old_file.write_text("{}")
117 attempts: list[dict[str, Any]] = []
118
119 async def finish_handler(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
120 attempts.append(dict(values))
121 if len(attempts) == 1:
122 raise SetupFlowError("Provider did not load")
123 return {"instance_id": "audible--test"}
124
125 session = _make_session(
126 finish_handler, tmp_path, {CONF_LOCALE: "us", CONF_AUTH_FILE: str(old_file)}
127 )
128 prefix = "music_assistant.providers.audible.setup_flow"
129 with (
130 patch(f"{prefix}.audible_get_auth_info", return_value=("verifier", "http://l", "serial")),
131 patch(f"{prefix}.audible_custom_login", side_effect=lambda *_: _mock_registered_auth()),
132 patch(f"{prefix}.deregister_auth_file") as deregister_mock,
133 ):
134 task = asyncio.create_task(run_setup(session))
135 await _drive_to_finish(session)
136
137 # first finish fails: the rejected token file is gone, the old one is untouched
138 await _wait_for_form(session, "authenticate", with_errors=True)
139 rejected_file = attempts[0][CONF_AUTH_FILE]
140 assert not Path(rejected_file).exists()
141 assert old_file.exists()
142 deregister_mock.assert_not_called()
143
144 session.handle_submit({CONF_POST_LOGIN_URL: "https://example.com/?code=dummy"})
145 await _wait_for(lambda: session.finished)
146 await task
147
148 deregister_mock.assert_called_once_with(str(old_file))
149 assert not old_file.exists()
150