music-assistant-server

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