music-assistant-server

1"""Tests for the Plex Connect interactive setup flow (run_setup)."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from typing import Any
8from unittest import mock
9
10import pytest
11from music_assistant_models.enums import FlowStepType
12
13from music_assistant.models.setup_flow import AbortFlow, SetupFlowContext, SetupSession
14from music_assistant.providers.plex_connect import CONF_MASS_PLAYER_ID, CONF_PLEX_PROVIDER_ID
15from music_assistant.providers.plex_connect import setup_flow as pc_flow
16
17
18def _provider(instance_id: str, name: str) -> mock.Mock:
19    """Build a stand-in for a loaded plex provider instance."""
20    provider = mock.Mock()
21    provider.instance_id = instance_id
22    provider.name = name
23    return provider
24
25
26def _player(player_id: str, display_name: str) -> mock.Mock:
27    """Build a stand-in for a Music Assistant player."""
28    player = mock.Mock()
29    player.player_id = player_id
30    player.display_name = display_name
31    return player
32
33
34def _make_session(
35    finish_handler: Any,
36    plex_providers: list[mock.Mock] | None = None,
37    players: list[mock.Mock] | None = None,
38    setup_data: dict[str, Any] | None = None,
39    values: dict[str, Any] | None = None,
40) -> SetupSession:
41    """Build a real SetupSession backed by a Mock mass with the given live state."""
42    mass = mock.Mock()
43    mass.get_provider_instances = mock.Mock(
44        return_value=plex_providers
45        if plex_providers is not None
46        else [_provider("plex--1", "Plex")]
47    )
48    mass.players.all_players = mock.Mock(
49        return_value=players if players is not None else [_player("player1", "Kitchen")]
50    )
51    context = SetupFlowContext(
52        kind="setup",
53        reason="user",
54        domain="plex_connect",
55        setup_data=setup_data or {},
56        values=values or {},
57    )
58    return SetupSession(mass, "flow-test", context, finish_handler)
59
60
61async def _await_user_form(session: SetupSession) -> Any:
62    """Wait until the user form is presented and return the step."""
63    deadline = time.monotonic() + 5.0
64    while time.monotonic() < deadline:
65        step = session.current_step
66        if step is not None and step.type == FlowStepType.FORM:
67            return step
68        await asyncio.sleep(0.01)
69    raise AssertionError("form step not published within timeout")
70
71
72async def _wait_for_finish(session: SetupSession) -> None:
73    """Wait until the flow finished."""
74    deadline = time.monotonic() + 5.0
75    while time.monotonic() < deadline:
76        if session.finished:
77            return
78        await asyncio.sleep(0.01)
79    raise AssertionError("flow did not finish within timeout")
80
81
82async def test_collects_plex_provider_and_player() -> None:
83    """The single form step collects both selections into setup_data."""
84    collected: dict[str, Any] = {}
85
86    async def finish(_s: SetupSession, values: dict[str, Any]) -> dict[str, str]:
87        collected.update(values)
88        return {"instance_id": "plex_connect--1"}
89
90    session = _make_session(
91        finish,
92        plex_providers=[_provider("plex--1", "Plex"), _provider("plex--2", "Plex Audiobooks")],
93        players=[_player("player2", "living room"), _player("player1", "Kitchen")],
94    )
95    task = asyncio.create_task(pc_flow.run_setup(session))
96    step = await _await_user_form(session)
97    # players are offered sorted case-insensitively by display name
98    assert [option.value for option in step.entries[1].options] == ["player1", "player2"]
99    session.handle_submit({CONF_PLEX_PROVIDER_ID: "plex--2", CONF_MASS_PLAYER_ID: "player2"})
100    await _wait_for_finish(session)
101    await task
102
103    assert collected == {CONF_PLEX_PROVIDER_ID: "plex--2", CONF_MASS_PLAYER_ID: "player2"}
104
105
106async def test_prefills_stored_selection() -> None:
107    """A reconfigure re-shows the stored selection, ignoring values that no longer exist."""
108
109    async def finish(_s: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
110        return {"instance_id": "plex_connect--1"}
111
112    session = _make_session(
113        finish,
114        plex_providers=[_provider("plex--1", "Plex"), _provider("plex--2", "Plex Audiobooks")],
115        players=[_player("player1", "Kitchen"), _player("player2", "Living Room")],
116        setup_data={CONF_PLEX_PROVIDER_ID: "plex--2", CONF_MASS_PLAYER_ID: "gone"},
117    )
118    task = asyncio.create_task(pc_flow.run_setup(session))
119    step = await _await_user_form(session)
120
121    assert step.entries[0].value == "plex--2"
122    # the stored player disappeared: fall back to the first available option
123    assert step.entries[1].value == "player1"
124
125    task.cancel()
126    with pytest.raises(asyncio.CancelledError):
127        await task
128
129
130async def test_prefills_from_legacy_option_values() -> None:
131    """Installs made before this flow existed keep their selection from the config values."""
132
133    async def finish(_s: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
134        return {"instance_id": "plex_connect--1"}
135
136    session = _make_session(
137        finish,
138        plex_providers=[_provider("plex--1", "Plex"), _provider("plex--2", "Plex Audiobooks")],
139        players=[_player("player1", "Kitchen"), _player("player2", "Living Room")],
140        values={CONF_PLEX_PROVIDER_ID: "plex--2", CONF_MASS_PLAYER_ID: "player2"},
141    )
142    task = asyncio.create_task(pc_flow.run_setup(session))
143    step = await _await_user_form(session)
144
145    assert step.entries[0].value == "plex--2"
146    assert step.entries[1].value == "player2"
147
148    task.cancel()
149    with pytest.raises(asyncio.CancelledError):
150        await task
151
152
153async def test_aborts_without_loaded_plex_provider() -> None:
154    """Nothing to pick from: the flow aborts instead of showing an empty dropdown."""
155
156    async def finish(_s: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
157        raise AssertionError("finish must not be called")
158
159    session = _make_session(finish, plex_providers=[])
160    with pytest.raises(AbortFlow) as err:
161        await pc_flow.run_setup(session)
162
163    assert err.value.reason == "no_plex_provider"
164
165
166async def test_aborts_without_players() -> None:
167    """No player to expose: the flow aborts instead of showing an empty dropdown."""
168
169    async def finish(_s: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
170        raise AssertionError("finish must not be called")
171
172    session = _make_session(finish, players=[])
173    with pytest.raises(AbortFlow) as err:
174        await pc_flow.run_setup(session)
175
176    assert err.value.reason == "no_players"
177