music-assistant-server

1"""Tests for the setup flow engine (SetupFlowMixin + SetupSession)."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from collections.abc import AsyncGenerator, Awaitable, Callable
8from dataclasses import replace
9from types import SimpleNamespace
10from typing import TYPE_CHECKING, Any, cast
11from unittest.mock import AsyncMock, MagicMock, patch
12from urllib.parse import parse_qs, urlsplit
13
14import pytest
15from aiohttp.test_utils import make_mocked_request
16from music_assistant_models.auth import Scope
17from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, PlayerConfig
18from music_assistant_models.enums import (
19    ConfigEntryType,
20    EventType,
21    FlowStepType,
22    PlayerType,
23    ProviderType,
24)
25from music_assistant_models.errors import (
26    ActionUnavailable,
27    LoginFailed,
28    PlayerUnavailableError,
29    SetupFailedError,
30)
31from music_assistant_models.player import OutputProtocol
32from music_assistant_models.provider import ProviderManifest
33
34from music_assistant.constants import CONF_PLAYERS, CONF_PROVIDERS, ENCRYPT_SUFFIX
35from music_assistant.mass import MusicAssistant
36from music_assistant.models.music_provider import MusicProvider
37from music_assistant.models.player import Player, _state_fingerprint
38from music_assistant.models.setup_flow import AbortFlow, SetupSession, StepExpiredError
39from music_assistant.providers.filesystem_local.setup_flow import (
40    run_setup as filesystem_local_run_setup,
41)
42from music_assistant.providers.qobuz.setup_flow import run_setup as qobuz_run_setup
43from tests.common import MockPlayer, MockProvider
44
45if TYPE_CHECKING:
46    from music_assistant_models.config_entries import ProviderConfig
47    from music_assistant_models.event import MassEvent
48    from music_assistant_models.setup_flow import SetupFlowStep
49
50FAKE_DOMAIN = "_setup_flow_test"
51
52USERNAME_ENTRY = ConfigEntry(key="username", type=ConfigEntryType.STRING, required=True)
53PORT_ENTRY = ConfigEntry(key="port", type=ConfigEntryType.INTEGER, required=False, default_value=80)
54PASSWORD_ENTRY = ConfigEntry(key="password", type=ConfigEntryType.SECURE_STRING, required=True)
55REGION_ENTRY = ConfigEntry(
56    key="region", type=ConfigEntryType.STRING, required=True, default_value="eu"
57)
58USE_PROXY_ENTRY = ConfigEntry(key="use_proxy", type=ConfigEntryType.BOOLEAN, default_value=False)
59# required with no default to fall back on, so only the gate keeps it satisfiable
60PROXY_HOST_ENTRY = ConfigEntry(
61    key="proxy_host", type=ConfigEntryType.STRING, required=True, depends_on=USE_PROXY_ENTRY.key
62)
63
64
65@pytest.fixture
66async def flow_mass(mass_minimal: MusicAssistant) -> AsyncGenerator[MusicAssistant]:
67    """
68    Provide a minimal server with a fake (flow-capable) provider manifest injected.
69
70    Builds on mass_minimal (no webserver/ports bound) and stubs the narrow surface
71    the flow engine touches: the dynamic-route webserver API and the players/music
72    controllers.
73    """
74    manifest = ProviderManifest(
75        type=ProviderType.MUSIC,
76        domain=FAKE_DOMAIN,
77        name="Setup Flow Test Provider",
78        description="Fake provider for setup flow tests",
79        codeowners=[],
80    )
81    mass_minimal._provider_manifests[FAKE_DOMAIN] = manifest
82    # stub the dynamic-route surface so external-step tests need no bound port
83    routes: dict[str, Any] = {}
84
85    def register_dynamic_route(path: str, handler: Any, _method: str = "*") -> Any:
86        routes[path] = handler
87        return lambda: routes.pop(path, None)
88
89    mass_minimal.webserver = SimpleNamespace(  # type: ignore[assignment]
90        base_url="http://test.local:8095",
91        register_dynamic_route=register_dynamic_route,
92        unregister_dynamic_route=lambda path, _method="*": routes.pop(path, None),
93        routes=routes,
94    )
95    mass_minimal.music = MagicMock()
96    # awaited at the tail of the real provider load path
97    mass_minimal.music.on_provider_loaded = AsyncMock()
98    # awaited at the head of the real provider unload path
99    mass_minimal.music.unschedule_provider_sync = AsyncMock()
100    mass_minimal.players = MagicMock()
101    mass_minimal.players.on_player_config_change = AsyncMock()
102    try:
103        yield mass_minimal
104    finally:
105        # abort any flow a test left behind so no tasks outlive the loop
106        for flow in list(mass_minimal.config._setup_flows.values()):
107            await mass_minimal.config._abort_flow(flow, reason="aborted")
108        if (sweep_handle := mass_minimal.config._flow_sweep_handle) is not None:
109            sweep_handle.cancel()
110        mass_minimal._provider_manifests.pop(FAKE_DOMAIN, None)
111
112
113@pytest.fixture
114def flow_events(flow_mass: MusicAssistant) -> list[MassEvent]:
115    """Capture all SETUP_FLOW_UPDATED events emitted during the test."""
116    events: list[MassEvent] = []
117    flow_mass.subscribe(events.append, EventType.SETUP_FLOW_UPDATED)
118    return events
119
120
121def _use_flow(mass: MusicAssistant, flow: Callable[[SetupSession], Awaitable[Any]]) -> Any:
122    """Patch the engine's flow module loader to serve the given run_setup coroutine."""
123    return patch.object(
124        mass.config,
125        "_get_setup_flow_module",
126        AsyncMock(return_value=SimpleNamespace(run_setup=flow)),
127    )
128
129
130async def _wait_for(predicate: Callable[[], Any], timeout: float = 5.0) -> Any:
131    """Wait until the predicate returns a truthy value (or fail the test)."""
132    deadline = time.monotonic() + timeout
133    while time.monotonic() < deadline:
134        if result := predicate():
135            return result
136        await asyncio.sleep(0.01)
137    raise AssertionError("condition not met within timeout")
138
139
140def _abort_events(events: list[MassEvent]) -> list[SetupFlowStep]:
141    return [event.data for event in events if event.data.type == FlowStepType.ABORT]
142
143
144def _fake_json_session(payload: dict[str, Any]) -> Any:
145    """Return a stub http_session whose .post() yields a 200 JSON response (token exchange)."""
146    response = SimpleNamespace(
147        status=200,
148        json=AsyncMock(return_value=payload),
149        text=AsyncMock(return_value=""),
150    )
151    post_cm = MagicMock()
152    post_cm.__aenter__ = AsyncMock(return_value=response)
153    post_cm.__aexit__ = AsyncMock(return_value=False)
154    session = MagicMock()
155    session.post = MagicMock(return_value=post_cm)
156    return session
157
158
159async def _fire_callback(flow_mass: MusicAssistant, flow_id: str, query: str) -> None:
160    """Hit a running flow's external-step callback route with the given query string."""
161    callback_path = f"/setup_flow/callback/{flow_id}"
162    handler = cast("Any", flow_mass.webserver).routes[callback_path]
163    response = await handler(make_mocked_request("GET", f"{callback_path}?{query}"))
164    assert response.status == 200
165
166
167async def test_zero_input_provider_immediate_finish(flow_mass: MusicAssistant) -> None:
168    """A provider without a setup_flow module is created right away with a FINISH step."""
169    no_module = AsyncMock(return_value=None)
170    with (
171        patch.object(flow_mass.config, "_get_setup_flow_module", no_module),
172        patch.object(flow_mass, "load_provider_config", AsyncMock()) as mock_load,
173    ):
174        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
175    assert step.type == FlowStepType.FINISH
176    assert step.result == {"instance_id": FAKE_DOMAIN}
177    mock_load.assert_awaited_once()
178    # the config was created and persisted (with empty values/setup_data)
179    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
180    assert raw_conf["domain"] == FAKE_DOMAIN
181    assert raw_conf["setup_data"] == {}
182    # no flow session was registered for the synthesized step
183    assert not flow_mass.config._setup_flows
184
185
186class _FlowlessProvider(MusicProvider):
187    """
188    A genuine (minimal) provider for a domain that ships no setup_flow module.
189
190    Only get_config_entries is overridden so everything the load path does around it -
191    rehydrating the stored config against those entries, validating it and running async
192    init - behaves exactly as in production.
193    """
194
195    declared_entries: tuple[ConfigEntry, ...] = ()
196
197    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
198        """Return the (options) config entries this provider declares."""
199        return self.declared_entries
200
201
202def _use_provider_module(entries: tuple[ConfigEntry, ...]) -> Any:
203    """
204    Patch the module loader so the fake domain really loads a provider instance.
205
206    The fake domain has no importable package, so the module loader ``_load_provider`` uses
207    is served a stub module whose ``setup`` returns a real provider declaring the given entries.
208
209    :param entries: The (options) config entries the loaded provider declares.
210    """
211
212    async def setup(
213        mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
214    ) -> _FlowlessProvider:
215        provider = _FlowlessProvider(mass, manifest, config)
216        provider.declared_entries = entries
217        return provider
218
219    return patch(
220        "music_assistant.mass.load_provider_module",
221        AsyncMock(return_value=SimpleNamespace(setup=setup)),
222    )
223
224
225async def test_flowless_provider_loads_with_resolvable_entries(flow_mass: MusicAssistant) -> None:
226    """A flow-less provider whose options entries all resolve is added and actually loads."""
227    with (
228        patch.object(flow_mass.config, "_get_setup_flow_module", AsyncMock(return_value=None)),
229        _use_provider_module((PORT_ENTRY, REGION_ENTRY)),
230    ):
231        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
232    assert step.type == FlowStepType.FINISH
233    assert step.result == {"instance_id": FAKE_DOMAIN}
234    # the instance went through the real load path, not a stubbed one
235    provider = flow_mass.get_provider(FAKE_DOMAIN)
236    assert isinstance(provider, _FlowlessProvider)
237    assert provider.available
238    # it was created with values={}, yet the rehydrated config holds - and validates
239    # against - the provider's own entries
240    assert flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")["values"] == {}
241    assert {"port", "region"} <= set(provider.config.values)
242    provider.config.validate()
243    assert provider.config.get_value("port") == 80
244    assert provider.config.get_value("region") == "eu"
245
246
247async def test_flowless_provider_required_entry_without_default_rolls_back(
248    flow_mass: MusicAssistant,
249) -> None:
250    """A flow-less provider declaring a required entry with no default can not be added."""
251    with (
252        patch.object(flow_mass.config, "_get_setup_flow_module", AsyncMock(return_value=None)),
253        _use_provider_module((USERNAME_ENTRY,)),
254        # created with values={}, so validation of the rehydrated config has nothing to
255        # resolve the required entry from
256        pytest.raises(SetupFailedError, match="Configuration is invalid: username is required"),
257    ):
258        await flow_mass.config.setup_provider(FAKE_DOMAIN)
259    # the config created before the load attempt is rolled back, leaving nothing half-created
260    assert flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}") is None
261    assert flow_mass.get_provider(FAKE_DOMAIN, return_unavailable=True) is None
262
263
264async def test_form_flow_finish_success(
265    flow_mass: MusicAssistant, flow_events: list[MassEvent]
266) -> None:
267    """A form flow persists its (encrypted) values as setup_data and loads the provider."""
268
269    async def run_setup(session: SetupSession) -> None:
270        values = await session.form([USERNAME_ENTRY, PASSWORD_ENTRY], step_id="credentials")
271        await session.finish(values)
272
273    with (
274        _use_flow(flow_mass, run_setup),
275        patch.object(flow_mass, "load_provider_config", AsyncMock()) as mock_load,
276    ):
277        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
278        assert step.type == FlowStepType.FORM
279        assert step.step_id == "credentials"
280        # entries are stamped with the provider's translation owner
281        assert all(x.translation_owner == f"provider.{FAKE_DOMAIN}" for x in step.entries)
282        finish_step = await flow_mass.config.submit_setup_flow(
283            step.flow_id, {"username": "marcel", "password": "secret"}
284        )
285    assert finish_step.type == FlowStepType.FINISH
286    assert finish_step.result == {"instance_id": FAKE_DOMAIN}
287    mock_load.assert_awaited_once()
288    # values are stored in setup_data with strings encrypted at rest
289    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
290    assert raw_conf["setup_data"]["username"].startswith(ENCRYPT_SUFFIX)
291    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["username"]) == "marcel"
292    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["password"]) == "secret"
293    # the flow is cleaned up after finishing
294    assert step.flow_id not in flow_mass.config._setup_flows
295    # both steps were pushed as events
296    await _wait_for(lambda: len(flow_events) >= 2)
297    assert [event.data.type for event in flow_events[:2]] == [
298        FlowStepType.FORM,
299        FlowStepType.FINISH,
300    ]
301    assert all(event.object_id == step.flow_id for event in flow_events)
302
303
304async def test_form_flow_translation_params(flow_mass: MusicAssistant) -> None:
305    """A form step exposes translation parameters for its title and description."""
306
307    async def run_setup(session: SetupSession) -> None:
308        await session.form(
309            [USERNAME_ENTRY],
310            step_id="credentials",
311            translation_params=["https://example.com/callback"],
312        )
313
314    with _use_flow(flow_mass, run_setup):
315        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
316        assert step.translation_params == ["https://example.com/callback"]
317        await flow_mass.config.abort_setup_flow(step.flow_id)
318
319
320async def test_form_validation_errors(flow_mass: MusicAssistant) -> None:
321    """Invalid submitted values return the FORM step with errors, without advancing the flow."""
322    advanced = asyncio.Event()
323
324    async def run_setup(session: SetupSession) -> None:
325        values = await session.form([USERNAME_ENTRY, PORT_ENTRY])
326        advanced.set()
327        await session.finish(values)
328
329    with (
330        _use_flow(flow_mass, run_setup),
331        patch.object(flow_mass, "load_provider_config", AsyncMock()),
332    ):
333        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
334        # missing required value + unparsable integer
335        error_step = await flow_mass.config.submit_setup_flow(
336            step.flow_id, {"port": "not-a-number"}
337        )
338        assert error_step.type == FlowStepType.FORM
339        assert error_step.errors == {"username": "required", "port": "invalid_value"}
340        assert not advanced.is_set()
341        assert step.flow_id in flow_mass.config._setup_flows
342        # a valid re-submit picks up where the form left off
343        finish_step = await flow_mass.config.submit_setup_flow(
344            step.flow_id, {"username": "marcel", "port": 8095}
345        )
346    assert advanced.is_set()
347    assert finish_step.type == FlowStepType.FINISH
348    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
349    assert raw_conf["setup_data"]["port"] == 8095
350
351
352async def test_gated_required_entry_does_not_block_submit(flow_mass: MusicAssistant) -> None:
353    """A required entry behind an unmet dependency renders disabled, so it may stay empty."""
354    submitted: dict[str, ConfigValueType] = {}
355
356    async def run_setup(session: SetupSession) -> None:
357        submitted.update(await session.form([USE_PROXY_ENTRY, PROXY_HOST_ENTRY]))
358        await session.finish(submitted)
359
360    with (
361        _use_flow(flow_mass, run_setup),
362        patch.object(flow_mass, "load_provider_config", AsyncMock()),
363    ):
364        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
365        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"use_proxy": False})
366    assert finish_step.type == FlowStepType.FINISH
367    assert submitted == {"use_proxy": False, "proxy_host": None}
368
369
370async def test_gated_required_entry_is_demanded_once_its_gate_opens(
371    flow_mass: MusicAssistant,
372) -> None:
373    """Opening the gate in the same submit makes the entry required again."""
374
375    async def run_setup(session: SetupSession) -> None:
376        await session.finish(await session.form([USE_PROXY_ENTRY, PROXY_HOST_ENTRY]))
377
378    with (
379        _use_flow(flow_mass, run_setup),
380        patch.object(flow_mass, "load_provider_config", AsyncMock()),
381    ):
382        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
383        error_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"use_proxy": True})
384        assert error_step.type == FlowStepType.FORM
385        assert error_step.errors == {"proxy_host": "required"}
386        await flow_mass.config.abort_setup_flow(step.flow_id)
387
388
389async def test_gate_is_read_from_the_submitted_values(flow_mass: MusicAssistant) -> None:
390    """Closing a prefilled gate takes effect wherever the gate sits on the form."""
391    submitted: dict[str, ConfigValueType] = {}
392
393    async def run_setup(session: SetupSession) -> None:
394        # gate listed behind what it gates, and prefilled as a reconfigure run would
395        submitted.update(
396            await session.form([PROXY_HOST_ENTRY, replace(USE_PROXY_ENTRY, value=True)])
397        )
398        await session.finish(submitted)
399
400    with (
401        _use_flow(flow_mass, run_setup),
402        patch.object(flow_mass, "load_provider_config", AsyncMock()),
403    ):
404        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
405        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"use_proxy": False})
406    assert finish_step.type == FlowStepType.FINISH
407    assert submitted == {"use_proxy": False, "proxy_host": None}
408
409
410async def test_submit_returns_next_form_step(flow_mass: MusicAssistant) -> None:
411    """Submitting a multi-step flow returns the next FORM step."""
412
413    async def run_setup(session: SetupSession) -> None:
414        await session.form([USERNAME_ENTRY], step_id="first")
415        values = await session.form([PORT_ENTRY], step_id="second", last_step=True)
416        await session.finish(values)
417
418    with (
419        _use_flow(flow_mass, run_setup),
420        patch.object(flow_mass, "load_provider_config", AsyncMock()),
421    ):
422        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
423        assert step.step_id == "first"
424        second_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"username": "marcel"})
425        assert second_step.type == FlowStepType.FORM
426        assert second_step.step_id == "second"
427        assert second_step.last_step is True
428        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"port": 1234})
429        assert finish_step.type == FlowStepType.FINISH
430
431
432@pytest.mark.usefixtures("flow_events")
433async def test_finish_failure_rolls_back_provider_config(flow_mass: MusicAssistant) -> None:
434    """A failing provider load on finish removes the created config again."""
435
436    async def run_setup(session: SetupSession) -> None:
437        values = await session.form([USERNAME_ENTRY])
438        # SetupFlowError deliberately not caught: flow ends with the failure message
439        await session.finish(values)
440
441    with (
442        _use_flow(flow_mass, run_setup),
443        patch.object(
444            flow_mass, "load_provider_config", AsyncMock(side_effect=LoginFailed("bad creds"))
445        ),
446    ):
447        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
448        abort_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"username": "x"})
449    assert abort_step.type == FlowStepType.ABORT
450    assert abort_step.reason == "bad creds"
451    # the config created during finish was removed again
452    assert flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}") is None
453    assert step.flow_id not in flow_mass.config._setup_flows
454
455
456async def test_finish_failure_author_retry_loop(flow_mass: MusicAssistant) -> None:
457    """An author can catch SetupFlowError and re-render the form with the error."""
458
459    async def run_setup(session: SetupSession) -> None:
460        values = await session.form([USERNAME_ENTRY])
461        while True:
462            try:
463                await session.finish(values)
464                return
465            except Exception as err:
466                values = await session.form([USERNAME_ENTRY], errors={"base": str(err)})
467
468    load_mock = AsyncMock(side_effect=[LoginFailed("bad creds"), None])
469    with (
470        _use_flow(flow_mass, run_setup),
471        patch.object(flow_mass, "load_provider_config", load_mock),
472    ):
473        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
474        retry_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"username": "x"})
475        assert retry_step.type == FlowStepType.FORM
476        assert retry_step.errors == {"base": "bad creds"}
477        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"username": "y"})
478    assert finish_step.type == FlowStepType.FINISH
479    assert flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}") is not None
480
481
482async def test_external_step_callback_roundtrip(flow_mass: MusicAssistant) -> None:
483    """An external step resumes the flow with the merged callback (query) params."""
484    received: dict[str, str] = {}
485
486    async def run_setup(session: SetupSession) -> None:
487        assert session.callback_path == f"/setup_flow/callback/{session.flow_id}"
488        assert session.callback_url.endswith(f"/setup_flow/callback/{session.flow_id}")
489        received.update(await session.external("https://example.com/authorize"))
490        await session.finish({"token": received["code"]})
491
492    with (
493        _use_flow(flow_mass, run_setup),
494        patch.object(flow_mass, "load_provider_config", AsyncMock()),
495    ):
496        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
497        assert step.type == FlowStepType.EXTERNAL
498        assert step.url == "https://example.com/authorize"
499        session = flow_mass.config._setup_flows[step.flow_id].session
500        # the callback route was registered on the webserver (stub) for this flow
501        registered_routes = cast("Any", flow_mass.webserver).routes
502        callback_path = f"/setup_flow/callback/{step.flow_id}"
503        handler = registered_routes[callback_path]
504        request = make_mocked_request("GET", f"{callback_path}?code=abc&state=xyz")
505        response = await handler(request)
506        assert response.status == 200
507        await _wait_for(lambda: session.finished)
508    assert received == {"code": "abc", "state": "xyz"}
509    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
510    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["token"]) == "abc"
511    # the route is released again when the flow ends
512    await _wait_for(lambda: callback_path not in registered_routes)
513
514
515async def test_external_step_replaced_by_progress_on_callback(flow_mass: MusicAssistant) -> None:
516    """The external step gives way to a progress step while the callback is processed."""
517    release = asyncio.Event()
518
519    async def run_setup(session: SetupSession) -> None:
520        await session.external("https://example.com/authorize")
521        await release.wait()
522        await session.finish({})
523
524    with (
525        _use_flow(flow_mass, run_setup),
526        patch.object(flow_mass, "load_provider_config", AsyncMock()),
527    ):
528        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
529        session = flow_mass.config._setup_flows[step.flow_id].session
530        callback_path = f"/setup_flow/callback/{step.flow_id}"
531        handler = cast("Any", flow_mass.webserver).routes[callback_path]
532        await handler(make_mocked_request("GET", f"{callback_path}?code=abc"))
533        progress_step = await _wait_for(
534            lambda: (
535                session.current_step
536                if session.current_step is not None
537                and session.current_step.type == FlowStepType.PROGRESS
538                else None
539            )
540        )
541        assert progress_step.step_id == "working"
542        release.set()
543        await _wait_for(lambda: session.finished)
544
545
546async def test_external_expiry_aborts_flow(
547    flow_mass: MusicAssistant, flow_events: list[MassEvent]
548) -> None:
549    """An external step whose callback never arrives times out and releases its route."""
550
551    async def run_setup(session: SetupSession) -> None:
552        await session.external("https://example.com/authorize", expires_in=0.05)
553        await session.finish({})
554
555    with _use_flow(flow_mass, run_setup):
556        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
557        assert step.expires_at is not None
558        abort_step = await _wait_for(
559            lambda: next(iter(_abort_events(flow_events)), None), timeout=2
560        )
561    assert abort_step.reason == "timed_out"
562    assert step.flow_id not in flow_mass.config._setup_flows
563    # the expired step is not followed by a progress step: nothing was completed
564    assert not [event.data for event in flow_events if event.data.type == FlowStepType.PROGRESS]
565    registered_routes = cast("Any", flow_mass.webserver).routes
566    assert f"/setup_flow/callback/{step.flow_id}" not in registered_routes
567
568
569async def test_external_callback_json_body_coerced(flow_mass: MusicAssistant) -> None:
570    """A JSON callback body with non-string values is coerced to the str params contract."""
571    received: dict[str, str] = {}
572
573    async def run_setup(session: SetupSession) -> None:
574        received.update(await session.external("https://example.com/authorize"))
575        await session.finish({})
576
577    with (
578        _use_flow(flow_mass, run_setup),
579        patch.object(flow_mass, "load_provider_config", AsyncMock()),
580    ):
581        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
582        session = flow_mass.config._setup_flows[step.flow_id].session
583        request = SimpleNamespace(
584            query={"state": "xyz"},
585            method="POST",
586            can_read_body=True,
587            content_type="application/json",
588            read=AsyncMock(return_value=b'{"code": 123, "granted": true}'),
589        )
590        response = await session.handle_callback(cast("Any", request))
591        assert response.status == 200
592        await _wait_for(lambda: session.finished)
593    assert received == {"state": "xyz", "code": "123", "granted": "True"}
594
595
596async def test_form_expiry_aborts_flow(
597    flow_mass: MusicAssistant, flow_events: list[MassEvent]
598) -> None:
599    """An uncaught form deadline converts into a timed_out ABORT."""
600
601    async def run_setup(session: SetupSession) -> None:
602        await session.form([USERNAME_ENTRY], expires_in=0.05)
603        await session.finish({})
604
605    with _use_flow(flow_mass, run_setup):
606        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
607        assert step.expires_at is not None
608        abort_step = await _wait_for(
609            lambda: next(iter(_abort_events(flow_events)), None), timeout=2
610        )
611    assert abort_step.reason == "timed_out"
612    assert step.flow_id not in flow_mass.config._setup_flows
613
614
615async def test_form_expiry_author_refresh(flow_mass: MusicAssistant) -> None:
616    """An author can catch StepExpiredError and refresh the step in place."""
617
618    async def run_setup(session: SetupSession) -> None:
619        try:
620            await session.form([USERNAME_ENTRY], step_id="short_lived", expires_in=0.05)
621        except StepExpiredError:
622            await session.form([USERNAME_ENTRY], step_id="refreshed")
623        await session.finish({})
624
625    with (
626        _use_flow(flow_mass, run_setup),
627        patch.object(flow_mass, "load_provider_config", AsyncMock()),
628    ):
629        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
630        assert step.step_id == "short_lived"
631        flow = flow_mass.config._setup_flows[step.flow_id]
632        await _wait_for(
633            lambda: flow.session.current_step and flow.session.current_step.step_id == "refreshed",
634            timeout=2,
635        )
636        # the refreshed step accepts input as usual
637        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"username": "x"})
638        assert finish_step.type == FlowStepType.FINISH
639
640
641async def test_progress_until_expiry(
642    flow_mass: MusicAssistant, flow_events: list[MassEvent]
643) -> None:
644    """progress_until enforces its deadline by raising StepExpiredError into the flow."""
645
646    async def run_setup(session: SetupSession) -> None:
647        try:
648            await session.progress_until(
649                asyncio.sleep(30), "waiting_for_device", text="press_button", expires_in=0.05
650            )
651        except StepExpiredError:
652            raise AbortFlow("pairing_window_closed") from None
653
654    with _use_flow(flow_mass, run_setup):
655        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
656        assert step.type == FlowStepType.PROGRESS
657        assert step.progress_text == "press_button"
658        assert step.expires_at is not None
659        abort_step = await _wait_for(
660            lambda: next(iter(_abort_events(flow_events)), None), timeout=2
661        )
662    assert abort_step.reason == "pairing_window_closed"
663
664
665async def test_abort_runs_author_cleanup(
666    flow_mass: MusicAssistant, flow_events: list[MassEvent]
667) -> None:
668    """Aborting a flow cancels the coroutine so the author's finally cleanup runs."""
669    cleanup_ran = asyncio.Event()
670
671    async def run_setup(session: SetupSession) -> None:
672        try:
673            await session.form([USERNAME_ENTRY])
674        finally:
675            cleanup_ran.set()
676
677    with _use_flow(flow_mass, run_setup):
678        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
679        await flow_mass.config.abort_setup_flow(step.flow_id)
680    assert cleanup_ran.is_set()
681    assert step.flow_id not in flow_mass.config._setup_flows
682    abort_step = await _wait_for(lambda: next(iter(_abort_events(flow_events)), None))
683    assert abort_step.reason == "aborted"
684    # continuing an aborted flow is rejected
685    with pytest.raises(KeyError):
686        await flow_mass.config.get_setup_flow(step.flow_id)
687
688
689async def test_idle_flow_ttl_sweep(flow_mass: MusicAssistant, flow_events: list[MassEvent]) -> None:
690    """The periodic sweeper aborts flows that have been idle for longer than the TTL."""
691
692    async def run_setup(session: SetupSession) -> None:
693        await session.form([USERNAME_ENTRY])
694        await session.finish({})
695
696    with _use_flow(flow_mass, run_setup):
697        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
698        flow = flow_mass.config._setup_flows[step.flow_id]
699        flow.session.last_activity = time.monotonic() - 16 * 60
700        flow_mass.config._sweep_idle_flows()
701        await _wait_for(lambda: step.flow_id not in flow_mass.config._setup_flows)
702    abort_step = await _wait_for(lambda: next(iter(_abort_events(flow_events)), None))
703    assert abort_step.reason == "timed_out"
704
705
706async def test_idle_sweep_respects_live_step_deadline(
707    flow_mass: MusicAssistant, flow_events: list[MassEvent]
708) -> None:
709    """The sweeper must not abort a flow whose current step advertises a longer deadline."""
710
711    async def run_setup(session: SetupSession) -> None:
712        await session.form([USERNAME_ENTRY], expires_in=3600)
713        await session.finish({})
714
715    with _use_flow(flow_mass, run_setup):
716        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
717        flow = flow_mass.config._setup_flows[step.flow_id]
718        flow.session.last_activity = time.monotonic() - 16 * 60
719        flow_mass.config._sweep_idle_flows()
720        await asyncio.sleep(0.05)
721        # still alive: the step's own countdown is the authoritative deadline
722        assert step.flow_id in flow_mass.config._setup_flows
723        assert not _abort_events(flow_events)
724
725
726async def test_concurrent_starts_keep_single_flow(flow_mass: MusicAssistant) -> None:
727    """Two concurrent starts for the same target never leave two live flows behind."""
728
729    async def run_setup(session: SetupSession) -> None:
730        try:
731            await session.form([USERNAME_ENTRY])
732        finally:
733            # slow author cleanup: widens the abort window the re-scan protects against
734            await asyncio.sleep(0.05)
735        await session.finish({})
736
737    with _use_flow(flow_mass, run_setup):
738        await flow_mass.config.setup_provider(FAKE_DOMAIN)
739        await asyncio.gather(
740            flow_mass.config.setup_provider(FAKE_DOMAIN),
741            flow_mass.config.setup_provider(FAKE_DOMAIN),
742        )
743        target_flows = [
744            flow
745            for flow in flow_mass.config._setup_flows.values()
746            if flow.target_key == f"provider_setup:{FAKE_DOMAIN}"
747        ]
748        assert len(target_flows) <= 1
749
750
751async def test_finish_twice_raises(flow_mass: MusicAssistant, flow_events: list[MassEvent]) -> None:
752    """A second finish() call is a flow-author bug and must not create a second target."""
753
754    async def run_setup(session: SetupSession) -> None:
755        await session.finish({})
756        await session.finish({})
757
758    with (
759        _use_flow(flow_mass, run_setup),
760        patch.object(flow_mass, "load_provider_config", AsyncMock()),
761    ):
762        await flow_mass.config.setup_provider(FAKE_DOMAIN)
763        abort_step = await _wait_for(lambda: next(iter(_abort_events(flow_events)), None))
764    assert abort_step.reason == "internal_error"
765    # the first finish still created (exactly one) provider config
766    assert flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")["domain"] == FAKE_DOMAIN
767
768
769async def test_bare_callback_does_not_extend_ttl(flow_mass: MusicAssistant) -> None:
770    """The unauthenticated callback route must not keep an idle flow alive (no pending external)."""
771
772    async def run_setup(session: SetupSession) -> None:
773        await session.form([USERNAME_ENTRY])
774        await session.finish({})
775
776    with _use_flow(flow_mass, run_setup):
777        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
778        flow = flow_mass.config._setup_flows[step.flow_id]
779        activity_before = flow.session.last_activity
780        # the callback route is only registered while an external step is pending,
781        # but the session handler itself must also not count a bare hit as activity
782        request = make_mocked_request("GET", f"/setup_flow/callback/{step.flow_id}?foo=bar")
783        await flow.session.handle_callback(request)
784        assert flow.session.last_activity == activity_before
785
786
787async def test_submit_timeout_returns_transient_progress(flow_mass: MusicAssistant) -> None:
788    """A submit whose next step is slow returns a progress step, not the consumed form."""
789
790    async def run_setup(session: SetupSession) -> None:
791        await session.form([USERNAME_ENTRY])
792        await asyncio.sleep(0.3)
793        await session.form([PORT_ENTRY], step_id="second")
794        await session.finish({})
795
796    with (
797        _use_flow(flow_mass, run_setup),
798        patch("music_assistant.controllers.config.flows.NEXT_STEP_TIMEOUT", 0.05),
799    ):
800        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
801        result = await flow_mass.config.submit_setup_flow(step.flow_id, {"username": "x"})
802        assert result.type == FlowStepType.PROGRESS
803        assert result.flow_id == step.flow_id
804
805        # the real next step still arrives (idempotent get picks it up)
806        def _second_form() -> SetupFlowStep | None:
807            current = flow_mass.config._setup_flows[step.flow_id].session.current_step
808            if current is not None and current.step_id == "second":
809                return current
810            return None
811
812        next_step = await _wait_for(_second_form)
813        assert next_step.type == FlowStepType.FORM
814
815
816async def test_aborted_flow_scope_still_resolvable(flow_mass: MusicAssistant) -> None:
817    """A cancel-driven ABORT publishes after the pop; its scope must still resolve."""
818
819    async def run_setup(session: SetupSession) -> None:
820        await session.form([USERNAME_ENTRY])
821        await session.finish({})
822
823    resolvable_at_publish: list[bool] = []
824
825    def on_event(event: MassEvent) -> None:
826        if event.data.type == FlowStepType.ABORT and event.object_id:
827            resolvable_at_publish.append(
828                flow_mass.config.get_setup_flow_required_scope(event.object_id) is not None
829            )
830
831    flow_mass.subscribe(on_event, EventType.SETUP_FLOW_UPDATED)
832    with _use_flow(flow_mass, run_setup):
833        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
834        await flow_mass.config.abort_setup_flow(step.flow_id)
835    # event delivery is async (call_soon); wait for the callback to run
836    await _wait_for(lambda: resolvable_at_publish)
837    assert resolvable_at_publish == [True]
838    assert (
839        flow_mass.config.get_setup_flow_required_scope(step.flow_id) == Scope.CONFIG_PROVIDERS_WRITE
840    )
841
842
843async def test_setup_flow_required_scope_accessor(flow_mass: MusicAssistant) -> None:
844    """The event filter's scope accessor reports the flow's scope while it runs."""
845
846    async def run_setup(session: SetupSession) -> None:
847        await session.form([USERNAME_ENTRY])
848        await session.finish({})
849
850    with _use_flow(flow_mass, run_setup):
851        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
852        assert (
853            flow_mass.config.get_setup_flow_required_scope(step.flow_id)
854            == Scope.CONFIG_PROVIDERS_WRITE
855        )
856    assert flow_mass.config.get_setup_flow_required_scope("nonexistent") is None
857
858
859async def test_one_flow_per_target_replaces(
860    flow_mass: MusicAssistant, flow_events: list[MassEvent]
861) -> None:
862    """Starting a flow for a target that already has one aborts the old flow."""
863
864    async def run_setup(session: SetupSession) -> None:
865        await session.form([USERNAME_ENTRY])
866        await session.finish({})
867
868    with _use_flow(flow_mass, run_setup):
869        first_step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
870        second_step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
871    assert first_step.flow_id != second_step.flow_id
872    assert first_step.flow_id not in flow_mass.config._setup_flows
873    assert second_step.flow_id in flow_mass.config._setup_flows
874    replaced = await _wait_for(lambda: next(iter(_abort_events(flow_events)), None))
875    assert replaced.flow_id == first_step.flow_id
876    assert replaced.reason == "replaced"
877
878
879async def test_get_flow_is_idempotent(flow_mass: MusicAssistant) -> None:
880    """config/flows/get re-renders the current step without ever advancing the flow."""
881
882    async def run_setup(session: SetupSession) -> None:
883        values = await session.form([USERNAME_ENTRY])
884        await session.finish(values)
885
886    with (
887        _use_flow(flow_mass, run_setup),
888        patch.object(flow_mass, "load_provider_config", AsyncMock()),
889    ):
890        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
891        for _ in range(3):
892            fetched = await flow_mass.config.get_setup_flow(step.flow_id)
893            assert fetched.type == FlowStepType.FORM
894            assert fetched.step_id == step.step_id
895            assert fetched.flow_id == step.flow_id
896        # the flow still accepts input afterwards
897        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"username": "x"})
898        assert finish_step.type == FlowStepType.FINISH
899
900
901async def test_submit_without_pending_form(flow_mass: MusicAssistant) -> None:
902    """Submitting while no FORM step is pending is rejected."""
903
904    async def run_setup(session: SetupSession) -> None:
905        await session.external("https://example.com/authorize")
906        await session.finish({})
907
908    with _use_flow(flow_mass, run_setup):
909        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
910        assert step.type == FlowStepType.EXTERNAL
911        with pytest.raises(ActionUnavailable):
912            await flow_mass.config.submit_setup_flow(step.flow_id, {"username": "x"})
913        await flow_mass.config.abort_setup_flow(step.flow_id)
914
915
916@pytest.mark.usefixtures("flow_events")
917async def test_form_entries_with_action_rejected(flow_mass: MusicAssistant) -> None:
918    """The engine rejects FORM entries that carry an action (banned inside flows)."""
919    action_entry = ConfigEntry(
920        key="authenticate", type=ConfigEntryType.STRING, action="auth", required=False
921    )
922
923    async def run_setup(session: SetupSession) -> None:
924        await session.form([action_entry])
925        await session.finish({})
926
927    with _use_flow(flow_mass, run_setup):
928        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
929    # the ValueError ends the flow as an internal error
930    assert step.type == FlowStepType.ABORT
931    assert step.reason == "internal_error"
932
933
934@pytest.mark.usefixtures("flow_events")
935async def test_form_entries_with_action_type_rejected(flow_mass: MusicAssistant) -> None:
936    """The engine rejects FORM entries of type ACTION (they render dead in the dialog)."""
937    action_entry = ConfigEntry(key="authenticate", type=ConfigEntryType.ACTION, required=False)
938
939    async def run_setup(session: SetupSession) -> None:
940        await session.form([action_entry])
941        await session.finish({})
942
943    with _use_flow(flow_mass, run_setup):
944        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
945    # the ValueError ends the flow as an internal error
946    assert step.type == FlowStepType.ABORT
947    assert step.reason == "internal_error"
948
949
950async def test_reconfigure_prefill_and_success(flow_mass: MusicAssistant) -> None:
951    """Reconfigure decrypts setup_data for prefill and merges the new values on finish."""
952    instance_id = FAKE_DOMAIN
953    flow_mass.config.set(
954        f"{CONF_PROVIDERS}/{instance_id}",
955        {
956            "type": "music",
957            "domain": FAKE_DOMAIN,
958            "instance_id": instance_id,
959            "enabled": True,
960            "values": {"region": "eu"},
961            "setup_data": {
962                "token": flow_mass.config.encrypt_string("old-secret"),
963                "device_id": flow_mass.config.encrypt_string("device-1"),
964            },
965            "last_error": {"error_code": LoginFailed.error_code, "message": "expired"},
966        },
967    )
968    contexts: list[Any] = []
969
970    async def run_setup(session: SetupSession) -> None:
971        contexts.append(session.context)
972        values = await session.form([ConfigEntry(key="token", type=ConfigEntryType.STRING)])
973        await session.finish(values)
974
975    with (
976        _use_flow(flow_mass, run_setup),
977        patch.object(flow_mass, "load_provider_config", AsyncMock()) as mock_load,
978    ):
979        step = await flow_mass.config.reconfigure_provider(instance_id)
980        assert step.type == FlowStepType.FORM
981        context = contexts[0]
982        assert context.kind == "reconfigure"
983        assert context.reason == "auth"
984        assert context.instance_id == instance_id
985        assert context.setup_data == {"token": "old-secret", "device_id": "device-1"}
986        assert context.values == {"region": "eu"}
987        finish_step = await flow_mass.config.submit_setup_flow(
988            step.flow_id, {"token": "new-secret"}
989        )
990    assert finish_step.type == FlowStepType.FINISH
991    assert finish_step.result == {"instance_id": instance_id}
992    mock_load.assert_awaited_once()
993    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{instance_id}")
994    # new value merged in (encrypted), untouched keys preserved, last_error cleared
995    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["token"]) == "new-secret"
996    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["device_id"]) == "device-1"
997    assert raw_conf["last_error"] is None
998
999
1000async def test_reconfigure_failure_restores_setup_data(flow_mass: MusicAssistant) -> None:
1001    """A failing reload on reconfigure finish restores the previous setup_data."""
1002    instance_id = FAKE_DOMAIN
1003    original_setup_data = {"token": flow_mass.config.encrypt_string("old-secret")}
1004    flow_mass.config.set(
1005        f"{CONF_PROVIDERS}/{instance_id}",
1006        {
1007            "type": "music",
1008            "domain": FAKE_DOMAIN,
1009            "instance_id": instance_id,
1010            "enabled": True,
1011            "values": {},
1012            "setup_data": dict(original_setup_data),
1013        },
1014    )
1015
1016    async def run_setup(session: SetupSession) -> None:
1017        values = await session.form([ConfigEntry(key="token", type=ConfigEntryType.STRING)])
1018        await session.finish(values)
1019
1020    with (
1021        _use_flow(flow_mass, run_setup),
1022        patch.object(
1023            flow_mass, "load_provider_config", AsyncMock(side_effect=LoginFailed("still bad"))
1024        ),
1025    ):
1026        step = await flow_mass.config.reconfigure_provider(instance_id)
1027        abort_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"token": "bogus"})
1028    assert abort_step.type == FlowStepType.ABORT
1029    assert abort_step.reason == "still bad"
1030    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{instance_id}")
1031    assert raw_conf["setup_data"] == original_setup_data
1032
1033
1034async def test_reconfigure_without_flow_module_aborts(flow_mass: MusicAssistant) -> None:
1035    """Reconfigure on a flow-less provider returns a nothing_to_configure ABORT."""
1036    instance_id = FAKE_DOMAIN
1037    flow_mass.config.set(
1038        f"{CONF_PROVIDERS}/{instance_id}",
1039        {"type": "music", "domain": FAKE_DOMAIN, "instance_id": instance_id, "enabled": True},
1040    )
1041    with patch.object(flow_mass.config, "_get_setup_flow_module", AsyncMock(return_value=None)):
1042        step = await flow_mass.config.reconfigure_provider(instance_id)
1043    assert step.type == FlowStepType.ABORT
1044    assert step.reason == "nothing_to_configure"
1045    assert not flow_mass.config._setup_flows
1046
1047
1048async def test_setup_provider_single_instance_guard(flow_mass: MusicAssistant) -> None:
1049    """Setting up a non-multi-instance provider that already exists aborts fast."""
1050    flow_mass.config.set(
1051        f"{CONF_PROVIDERS}/{FAKE_DOMAIN}",
1052        {"type": "music", "domain": FAKE_DOMAIN, "instance_id": FAKE_DOMAIN, "enabled": True},
1053    )
1054    step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1055    assert step.type == FlowStepType.ABORT
1056    assert step.reason == "already_configured"
1057
1058
1059async def test_setup_unknown_provider_domain(flow_mass: MusicAssistant) -> None:
1060    """Setting up an unknown provider domain raises."""
1061    with pytest.raises(KeyError):
1062        await flow_mass.config.setup_provider("_no_such_provider")
1063
1064
1065class _FlowPlayer(MockPlayer):
1066    """Mock player that implements a setup (pairing) flow."""
1067
1068    async def run_setup_flow(self, session: SetupSession) -> None:
1069        """Run a minimal pairing flow."""
1070        values = await session.form([ConfigEntry(key="pin", type=ConfigEntryType.STRING)])
1071        await session.finish(values)
1072
1073
1074async def test_player_setup_without_flow_aborts(flow_mass: MusicAssistant) -> None:
1075    """Player setup for a player without a run_setup_flow override aborts."""
1076    provider = MockProvider("test_players", instance_id="test_players--1")
1077    player = MockPlayer(provider, "test_player_1", "Player One")
1078    assert type(player).run_setup_flow is Player.run_setup_flow
1079    with patch.object(flow_mass.players, "get_player", return_value=player):
1080        step = await flow_mass.config.setup_player("test_player_1")
1081    assert step.type == FlowStepType.ABORT
1082    assert step.reason == "nothing_to_configure"
1083
1084
1085async def test_player_setup_abort_mid_finish_restores_setup_data(
1086    flow_mass: MusicAssistant,
1087) -> None:
1088    """Aborting a player flow while finish() is in flight restores the previous setup_data."""
1089    player_id = "test_player_1"
1090    provider = MockProvider("test_players", instance_id="test_players--1")
1091    player = _FlowPlayer(provider, player_id, "Player One")
1092    original_setup_data = {"pin": flow_mass.config.encrypt_string("0000")}
1093    flow_mass.config.set(
1094        f"{CONF_PLAYERS}/{player_id}",
1095        {
1096            "player_id": player_id,
1097            "provider": provider.instance_id,
1098            "enabled": True,
1099            "setup_data": dict(original_setup_data),
1100        },
1101    )
1102    finish_reached = asyncio.Event()
1103
1104    async def hanging_get_player_config(_player_id: str) -> Any:
1105        finish_reached.set()
1106        await asyncio.sleep(3600)
1107
1108    with (
1109        patch.object(flow_mass.players, "get_player", return_value=player),
1110        patch.object(flow_mass.config, "get_player_config", hanging_get_player_config),
1111    ):
1112        step = await flow_mass.config.setup_player(player_id)
1113        assert step.type == FlowStepType.FORM
1114        submit_task = asyncio.create_task(
1115            flow_mass.config.submit_setup_flow(step.flow_id, {"pin": "1234"})
1116        )
1117        await asyncio.wait_for(finish_reached.wait(), timeout=5)
1118        await flow_mass.config.abort_setup_flow(step.flow_id)
1119        submit_step = await submit_task
1120    # the abort interrupted finish(): the previous setup_data was restored
1121    raw_conf = flow_mass.config.get(f"{CONF_PLAYERS}/{player_id}")
1122    assert raw_conf["setup_data"] == original_setup_data
1123    assert submit_step.type == FlowStepType.ABORT
1124
1125
1126async def test_player_setup_reaches_unavailable_player(flow_mass: MusicAssistant) -> None:
1127    """
1128    Player setup must not gate on player availability.
1129
1130    A player that needs setup is serialized as unavailable (available folds in
1131    needs_setup), so requesting availability here would lock out exactly the
1132    players this command exists for.
1133    """
1134    provider = MockProvider("test_players", instance_id="test_players--1")
1135    player = MockPlayer(provider, "test_player_1", "Player One")
1136
1137    def get_player_needs_setup(player_id: str, raise_unavailable: bool = False) -> MockPlayer:
1138        # mimic the real controller for a needs_setup player: state.available is
1139        # False, so raise_unavailable=True would raise PlayerUnavailableError
1140        if raise_unavailable:
1141            raise PlayerUnavailableError(f"Player {player_id} is not available")
1142        return player
1143
1144    with patch.object(flow_mass.players, "get_player", side_effect=get_player_needs_setup):
1145        step = await flow_mass.config.setup_player("test_player_1")
1146    assert step.type == FlowStepType.ABORT
1147    assert step.reason == "nothing_to_configure"
1148
1149
1150async def test_player_setup_flow_finish(flow_mass: MusicAssistant) -> None:
1151    """A player flow persists and applies encrypted setup_data to the active player."""
1152    events: list[MassEvent] = []
1153    flow_mass.subscribe(events.append, EventType.PLAYER_CONFIG_UPDATED)
1154    player_id = "test_player_1"
1155    provider = MockProvider("test_players", instance_id="test_players--1")
1156    player = _FlowPlayer(provider, player_id, "Player One")
1157    flow_mass.config.set(
1158        f"{CONF_PLAYERS}/{player_id}",
1159        {"player_id": player_id, "provider": provider.instance_id, "enabled": True},
1160    )
1161    player_config = PlayerConfig(values={}, provider=provider.instance_id, player_id=player_id)
1162    with (
1163        patch.object(flow_mass.players, "get_player", return_value=player),
1164        patch.object(flow_mass.config, "get_player_config", AsyncMock(return_value=player_config)),
1165    ):
1166        step = await flow_mass.config.setup_player(player_id)
1167        assert step.type == FlowStepType.FORM
1168        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"pin": "1234"})
1169    assert finish_step.type == FlowStepType.FINISH
1170    assert finish_step.result == {"player_id": player_id}
1171    raw_conf = flow_mass.config.get(f"{CONF_PLAYERS}/{player_id}")
1172    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["pin"]) == "1234"
1173    cast("AsyncMock", flow_mass.players.on_player_config_change).assert_awaited_once_with(
1174        player_config, {"setup_data/pin"}
1175    )
1176    config_event = await _wait_for(lambda: next(iter(events), None))
1177    assert config_event.object_id == player_id
1178
1179
1180async def test_player_setup_apply_failure_restores_setup_data(
1181    flow_mass: MusicAssistant,
1182) -> None:
1183    """A player setup failure restores the previous setup_data."""
1184    player_id = "test_player_1"
1185    provider = MockProvider("test_players", instance_id="test_players--1")
1186    player = _FlowPlayer(provider, player_id, "Player One")
1187    original_setup_data = {"pin": flow_mass.config.encrypt_string("0000")}
1188    flow_mass.config.set(
1189        f"{CONF_PLAYERS}/{player_id}",
1190        {
1191            "player_id": player_id,
1192            "provider": provider.instance_id,
1193            "enabled": True,
1194            "setup_data": dict(original_setup_data),
1195        },
1196    )
1197    player_config = PlayerConfig(values={}, provider=provider.instance_id, player_id=player_id)
1198    cast("AsyncMock", flow_mass.players.on_player_config_change).side_effect = RuntimeError(
1199        "apply failed"
1200    )
1201    with (
1202        patch.object(flow_mass.players, "get_player", return_value=player),
1203        patch.object(flow_mass.config, "get_player_config", AsyncMock(return_value=player_config)),
1204    ):
1205        step = await flow_mass.config.setup_player(player_id)
1206        abort_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"pin": "1234"})
1207    assert abort_step.type == FlowStepType.ABORT
1208    assert abort_step.reason == "apply failed"
1209    raw_conf = flow_mass.config.get(f"{CONF_PLAYERS}/{player_id}")
1210    assert raw_conf["setup_data"] == original_setup_data
1211
1212
1213class _ProtocolChildPlayer(MockPlayer):
1214    """A protocol child player that needs setup and implements a pairing flow."""
1215
1216    def __init__(self, provider: MockProvider, player_id: str, name: str) -> None:
1217        super().__init__(provider, player_id, name, player_type=PlayerType.PROTOCOL)
1218        self._attr_needs_setup = True
1219        self._attr_setup_reason = "pairing_required"
1220
1221    async def run_setup_flow(self, session: SetupSession) -> None:
1222        values = await session.form(
1223            [ConfigEntry(key="pin", type=ConfigEntryType.STRING)], step_id="user"
1224        )
1225        await session.finish(values)
1226
1227
1228def _child_output_protocol(child: MockPlayer) -> OutputProtocol:
1229    """Build a (non-native) output protocol entry pointing at the given child player."""
1230    return OutputProtocol(
1231        output_protocol_id=child.player_id,
1232        name=child.display_name,
1233        protocol_domain=child.provider.domain,
1234        is_native=False,
1235        priority=10,
1236        available=False,
1237    )
1238
1239
1240def _set_player_conf(flow_mass: MusicAssistant, player: MockPlayer) -> None:
1241    """Persist a minimal player config for the given player."""
1242    flow_mass.config.set(
1243        f"{CONF_PLAYERS}/{player.player_id}",
1244        {
1245            "player_id": player.player_id,
1246            "provider": player.provider.instance_id,
1247            "enabled": True,
1248        },
1249    )
1250
1251
1252async def test_player_setup_delegates_to_single_child(flow_mass: MusicAssistant) -> None:
1253    """A wrapper player with one protocol child needing setup runs that child's flow."""
1254    parent_provider = MockProvider("universal_player", instance_id="universal_player")
1255    child_provider = MockProvider("airplay", instance_id="airplay")
1256    parent = MockPlayer(parent_provider, "up_parent", "Living Room")
1257    child = _ProtocolChildPlayer(child_provider, "ap_child", "Living Room AirPlay")
1258    # native option (skipped) + the protocol child that needs setup
1259    parent._cache["output_protocols"] = [
1260        OutputProtocol(
1261            output_protocol_id="native", name="", protocol_domain="", is_native=True, available=True
1262        ),
1263        _child_output_protocol(child),
1264    ]
1265    _set_player_conf(flow_mass, parent)
1266    _set_player_conf(flow_mass, child)
1267    players = {parent.player_id: parent, child.player_id: child}
1268    child_config = PlayerConfig(
1269        values={}, provider=child_provider.instance_id, player_id=child.player_id
1270    )
1271    with (
1272        patch.object(
1273            flow_mass.players, "get_player", side_effect=lambda pid, *_a, **_k: players.get(pid)
1274        ),
1275        patch.object(flow_mass.config, "get_player_config", AsyncMock(return_value=child_config)),
1276    ):
1277        step = await flow_mass.config.setup_player(parent.player_id)
1278        assert step.type == FlowStepType.FORM
1279        assert step.step_id == "user"
1280        # the flow runs as the CHILD: its steps localize under the child provider
1281        assert step.translation_owner == "provider.airplay"
1282        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"pin": "4321"})
1283    assert finish_step.type == FlowStepType.FINISH
1284    # persisted to the CHILD's config, not the parent's
1285    assert finish_step.result == {"player_id": child.player_id}
1286    child_raw = flow_mass.config.get(f"{CONF_PLAYERS}/{child.player_id}")
1287    assert flow_mass.config.decrypt_string(child_raw["setup_data"]["pin"]) == "4321"
1288    assert "setup_data" not in flow_mass.config.get(f"{CONF_PLAYERS}/{parent.player_id}")
1289
1290
1291async def test_player_setup_multi_child_selection(flow_mass: MusicAssistant) -> None:
1292    """A wrapper player with several children needing setup first asks which to set up."""
1293    parent_provider = MockProvider("universal_player", instance_id="universal_player")
1294    ap_provider = MockProvider("airplay", instance_id="airplay")
1295    ss_provider = MockProvider("sendspin", instance_id="sendspin")
1296    parent = MockPlayer(parent_provider, "up_parent", "Kitchen")
1297    child_a = _ProtocolChildPlayer(ap_provider, "ap_child", "Kitchen AirPlay")
1298    child_b = _ProtocolChildPlayer(ss_provider, "ss_child", "Kitchen Sendspin")
1299    parent._cache["output_protocols"] = [
1300        _child_output_protocol(child_a),
1301        _child_output_protocol(child_b),
1302    ]
1303    for player in (parent, child_a, child_b):
1304        _set_player_conf(flow_mass, player)
1305    players = {p.player_id: p for p in (parent, child_a, child_b)}
1306
1307    def _get_config(pid: str) -> PlayerConfig:
1308        return PlayerConfig(values={}, provider=players[pid].provider.instance_id, player_id=pid)
1309
1310    with (
1311        patch.object(
1312            flow_mass.players, "get_player", side_effect=lambda pid, *_a, **_k: players.get(pid)
1313        ),
1314        patch.object(flow_mass.config, "get_player_config", AsyncMock(side_effect=_get_config)),
1315    ):
1316        step = await flow_mass.config.setup_player(parent.player_id)
1317        # first a parent-owned selection form listing both children
1318        assert step.type == FlowStepType.FORM
1319        assert step.step_id == "select_child"
1320        assert step.translation_owner == "provider.universal_player"
1321        child_entry = next(entry for entry in step.entries if entry.key == "child")
1322        assert {option.value for option in child_entry.options} == {"ap_child", "ss_child"}
1323        # pick the sendspin child -> the flow retargets and runs the child's flow
1324        next_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"child": "ss_child"})
1325        assert next_step.type == FlowStepType.FORM
1326        assert next_step.step_id == "user"
1327        assert next_step.translation_owner == "provider.sendspin"
1328        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"pin": "9999"})
1329    assert finish_step.type == FlowStepType.FINISH
1330    assert finish_step.result == {"player_id": "ss_child"}
1331    child_raw = flow_mass.config.get(f"{CONF_PLAYERS}/ss_child")
1332    assert flow_mass.config.decrypt_string(child_raw["setup_data"]["pin"]) == "9999"
1333
1334
1335async def test_player_setup_reruns_child_flow_when_nothing_needs_setup(
1336    flow_mass: MusicAssistant,
1337) -> None:
1338    """A wrapper player whose children are all set up still reaches the child's flow."""
1339    parent_provider = MockProvider("universal_player", instance_id="universal_player")
1340    child_provider = MockProvider("airplay", instance_id="airplay")
1341    parent = MockPlayer(parent_provider, "up_parent", "Bedroom")
1342    child = _ProtocolChildPlayer(child_provider, "ap_child", "Bedroom AirPlay")
1343    child._attr_needs_setup = False  # already set up: this is an on-demand re-run
1344    parent._cache["output_protocols"] = [_child_output_protocol(child)]
1345    _set_player_conf(flow_mass, parent)
1346    _set_player_conf(flow_mass, child)
1347    players = {parent.player_id: parent, child.player_id: child}
1348    child_config = PlayerConfig(
1349        values={}, provider=child_provider.instance_id, player_id=child.player_id
1350    )
1351    with (
1352        patch.object(
1353            flow_mass.players, "get_player", side_effect=lambda pid, *_a, **_k: players.get(pid)
1354        ),
1355        patch.object(flow_mass.config, "get_player_config", AsyncMock(return_value=child_config)),
1356    ):
1357        step = await flow_mass.config.setup_player(parent.player_id)
1358        assert step.type == FlowStepType.FORM
1359        assert step.step_id == "user"
1360        assert step.translation_owner == "provider.airplay"
1361        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"pin": "1234"})
1362    assert finish_step.type == FlowStepType.FINISH
1363    assert finish_step.result == {"player_id": child.player_id}
1364
1365
1366async def test_player_setup_without_flow_capable_children_aborts(
1367    flow_mass: MusicAssistant,
1368) -> None:
1369    """A wrapper player whose children offer no flow at all reports nothing to configure."""
1370    parent_provider = MockProvider("universal_player", instance_id="universal_player")
1371    child_provider = MockProvider("dlna", instance_id="dlna")
1372    parent = MockPlayer(parent_provider, "up_parent", "Study")
1373    child = MockPlayer(child_provider, "dlna_child", "Study DLNA", player_type=PlayerType.PROTOCOL)
1374    parent._cache["output_protocols"] = [_child_output_protocol(child)]
1375    _set_player_conf(flow_mass, parent)
1376    players = {parent.player_id: parent, child.player_id: child}
1377    with patch.object(
1378        flow_mass.players, "get_player", side_effect=lambda pid, *_a, **_k: players.get(pid)
1379    ):
1380        step = await flow_mass.config.setup_player(parent.player_id)
1381    assert step.type == FlowStepType.ABORT
1382    assert step.reason == "nothing_to_configure"
1383
1384
1385async def test_player_setup_reason_in_state_fingerprint() -> None:
1386    """setup_reason is exposed on the player and tracked in the state fingerprint."""
1387    provider = MockProvider("test_players", instance_id="test_players--1")
1388    player = MockPlayer(provider, "sr_player", "Reason Player")
1389    player._attr_needs_setup = True
1390    player._attr_setup_reason = "pairing_required"
1391    assert player.setup_reason == "pairing_required"
1392    # the reason is a tracked leaf of the state fingerprint (so it drives change events)
1393    player_state = player.state
1394    player_state.setup_reason = "pairing_required"
1395    assert _state_fingerprint(player_state)["setup_reason"] == "pairing_required"
1396
1397
1398async def test_has_setup_flow_serialized_for_own_flow() -> None:
1399    """A player implementing its own flow serializes has_setup_flow (regardless of needs_setup)."""
1400    provider = MockProvider("sendspin", instance_id="sendspin")
1401    plain = MockPlayer(provider, "plain_player", "Plain Player")
1402    player = _FlowPlayer(provider, "flow_player", "Flow Player")
1403    assert plain.has_setup_flow is False
1404    assert player.has_setup_flow is True
1405    # not gated on needs_setup: the flow stays re-runnable once setup completed
1406    assert player.needs_setup is False
1407    player.update_state(force_update=True)
1408    assert player.state.has_setup_flow is True
1409    assert player.to_dict()["has_setup_flow"] is True
1410    # tracked leaf of the fingerprint, so a late-binding protocol child drives an event
1411    assert _state_fingerprint(player.state)["has_setup_flow"] is True
1412
1413
1414async def test_has_setup_flow_serialized_for_protocol_child() -> None:
1415    """A wrapper player inherits has_setup_flow from a linked protocol child with a flow."""
1416    parent_provider = MockProvider("universal_player", instance_id="universal_player")
1417    child_provider = MockProvider("airplay", instance_id="airplay")
1418    parent = MockPlayer(parent_provider, "up_parent", "Hallway")
1419    child = _ProtocolChildPlayer(child_provider, "ap_child", "Hallway AirPlay")
1420    child._attr_needs_setup = False  # already set up, but its flow can be re-run
1421    players = {parent.player_id: parent, child.player_id: child}
1422    # link the child: set_linked_output_protocols survives the update_state cache flush
1423    parent.set_linked_output_protocols([_child_output_protocol(child)])
1424    with patch.object(
1425        parent.mass.players, "get_player", side_effect=lambda pid, *_a, **_k: players.get(pid)
1426    ):
1427        assert parent.has_setup_flow is True
1428        parent.update_state(force_update=True)
1429    assert parent.to_dict()["has_setup_flow"] is True
1430    assert _state_fingerprint(parent.state)["has_setup_flow"] is True
1431
1432
1433async def test_secure_values_never_echoed_on_step(flow_mass: MusicAssistant) -> None:
1434    """Submitted SECURE_STRING values are handed to the flow but never kept on the step."""
1435    received: dict[str, Any] = {}
1436
1437    async def run_setup(session: SetupSession) -> None:
1438        received.update(await session.form([USERNAME_ENTRY, PASSWORD_ENTRY]))
1439        # keep the flow alive on a second form so the stored step can be inspected
1440        await session.form([USERNAME_ENTRY], step_id="second")
1441        await session.finish({})
1442
1443    with _use_flow(flow_mass, run_setup):
1444        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1445        # validation failure path: the password must not be echoed back
1446        error_step = await flow_mass.config.submit_setup_flow(step.flow_id, {"password": "sssh"})
1447        password_entry = next(x for x in error_step.entries if x.key == "password")
1448        assert password_entry.value is None
1449        # success path: the flow receives the value, the stored step does not keep it
1450        await flow_mass.config.submit_setup_flow(
1451            step.flow_id, {"username": "marcel", "password": "sssh"}
1452        )
1453        assert received["password"] == "sssh"
1454        await flow_mass.config.abort_setup_flow(step.flow_id)
1455
1456
1457# --- real provider setup flows driven through the engine (migrated plain-cred providers) ---
1458
1459
1460async def test_real_provider_flow_qobuz(flow_mass: MusicAssistant) -> None:
1461    """Qobuz's run_setup collects credentials and persists them as (encrypted) setup_data."""
1462    with (
1463        _use_flow(flow_mass, qobuz_run_setup),
1464        patch.object(flow_mass, "load_provider_config", AsyncMock()) as mock_load,
1465    ):
1466        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1467        assert step.type == FlowStepType.FORM
1468        # only the setup credentials are on the form; the quality option stays behind
1469        assert {entry.key for entry in step.entries} == {"username", "password"}
1470        finish_step = await flow_mass.config.submit_setup_flow(
1471            step.flow_id, {"username": "marcel", "password": "hunter2"}
1472        )
1473    assert finish_step.type == FlowStepType.FINISH
1474    mock_load.assert_awaited_once()
1475    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
1476    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["username"]) == "marcel"
1477    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["password"]) == "hunter2"
1478
1479
1480async def test_real_provider_flow_filesystem_local(flow_mass: MusicAssistant) -> None:
1481    """filesystem_local's run_setup collects the content type and path as setup_data."""
1482    with (
1483        _use_flow(flow_mass, filesystem_local_run_setup),
1484        patch.object(flow_mass, "load_provider_config", AsyncMock()) as mock_load,
1485    ):
1486        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1487        assert step.type == FlowStepType.FORM
1488        assert {"content_type", "path"} <= {entry.key for entry in step.entries}
1489        finish_step = await flow_mass.config.submit_setup_flow(
1490            step.flow_id, {"content_type": "podcasts", "path": "/media/podcasts"}
1491        )
1492    assert finish_step.type == FlowStepType.FINISH
1493    mock_load.assert_awaited_once()
1494    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
1495    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["content_type"]) == "podcasts"
1496    assert flow_mass.config.decrypt_string(raw_conf["setup_data"]["path"]) == "/media/podcasts"
1497
1498
1499async def test_real_provider_flow_retry_on_error(flow_mass: MusicAssistant) -> None:
1500    """A failing load makes the author's loop re-render the form with the error, then succeed."""
1501    load_mock = AsyncMock(side_effect=[LoginFailed("bad creds"), None])
1502    with (
1503        _use_flow(flow_mass, qobuz_run_setup),
1504        patch.object(flow_mass, "load_provider_config", load_mock),
1505    ):
1506        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1507        retry_step = await flow_mass.config.submit_setup_flow(
1508            step.flow_id, {"username": "marcel", "password": "wrong"}
1509        )
1510        assert retry_step.type == FlowStepType.FORM
1511        # the error slug (LoginFailed.translation_key) is used so it localizes
1512        # at serialization; the raw message is the fallback for keyless errors
1513        assert retry_step.errors == {"base": "login_failed"}
1514        finish_step = await flow_mass.config.submit_setup_flow(
1515            step.flow_id, {"username": "marcel", "password": "right"}
1516        )
1517    assert finish_step.type == FlowStepType.FINISH
1518    assert flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}") is not None
1519
1520
1521async def test_spotify_flow_hosted_bounce_roundtrip(
1522    flow_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
1523) -> None:
1524    """The real Spotify flow: hosted-bounce external auth then a stored refresh token."""
1525    from music_assistant.providers.spotify.constants import (  # noqa: PLC0415
1526        CONF_REFRESH_TOKEN_GLOBAL,
1527    )
1528    from music_assistant.providers.spotify.setup_flow import run_setup  # noqa: PLC0415
1529
1530    monkeypatch.setattr(
1531        "music_assistant.providers.spotify.setup_flow.app_var", lambda _key: "ma_client_id"
1532    )
1533    # seed the lazy http_session backing field so the token exchange uses our stub
1534    monkeypatch.setattr(
1535        flow_mass, "_http_session", _fake_json_session({"refresh_token": "rt_global"})
1536    )
1537    with (
1538        _use_flow(flow_mass, run_setup),
1539        patch.object(flow_mass, "load_provider_config", AsyncMock()),
1540    ):
1541        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1542        # first step is the (required) global authentication via the hosted bounce
1543        assert step.type == FlowStepType.EXTERNAL
1544        assert step.step_id == "authenticate"
1545        assert step.url is not None
1546        assert "accounts.spotify.com/authorize" in step.url
1547        # the fixed hosted redirect is used, with the local callback carried in `state`
1548        assert "music-assistant.io%2Fcallback" in step.url
1549        assert step.flow_id in step.url
1550        session = flow_mass.config._setup_flows[step.flow_id].session
1551        await _fire_callback(flow_mass, step.flow_id, "code=auth_code&state=xyz")
1552        # after the token exchange the optional developer step is shown
1553        await _wait_for(
1554            lambda: session.current_step is not None and session.current_step.step_id == "developer"
1555        )
1556        # skipping the developer client id (blank) finishes the flow
1557        finish_step = await flow_mass.config.submit_setup_flow(step.flow_id, {})
1558    assert finish_step.type == FlowStepType.FINISH
1559    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
1560    assert (
1561        flow_mass.config.decrypt_string(raw_conf["setup_data"][CONF_REFRESH_TOKEN_GLOBAL])
1562        == "rt_global"
1563    )
1564
1565
1566async def test_gdrive_flow_form_then_hosted_bounce(
1567    flow_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
1568) -> None:
1569    """The real Google Drive flow: credential form, hosted-bounce OAuth, stored refresh token."""
1570    from music_assistant.providers.filesystem_cloud.base import (  # noqa: PLC0415
1571        CONF_CLIENT_ID,
1572        CONF_CLIENT_SECRET,
1573        CONF_FOLDER_ID,
1574        CONF_REFRESH_TOKEN,
1575    )
1576    from music_assistant.providers.filesystem_google_drive.setup_flow import (  # noqa: PLC0415
1577        run_setup,
1578    )
1579
1580    # seed the lazy http_session backing field so the token exchange uses our stub
1581    monkeypatch.setattr(
1582        flow_mass, "_http_session", _fake_json_session({"refresh_token": "rt_drive"})
1583    )
1584    with (
1585        _use_flow(flow_mass, run_setup),
1586        patch.object(flow_mass, "load_provider_config", AsyncMock()),
1587    ):
1588        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1589        # the shared cloud flow first collects content type + OAuth client credentials + folder
1590        assert step.type == FlowStepType.FORM
1591        assert step.step_id == "user"
1592        session = flow_mass.config._setup_flows[step.flow_id].session
1593        external_step = await flow_mass.config.submit_setup_flow(
1594            step.flow_id,
1595            {
1596                "content_type": "music",
1597                CONF_CLIENT_ID: "cid",
1598                CONF_CLIENT_SECRET: "secret",
1599                CONF_FOLDER_ID: "root",
1600            },
1601        )
1602        assert external_step.type == FlowStepType.EXTERNAL
1603        assert external_step.url is not None
1604        auth_url = urlsplit(external_step.url)
1605        assert auth_url.netloc == "accounts.google.com"
1606        assert parse_qs(auth_url.query)["redirect_uri"] == ["https://music-assistant.io/callback"]
1607        await _fire_callback(flow_mass, step.flow_id, "code=auth_code")
1608        await _wait_for(lambda: session.finished)
1609    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
1610    setup_data = raw_conf["setup_data"]
1611    assert flow_mass.config.decrypt_string(setup_data[CONF_REFRESH_TOKEN]) == "rt_drive"
1612    assert flow_mass.config.decrypt_string(setup_data[CONF_CLIENT_ID]) == "cid"
1613    assert flow_mass.config.decrypt_string(setup_data[CONF_FOLDER_ID]) == "root"
1614
1615
1616async def test_tidal_flow_pkce_url_paste(
1617    flow_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
1618) -> None:
1619    """The real Tidal flow: paste the redirect URL, exchange it, store the tokens."""
1620    from music_assistant.providers.tidal.auth_manager import TidalAuthManager  # noqa: PLC0415
1621    from music_assistant.providers.tidal.constants import (  # noqa: PLC0415
1622        CONF_AUTH_TOKEN,
1623        CONF_OOPS_URL,
1624        CONF_REFRESH_TOKEN,
1625        CONF_USER_ID,
1626    )
1627    from music_assistant.providers.tidal.setup_flow import run_setup  # noqa: PLC0415
1628
1629    monkeypatch.setattr(flow_mass, "_http_session", MagicMock())
1630    auth_data = {
1631        "access_token": "at-123",
1632        "refresh_token": "rt-456",
1633        "expires_at": 4102444800.0,
1634        "userId": 42,
1635    }
1636    with (
1637        _use_flow(flow_mass, run_setup),
1638        patch.object(
1639            TidalAuthManager,
1640            "build_pkce_login",
1641            return_value=("https://login.tidal.com/authorize?x=1", {"code_verifier": "v"}),
1642        ),
1643        patch.object(
1644            TidalAuthManager, "process_pkce_login", AsyncMock(return_value=auth_data)
1645        ) as mock_exchange,
1646        patch.object(flow_mass, "load_provider_config", AsyncMock()),
1647    ):
1648        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1649        assert step.type == FlowStepType.FORM
1650        assert step.step_id == "user"
1651        # the authorize link rides along on the instructions label
1652        instructions = next(x for x in step.entries if x.key == "auth_instructions")
1653        assert instructions.help_link == "https://login.tidal.com/authorize?x=1"
1654        finish_step = await flow_mass.config.submit_setup_flow(
1655            step.flow_id,
1656            {CONF_OOPS_URL: "https://tidal.com/android/login/auth?code=abc"},
1657        )
1658    assert finish_step.type == FlowStepType.FINISH
1659    # the pasted redirect URL was handed to the token exchange
1660    assert mock_exchange.await_args is not None
1661    assert mock_exchange.await_args.args[2] == "https://tidal.com/android/login/auth?code=abc"
1662    raw_conf = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")
1663    setup_data = raw_conf["setup_data"]
1664    assert flow_mass.config.decrypt_string(setup_data[CONF_AUTH_TOKEN]) == "at-123"
1665    assert flow_mass.config.decrypt_string(setup_data[CONF_REFRESH_TOKEN]) == "rt-456"
1666    assert setup_data["expiry_time"] == 4102444800.0
1667    assert flow_mass.config.decrypt_string(setup_data[CONF_USER_ID]) == "42"
1668
1669
1670async def test_tidal_flow_exchange_error_retries(
1671    flow_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
1672) -> None:
1673    """A failed token exchange re-renders the form with an error, then succeeds on retry."""
1674    from music_assistant_models.errors import LoginFailed  # noqa: PLC0415
1675
1676    from music_assistant.providers.tidal.auth_manager import TidalAuthManager  # noqa: PLC0415
1677    from music_assistant.providers.tidal.constants import CONF_OOPS_URL  # noqa: PLC0415
1678    from music_assistant.providers.tidal.setup_flow import run_setup  # noqa: PLC0415
1679
1680    monkeypatch.setattr(flow_mass, "_http_session", MagicMock())
1681    exchange = AsyncMock(
1682        side_effect=[
1683            LoginFailed("No authorization code found in redirect URL"),
1684            {
1685                "access_token": "at",
1686                "refresh_token": "rt",
1687                "expires_at": 1.0,
1688                "userId": "u",
1689            },
1690        ]
1691    )
1692    with (
1693        _use_flow(flow_mass, run_setup),
1694        patch.object(
1695            TidalAuthManager, "build_pkce_login", return_value=("https://login.tidal.com/x", {})
1696        ),
1697        patch.object(TidalAuthManager, "process_pkce_login", exchange),
1698        patch.object(flow_mass, "load_provider_config", AsyncMock()),
1699    ):
1700        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1701        retry_step = await flow_mass.config.submit_setup_flow(
1702            step.flow_id, {CONF_OOPS_URL: "https://tidal.com/android/login/auth"}
1703        )
1704        assert retry_step.type == FlowStepType.FORM
1705        # the canonical retry pattern surfaces the error's translation key
1706        assert retry_step.errors == {"base": "login_failed"}
1707        finish_step = await flow_mass.config.submit_setup_flow(
1708            step.flow_id, {CONF_OOPS_URL: "https://tidal.com/android/login/auth?code=abc"}
1709        )
1710    assert finish_step.type == FlowStepType.FINISH
1711
1712
1713async def test_hue_pairing_flow_retry_then_success(
1714    flow_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
1715) -> None:
1716    """The real Hue flow: a missed button press re-forms, then pairing succeeds."""
1717    from music_assistant.providers.hue_entertainment.constants import (  # noqa: PLC0415
1718        CONF_BRIDGE_HOST,
1719        CONF_BRIDGE_ID,
1720        CONF_CLIENTKEY,
1721        CONF_USERNAME,
1722    )
1723    from music_assistant.providers.hue_entertainment.setup_flow import run_setup  # noqa: PLC0415
1724
1725    api_instance = MagicMock()
1726    # first pair attempt times out (button not pressed), second succeeds
1727    api_instance.pair = AsyncMock(
1728        side_effect=[TimeoutError("not pressed"), {"username": "abc", "clientkey": "def"}]
1729    )
1730    api_instance.get_bridge_id = AsyncMock(return_value="bridge-1")
1731    api_instance.close = AsyncMock()
1732    monkeypatch.setattr(
1733        "music_assistant.providers.hue_entertainment.setup_flow.HueEntertainmentAPI",
1734        MagicMock(return_value=api_instance),
1735    )
1736    with (
1737        _use_flow(flow_mass, run_setup),
1738        patch.object(flow_mass, "load_provider_config", AsyncMock()),
1739    ):
1740        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1741        assert step.type == FlowStepType.FORM
1742        assert step.step_id == "user"
1743        session = flow_mass.config._setup_flows[step.flow_id].session
1744        await flow_mass.config.submit_setup_flow(step.flow_id, {CONF_BRIDGE_HOST: "1.2.3.4"})
1745        # the missed button press loops back to the user form with an error
1746        retry = await _wait_for(
1747            lambda: (
1748                session.current_step
1749                if session.current_step
1750                and session.current_step.step_id == "user"
1751                and session.current_step.errors
1752                else None
1753            )
1754        )
1755        assert retry.errors == {"base": "button_not_pressed"}
1756        await flow_mass.config.submit_setup_flow(step.flow_id, {CONF_BRIDGE_HOST: "1.2.3.4"})
1757        await _wait_for(lambda: session.finished)
1758    setup_data = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")["setup_data"]
1759    assert flow_mass.config.decrypt_string(setup_data[CONF_USERNAME]) == "abc"
1760    assert flow_mass.config.decrypt_string(setup_data[CONF_CLIENTKEY]) == "def"
1761    assert flow_mass.config.decrypt_string(setup_data[CONF_BRIDGE_HOST]) == "1.2.3.4"
1762    assert flow_mass.config.decrypt_string(setup_data[CONF_BRIDGE_ID]) == "bridge-1"
1763
1764
1765async def test_netease_qr_flow_expiry_then_confirm(
1766    flow_mass: MusicAssistant, monkeypatch: pytest.MonkeyPatch
1767) -> None:
1768    """The real NetEase QR flow: the first shown QR expires and is refreshed, then confirmed."""
1769    from music_assistant.models.setup_flow import StepExpiredError  # noqa: PLC0415
1770    from music_assistant.providers.neteasecloudmusic import setup_flow as nc_flow  # noqa: PLC0415
1771    from music_assistant.providers.neteasecloudmusic.constants import (  # noqa: PLC0415
1772        CONF_API_BASE_URL,
1773        CONF_COOKIE,
1774        CONF_UID,
1775    )
1776
1777    monkeypatch.setattr(flow_mass, "_http_session", MagicMock())
1778    monkeypatch.setattr(nc_flow, "NcmApiClient", MagicMock())
1779    # a fresh QR is minted for the initial show and again after the first one expires
1780    create_qr = AsyncMock(
1781        side_effect=[("key1", "data:image/png;base64,AAA"), ("key2", "data:image/png;base64,BBB")]
1782    )
1783    monkeypatch.setattr(nc_flow, "_create_qr", create_qr)
1784    # first poll hits the expiry deadline, the second (refreshed) QR is confirmed
1785    monkeypatch.setattr(
1786        nc_flow,
1787        "_poll_qr_login",
1788        AsyncMock(side_effect=[StepExpiredError(), ("cookie-xyz", "42")]),
1789    )
1790    with (
1791        _use_flow(flow_mass, nc_flow.run_setup),
1792        patch.object(flow_mass, "load_provider_config", AsyncMock()),
1793    ):
1794        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1795        assert step.type == FlowStepType.FORM
1796        assert step.step_id == "user"
1797        session = flow_mass.config._setup_flows[step.flow_id].session
1798        await flow_mass.config.submit_setup_flow(
1799            step.flow_id, {CONF_API_BASE_URL: "http://127.0.0.1:3000"}
1800        )
1801        await _wait_for(lambda: session.finished)
1802    # the QR was minted twice (initial + one in-place refresh after expiry)
1803    assert create_qr.await_count == 2
1804    setup_data = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")["setup_data"]
1805    assert flow_mass.config.decrypt_string(setup_data[CONF_COOKIE]) == "cookie-xyz"
1806    assert flow_mass.config.decrypt_string(setup_data[CONF_UID]) == "42"
1807    assert flow_mass.config.decrypt_string(setup_data[CONF_API_BASE_URL]) == "http://127.0.0.1:3000"
1808
1809
1810async def test_external_until_completes_on_awaitable(flow_mass: MusicAssistant) -> None:
1811    """external_until shows an external step and completes on the awaitable (no callback)."""
1812    release = asyncio.Event()
1813
1814    async def _work() -> dict[str, str]:
1815        await release.wait()
1816        return {"token": "abc"}
1817
1818    async def run_setup(session: SetupSession) -> None:
1819        result = await session.external_until(
1820            _work(),
1821            url="https://example.com/device",
1822            step_id="device",
1823            expires_in=300,
1824        )
1825        await session.finish({"token": result["token"]})
1826
1827    with (
1828        _use_flow(flow_mass, run_setup),
1829        patch.object(flow_mass, "load_provider_config", AsyncMock()),
1830    ):
1831        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1832        assert step.type == FlowStepType.EXTERNAL
1833        assert step.step_id == "device"
1834        assert step.url == "https://example.com/device"
1835        session = flow_mass.config._setup_flows[step.flow_id].session
1836        # completion is driven by the awaitable resolving, not a browser callback
1837        release.set()
1838        await _wait_for(lambda: session.finished)
1839    setup_data = flow_mass.config.get(f"{CONF_PROVIDERS}/{FAKE_DOMAIN}")["setup_data"]
1840    assert flow_mass.config.decrypt_string(setup_data["token"]) == "abc"
1841
1842
1843async def test_external_until_raises_on_deadline(flow_mass: MusicAssistant) -> None:
1844    """external_until raises StepExpiredError when the awaitable outlives expires_in."""
1845    expired = asyncio.Event()
1846
1847    async def _never() -> None:
1848        await asyncio.Event().wait()
1849
1850    async def run_setup(session: SetupSession) -> None:
1851        try:
1852            await session.external_until(
1853                _never(),
1854                url="https://example.com/device",
1855                step_id="device",
1856                expires_in=0.1,
1857            )
1858        except StepExpiredError:
1859            expired.set()
1860            raise
1861
1862    with (
1863        _use_flow(flow_mass, run_setup),
1864        patch.object(flow_mass, "load_provider_config", AsyncMock()),
1865    ):
1866        step = await flow_mass.config.setup_provider(FAKE_DOMAIN)
1867        assert step.type == FlowStepType.EXTERNAL
1868        await _wait_for(lambda: expired.is_set())
1869