/
/
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 SetupFlowContext, SetupSession
15from music_assistant.providers.airplay_receiver import (
16 CONF_AIRPLAY_NAME,
17)
18from music_assistant.providers.airplay_receiver import (
19 CONF_MASS_PLAYER_ID as AIRPLAY_PLAYER_ID,
20)
21from music_assistant.providers.airplay_receiver import setup_flow as airplay_flow
22from music_assistant.providers.ariacast_receiver import (
23 CONF_ARIACAST_NAME,
24)
25from music_assistant.providers.ariacast_receiver import (
26 CONF_MASS_PLAYER_ID as ARIACAST_PLAYER_ID,
27)
28from music_assistant.providers.ariacast_receiver import setup_flow as ariacast_flow
29from music_assistant.providers.vban_receiver import setup_flow as vban_flow
30from music_assistant.providers.vban_receiver.constants import (
31 CONF_AUDIO_CHANNELS,
32 CONF_PCM_AUDIO_FORMAT,
33 CONF_PCM_SAMPLE_RATE,
34 CONF_SENDER_HOST,
35 CONF_VBAN_STREAM_NAME,
36)
37
38if TYPE_CHECKING:
39 from music_assistant_models.config_entries import ConfigValueType
40
41
42def _player(player_id: str, display_name: str) -> mock.Mock:
43 """Return a minimal player for setup-flow option generation."""
44 player = mock.Mock()
45 player.player_id = player_id
46 player.display_name = display_name
47 return player
48
49
50def _make_session(
51 domain: str,
52 *,
53 setup_data: dict[str, Any] | None = None,
54 values: dict[str, Any] | None = None,
55) -> tuple[SetupSession, dict[str, Any]]:
56 """Return a real setup session and the values collected by its finish handler."""
57 mass = mock.Mock()
58 mass.players.all_players.return_value = [
59 _player("living-room", "Living Room"),
60 _player("kitchen", "Kitchen"),
61 ]
62 collected: dict[str, Any] = {}
63
64 async def finish(_session: SetupSession, submitted: dict[str, Any]) -> dict[str, str]:
65 collected.update(submitted)
66 return {"instance_id": f"{domain}--test"}
67
68 context = SetupFlowContext(
69 kind="setup",
70 reason="user",
71 domain=domain,
72 setup_data=setup_data or {},
73 values=values or {},
74 )
75 return SetupSession(mass, "flow-test", context, finish), collected
76
77
78async def _start_form(session: SetupSession, flow_module: Any) -> tuple[asyncio.Task[None], Any]:
79 """Start a setup flow and wait for its form step."""
80 task = asyncio.create_task(flow_module.run_setup(session))
81 deadline = time.monotonic() + 5
82 while time.monotonic() < deadline:
83 if session.current_step and session.current_step.type == FlowStepType.FORM:
84 return task, session.current_step
85 await asyncio.sleep(0.01)
86 raise AssertionError("form step not published")
87
88
89async def _wait_finished(session: SetupSession) -> None:
90 """Wait for a setup session to finish."""
91 deadline = time.monotonic() + 5
92 while time.monotonic() < deadline:
93 if session.finished:
94 return
95 await asyncio.sleep(0.01)
96 raise AssertionError("setup flow did not finish")
97
98
99@pytest.mark.parametrize(
100 ("domain", "flow_module", "submitted"),
101 [
102 (
103 "airplay_receiver",
104 airplay_flow,
105 {
106 AIRPLAY_PLAYER_ID: "kitchen",
107 CONF_AIRPLAY_NAME: "Kitchen AirPlay",
108 },
109 ),
110 (
111 "ariacast_receiver",
112 ariacast_flow,
113 {
114 ARIACAST_PLAYER_ID: "kitchen",
115 CONF_ARIACAST_NAME: "Kitchen AriaCast",
116 },
117 ),
118 ],
119)
120async def test_player_receiver_flow_collects_instance_identity(
121 domain: str, flow_module: Any, submitted: dict[str, Any]
122) -> None:
123 """Player receiver flows persist their target player and advertised identity."""
124 session, collected = _make_session(domain)
125 task, step = await _start_form(session, flow_module)
126 player_entry = next(entry for entry in step.entries if entry.key.endswith("player_id"))
127 assert [option.value for option in player_entry.options] == [
128 "__auto__",
129 "kitchen",
130 "living-room",
131 ]
132
133 session.handle_submit(submitted)
134 await _wait_finished(session)
135 await task
136
137 assert collected == submitted
138
139
140async def test_player_receiver_flow_prefills_legacy_values() -> None:
141 """A pre-flow AirPlay instance keeps its stored player and receiver name."""
142 session, _collected = _make_session(
143 "airplay_receiver",
144 values={
145 AIRPLAY_PLAYER_ID: "living-room",
146 CONF_AIRPLAY_NAME: "Legacy Receiver",
147 },
148 )
149 task, step = await _start_form(session, airplay_flow)
150 entries = {entry.key: entry for entry in step.entries}
151
152 assert entries[AIRPLAY_PLAYER_ID].value == "living-room"
153 assert entries[CONF_AIRPLAY_NAME].value == "Legacy Receiver"
154
155 task.cancel()
156 with pytest.raises(asyncio.CancelledError):
157 await task
158
159
160async def test_vban_flow_collects_receiver_endpoint_and_stream_format() -> None:
161 """The VBAN flow collects all values needed before starting its UDP receiver."""
162 session, collected = _make_session("vban_receiver")
163 submitted: dict[str, ConfigValueType] = {
164 CONF_BIND_PORT: 6981,
165 CONF_VBAN_STREAM_NAME: "Studio",
166 CONF_SENDER_HOST: "192.0.2.10",
167 CONF_PCM_AUDIO_FORMAT: "S16LE",
168 CONF_PCM_SAMPLE_RATE: 48000,
169 CONF_AUDIO_CHANNELS: 2,
170 CONF_BIND_IP: "0.0.0.0",
171 }
172 with mock.patch.object(
173 vban_flow, "get_ip_addresses", mock.AsyncMock(return_value=["192.0.2.2"])
174 ):
175 task, step = await _start_form(session, vban_flow)
176 assert {entry.key for entry in step.entries} >= set(submitted)
177 session.handle_submit(submitted)
178 await _wait_finished(session)
179 await task
180
181 assert collected == submitted
182