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