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