/
/
1"""Tests for receiver-style plugin setup flows."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from typing import TYPE_CHECKING, Any
8from unittest import mock
9
10import pytest
11from music_assistant_models.enums import FlowStepType
12
13from music_assistant.constants import CONF_BIND_IP, CONF_BIND_PORT
14from music_assistant.models.setup_flow import AbortFlow, SetupFlowContext, SetupSession
15from music_assistant.providers.ariacast_receiver import (
16 CONF_MASS_PLAYER_ID as ARIACAST_PLAYER_ID,
17)
18from music_assistant.providers.ariacast_receiver import setup_flow as ariacast_flow
19from music_assistant.providers.vban_receiver import setup_flow as vban_flow
20from music_assistant.providers.vban_receiver.constants import (
21 CONF_AUDIO_CHANNELS,
22 CONF_PCM_AUDIO_FORMAT,
23 CONF_PCM_SAMPLE_RATE,
24 CONF_SENDER_HOST,
25 CONF_VBAN_STREAM_NAME,
26)
27
28if TYPE_CHECKING:
29 from music_assistant_models.config_entries import ConfigValueType
30
31
32def _player(player_id: str, display_name: str) -> mock.Mock:
33 """Return a minimal player for setup-flow option generation."""
34 player = mock.Mock()
35 player.player_id = player_id
36 player.display_name = display_name
37 return player
38
39
40def _make_session(
41 domain: str,
42 *,
43 setup_data: dict[str, Any] | None = None,
44 values: dict[str, Any] | None = None,
45) -> tuple[SetupSession, dict[str, Any]]:
46 """Return a real setup session and the values collected by its finish handler."""
47 mass = mock.Mock()
48 mass.players.all_players.return_value = [
49 _player("living-room", "Living Room"),
50 _player("kitchen", "Kitchen"),
51 ]
52 collected: dict[str, Any] = {}
53
54 async def finish(_session: SetupSession, submitted: dict[str, Any]) -> dict[str, str]:
55 collected.update(submitted)
56 return {"instance_id": f"{domain}--test"}
57
58 context = SetupFlowContext(
59 kind="setup",
60 reason="user",
61 domain=domain,
62 setup_data=setup_data or {},
63 values=values or {},
64 )
65 return SetupSession(mass, "flow-test", context, finish), collected
66
67
68async def _start_form(session: SetupSession, flow_module: Any) -> tuple[asyncio.Task[None], Any]:
69 """Start a setup flow and wait for its form step."""
70 task = asyncio.create_task(flow_module.run_setup(session))
71 deadline = time.monotonic() + 5
72 while time.monotonic() < deadline:
73 if session.current_step and session.current_step.type == FlowStepType.FORM:
74 return task, session.current_step
75 await asyncio.sleep(0.01)
76 raise AssertionError("form step not published")
77
78
79async def _wait_finished(session: SetupSession) -> None:
80 """Wait for a setup session to finish."""
81 deadline = time.monotonic() + 5
82 while time.monotonic() < deadline:
83 if session.finished:
84 return
85 await asyncio.sleep(0.01)
86 raise AssertionError("setup flow did not finish")
87
88
89async def test_ariacast_flow_collects_mandatory_player() -> None:
90 """The AriaCast flow persists the mandatory target player (no automatic option)."""
91 session, collected = _make_session("ariacast_receiver")
92 task, step = await _start_form(session, ariacast_flow)
93 player_entry = next(entry for entry in step.entries if entry.key == ARIACAST_PLAYER_ID)
94 assert [option.value for option in player_entry.options] == ["kitchen", "living-room"]
95
96 session.handle_submit({ARIACAST_PLAYER_ID: "kitchen"})
97 await _wait_finished(session)
98 await task
99
100 assert collected == {ARIACAST_PLAYER_ID: "kitchen"}
101
102
103async def test_ariacast_flow_requires_a_player_selection() -> None:
104 """Submitting without a selection re-renders the form with a required error."""
105 session, collected = _make_session("ariacast_receiver")
106 task, _step = await _start_form(session, ariacast_flow)
107
108 step = session.handle_submit({})
109 assert step is not None
110 assert step.errors == {ARIACAST_PLAYER_ID: "required"}
111 assert collected == {}
112
113 task.cancel()
114 with pytest.raises(asyncio.CancelledError):
115 await task
116
117
118async def test_ariacast_flow_vanished_stored_player_renders_unselected() -> None:
119 """A stored player that no longer exists is not preselected (nor silently replaced)."""
120 session, _collected = _make_session(
121 "ariacast_receiver", setup_data={ARIACAST_PLAYER_ID: "gone-player"}
122 )
123 task, step = await _start_form(session, ariacast_flow)
124 player_entry = next(entry for entry in step.entries if entry.key == ARIACAST_PLAYER_ID)
125 assert player_entry.value is None
126
127 task.cancel()
128 with pytest.raises(asyncio.CancelledError):
129 await task
130
131
132async def test_ariacast_flow_aborts_without_players() -> None:
133 """With no players registered the flow aborts with the no_players reason."""
134 session, _collected = _make_session("ariacast_receiver")
135 session.mass.players.all_players.return_value = [] # type: ignore[attr-defined]
136
137 with pytest.raises(AbortFlow) as excinfo:
138 await ariacast_flow.run_setup(session)
139 assert excinfo.value.reason == "no_players"
140
141
142async def test_vban_flow_collects_receiver_endpoint_and_stream_format() -> None:
143 """The VBAN flow collects all values needed before starting its UDP receiver."""
144 session, collected = _make_session("vban_receiver")
145 submitted: dict[str, ConfigValueType] = {
146 CONF_BIND_PORT: 6981,
147 CONF_VBAN_STREAM_NAME: "Studio",
148 CONF_SENDER_HOST: "192.0.2.10",
149 CONF_PCM_AUDIO_FORMAT: "S16LE",
150 CONF_PCM_SAMPLE_RATE: 48000,
151 CONF_AUDIO_CHANNELS: 2,
152 CONF_BIND_IP: "0.0.0.0",
153 }
154 with mock.patch.object(
155 vban_flow, "get_ip_addresses", mock.AsyncMock(return_value=["192.0.2.2"])
156 ):
157 task, step = await _start_form(session, vban_flow)
158 assert {entry.key for entry in step.entries} >= set(submitted)
159 session.handle_submit(submitted)
160 await _wait_finished(session)
161 await task
162
163 assert collected == submitted
164