music-assistant-server

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