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