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