/
/
/
1"""Tests for the AirPlay interactive setup (pairing) flow."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import logging
8import time
9from typing import TYPE_CHECKING, Any
10from unittest.mock import AsyncMock, MagicMock, patch
11
12import pytest
13from music_assistant_models.enums import ConfigEntryType, FlowStepType
14from music_assistant_models.errors import PlayerCommandFailed
15
16from music_assistant.models.setup_flow import AbortFlow, SetupFlowContext, SetupSession
17from music_assistant.providers.airplay.constants import (
18 AIRPLAY_DISCOVERY_TYPE,
19 COMPANION_DISCOVERY_TYPE,
20 CONF_AIRPLAY_CREDENTIALS,
21 CONF_COMPANION_CREDENTIALS,
22 CONF_COMPANION_PAIRING_PIN,
23 CONF_PAIR_NOW,
24 CONF_PAIRING_PASSWORD,
25 CONF_PAIRING_PIN,
26 CONF_PASSWORD,
27 CONF_RAOP_CREDENTIALS,
28)
29from music_assistant.providers.airplay.control_player import AirPlayControlPlayer
30from music_assistant.providers.airplay.player import AirPlayPlayer
31
32if TYPE_CHECKING:
33 from collections.abc import Callable
34
35 from music_assistant_models.setup_flow import SetupFlowStep
36
37# _airplay._tcp features bitmask with the AirPlay 2 feature bits set (bit 38/48).
38AP2_FEATURES = "0x4A7FDFD5,0x3C177FDE"
39# 192 hex chars, as produced by cliairplay --pair-setup
40FAKE_AP2_CREDS = "ab" * 96
41FAKE_RAOP_CREDS = "clientid:secret"
42_PAIRING_TARGET = "music_assistant.providers.airplay.pairing.AirPlayPairing"
43_PYATV_PAIR_TARGET = "music_assistant.providers.airplay.control_player.pyatv.pair"
44
45
46# --------------------------------------------------------------------------------------
47# Harness (direct-drive: run run_setup_flow as a task and pump the session by hand)
48# --------------------------------------------------------------------------------------
49
50
51def _make_session(
52 finish_handler: Any, *, player_id: str = "test_player"
53) -> tuple[SetupSession, MagicMock]:
54 """Build a real SetupSession backed by a Mock mass for driving run_setup_flow directly."""
55 mass = MagicMock()
56 context = SetupFlowContext(kind="setup", reason="user", domain="airplay", player_id=player_id)
57 return SetupSession(mass, "flow-test", context, finish_handler), mass
58
59
60def _published_steps(mass: MagicMock) -> list[SetupFlowStep]:
61 """Return the flow steps pushed through mass.signal_event, in order."""
62 return [call.kwargs["data"] for call in mass.signal_event.call_args_list]
63
64
65async def _wait_for(predicate: Callable[[], Any], timeout: float = 5.0) -> Any:
66 """Wait until the predicate returns a truthy value (or fail the test)."""
67 deadline = time.monotonic() + timeout
68 while time.monotonic() < deadline:
69 if result := predicate():
70 return result
71 await asyncio.sleep(0.01)
72 raise AssertionError("condition not met within timeout")
73
74
75async def _pump(
76 session: SetupSession,
77 task: asyncio.Task[None],
78 responder: Callable[[SetupFlowStep], dict[str, Any]],
79 *,
80 timeout: float = 5.0,
81) -> None:
82 """Answer each published FORM step via ``responder`` until the flow ends, then await it."""
83 handled: SetupFlowStep | None = None
84 while True:
85
86 def _ready(after: SetupFlowStep | None = handled) -> bool:
87 step = session.current_step
88 if task.done() or session.finished:
89 return True
90 return step is not None and step is not after and step.type == FlowStepType.FORM
91
92 await _wait_for(_ready, timeout)
93 if task.done() or session.finished:
94 break
95 handled = session.current_step
96 assert handled is not None
97 session.handle_submit(responder(handled))
98 await task
99
100
101def _pyatv_pairing(credentials: str) -> MagicMock:
102 """Return a mock pyatv PairingHandler that completes with the given credentials."""
103 pairing = MagicMock()
104 pairing.begin = AsyncMock()
105 pairing.finish = AsyncMock()
106 pairing.close = AsyncMock()
107 pairing.has_paired = True
108 pairing.service.credentials = credentials
109 return pairing
110
111
112def _service_info(
113 service_type: str,
114 properties: dict[str, str],
115 *,
116 address: str = "192.168.1.10",
117 port: int = 7000,
118) -> MagicMock:
119 """Create an mDNS service-info mock usable by the control-player pairing helpers."""
120 info = MagicMock()
121 info.type = service_type
122 info.name = f"test.{service_type}"
123 info.port = port
124 info.decoded_properties = dict(properties)
125 info.properties = {key.encode(): value.encode() for key, value in properties.items()}
126 info.addresses = [b"\xc0\xa8\x01\x0a"]
127 info.parsed_addresses.return_value = [address]
128 return info
129
130
131def _stub_setup_data(provider: MagicMock, player_id: str, setup_data: dict[str, Any]) -> None:
132 """Route the player's setup_data through the mocked mass.config get surface."""
133
134 def _config_get(key: str, default: Any = None) -> Any:
135 if key == f"players/{player_id}/setup_data":
136 return setup_data
137 if key == f"players/{player_id}":
138 return {"player_id": player_id}
139 return default
140
141 provider.mass.config.get.side_effect = _config_get
142 provider.mass.config.decrypt_string.side_effect = lambda value: value
143 provider.mass.config.encrypt_string.side_effect = lambda value: value
144 # Raw player config values (device password, password-invalid marker) come from
145 # their own store; without this they would read back as (truthy) mocks. The
146 # player's config reads from the same store so a stored password is observed.
147 raw_values: dict[str, Any] = {}
148 provider.mass.config.get_raw_player_config_value.side_effect = (
149 lambda _player_id, key, default=None: raw_values.get(key, default)
150 )
151 provider.mass.config.set_raw_player_config_value.side_effect = lambda _player_id, key, value: (
152 raw_values.__setitem__(key, value)
153 )
154 config = MagicMock()
155 config.get_value.side_effect = lambda key, default=None: raw_values.get(key, default)
156 provider.mass.config.get_base_player_config.return_value = config
157
158
159def _streaming_player(
160 *,
161 player_id: str = "test_player",
162 raop: bool = False,
163 setup_data: dict[str, Any] | None = None,
164 flags: str = "0x8",
165) -> AirPlayPlayer:
166 """Create a base AirPlay player; the default flags (0x8) require PIN pairing."""
167 provider = MagicMock()
168 provider.dacp_id = "0123456789ABCDEF"
169 _stub_setup_data(provider, player_id, setup_data or {})
170 if raop:
171 raop_info = _service_info("_raop._tcp.local.", {"sf": "0x200"}, port=5000)
172 airplay_info = None
173 else:
174 raop_info = None
175 airplay_info = _service_info(
176 AIRPLAY_DISCOVERY_TYPE, {"features": AP2_FEATURES, "flags": flags}
177 )
178 return AirPlayPlayer(
179 provider=provider,
180 player_id=player_id,
181 display_name="Test Player",
182 address="127.0.0.1",
183 manufacturer="Apple" if not raop else "Denon",
184 model="Apple TV" if not raop else "AVR",
185 raop_discovery_info=raop_info,
186 airplay_discovery_info=airplay_info,
187 )
188
189
190def _control_player(
191 *, player_id: str = "apctl", setup_data: dict[str, Any] | None = None
192) -> AirPlayControlPlayer:
193 """Create a control-capable Apple player requiring streaming PIN + Companion pairing."""
194 provider = MagicMock()
195 provider.instance_id = "airplay"
196 provider.dacp_id = "0123456789ABCDEF"
197 provider.logger = logging.getLogger("test.airplay.flow")
198 config = MagicMock()
199 config.get_value.side_effect = lambda _key, default=None: default
200 provider.mass.config.get_base_player_config.return_value = config
201 _stub_setup_data(provider, player_id, setup_data or {})
202 airplay_info = _service_info(
203 AIRPLAY_DISCOVERY_TYPE,
204 {
205 "deviceid": "AA:BB:CC:DD:EE:FF",
206 "features": AP2_FEATURES,
207 "model": "AppleTV11,1",
208 "osvers": "26.0",
209 "flags": "0x8", # require streaming PIN pairing
210 },
211 )
212 companion_info = _service_info(COMPANION_DISCOVERY_TYPE, {"rpFl": "0x367A2"}, port=49152)
213 return AirPlayControlPlayer(
214 provider=provider,
215 player_id=player_id,
216 raop_discovery_info=None,
217 airplay_discovery_info=airplay_info,
218 companion_discovery_info=companion_info,
219 mrp_discovery_info=None,
220 address="192.168.1.10",
221 display_name="Test Apple Device",
222 manufacturer="Apple",
223 model="Apple TV 4K",
224 initial_volume=25,
225 )
226
227
228# --------------------------------------------------------------------------------------
229# Streaming (base player) pairing
230# --------------------------------------------------------------------------------------
231
232
233async def test_streaming_pin_pairing_persists_airplay_credentials() -> None:
234 """The AirPlay 2 PIN happy path finishes with the credentials under the AirPlay key."""
235 collected: dict[str, Any] = {}
236
237 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
238 collected.update(values)
239 return {"player_id": "test_player"}
240
241 session, mass = _make_session(finish)
242 player = _streaming_player()
243 pairing = AsyncMock()
244 pairing.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
245
246 with patch(_PAIRING_TARGET, return_value=pairing):
247 task = asyncio.create_task(player.run_setup_flow(session))
248 await _pump(session, task, lambda _step: {CONF_PAIRING_PIN: "1234"})
249
250 assert collected == {CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS}
251 pairing.start_pairing_session.assert_awaited_once()
252 pairing.start_pin_pairing.assert_awaited_once()
253 pairing.finish_pairing.assert_awaited_once_with(pin="1234")
254 pairing.close.assert_awaited()
255 forms = [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
256 assert forms[0].step_id == "pair_pin"
257 # steps localize under the provider's own namespace
258 assert forms[0].translation_owner == "provider.airplay"
259 # the PIN renders as a 4-digit code input
260 pin_entry = next(entry for entry in forms[0].entries if entry.key == CONF_PAIRING_PIN)
261 assert pin_entry.type is ConfigEntryType.PAIRING_CODE
262 assert pin_entry.format == "####"
263
264
265async def test_streaming_pin_pairing_uses_raop_credentials_key() -> None:
266 """A legacy RAOP device stores its pairing secret under the RAOP-specific key."""
267 collected: dict[str, Any] = {}
268
269 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
270 collected.update(values)
271 return {"player_id": "test_player"}
272
273 session, _mass = _make_session(finish)
274 player = _streaming_player(raop=True)
275 pairing = AsyncMock()
276 pairing.finish_pairing = AsyncMock(return_value=FAKE_RAOP_CREDS)
277
278 with patch(_PAIRING_TARGET, return_value=pairing):
279 task = asyncio.create_task(player.run_setup_flow(session))
280 await _pump(session, task, lambda _step: {CONF_PAIRING_PIN: "4321"})
281
282 assert collected == {CONF_RAOP_CREDENTIALS: FAKE_RAOP_CREDS}
283
284
285async def test_streaming_pin_pairing_retries_on_wrong_pin() -> None:
286 """A rejected PIN re-renders the form (with an error) and a fresh pairing is started."""
287 collected: dict[str, Any] = {}
288
289 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
290 collected.update(values)
291 return {"player_id": "test_player"}
292
293 session, mass = _make_session(finish)
294 player = _streaming_player()
295 rejected = AsyncMock()
296 rejected.finish_pairing = AsyncMock(side_effect=PlayerCommandFailed("wrong pin"))
297 accepted = AsyncMock()
298 accepted.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
299 attempts = [rejected, accepted]
300 pins = iter(["0000", "1234"])
301
302 # snapshot each step's errors at publish time: a successful submit later clears
303 # them on the (same) step object, so the live object cannot be inspected afterwards
304 published: list[tuple[str, dict[str, str]]] = []
305 mass.signal_event.side_effect = lambda *_a, **kwargs: published.append(
306 (kwargs["data"].step_id, dict(kwargs["data"].errors))
307 )
308
309 with patch(_PAIRING_TARGET, side_effect=lambda **_kwargs: attempts.pop(0)):
310 task = asyncio.create_task(player.run_setup_flow(session))
311 await _pump(session, task, lambda _step: {CONF_PAIRING_PIN: next(pins)})
312
313 assert collected == {CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS}
314 # each attempt uses (and tears down) its own live pairing session
315 rejected.close.assert_awaited()
316 accepted.close.assert_awaited()
317 pin_form_errors = [errors for step_id, errors in published if step_id == "pair_pin"]
318 assert len(pin_form_errors) == 2
319 # first attempt shows a clean form, the retry surfaces the failure
320 assert pin_form_errors[0] == {}
321 assert pin_form_errors[1].get("base")
322
323
324async def test_streaming_repair_offer_declined_keeps_stored_pairing() -> None:
325 """Re-running the flow when already paired offers re-pairing; declining changes nothing."""
326 finished_values: dict[str, Any] = {}
327
328 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
329 finished_values.update({"values": values})
330 return {"player_id": "test_player"}
331
332 session, mass = _make_session(finish)
333 player = _streaming_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS})
334
335 with patch(_PAIRING_TARGET) as pairing_cls:
336 task = asyncio.create_task(player.run_setup_flow(session))
337 await _pump(session, task, lambda _step: {CONF_PAIR_NOW: False})
338
339 pairing_cls.assert_not_called()
340 assert finished_values["values"] == {}
341 forms = [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
342 assert [step.step_id for step in forms] == ["streaming_repair_offer"]
343
344
345async def test_streaming_repair_offer_accepted_replaces_credentials() -> None:
346 """Accepting the re-pair offer runs a fresh pairing and stores the new credentials."""
347 new_creds = "cd" * 96
348 collected: dict[str, Any] = {}
349
350 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
351 collected.update(values)
352 return {"player_id": "test_player"}
353
354 session, _mass = _make_session(finish)
355 player = _streaming_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS})
356 pairing = AsyncMock()
357 pairing.finish_pairing = AsyncMock(return_value=new_creds)
358 responses: dict[str, dict[str, Any]] = {
359 "streaming_repair_offer": {CONF_PAIR_NOW: True},
360 "pair_pin": {CONF_PAIRING_PIN: "1234"},
361 }
362
363 with patch(_PAIRING_TARGET, return_value=pairing):
364 task = asyncio.create_task(player.run_setup_flow(session))
365 await _pump(session, task, lambda step: responses[step.step_id])
366
367 assert collected == {CONF_AIRPLAY_CREDENTIALS: new_creds}
368 pairing.finish_pairing.assert_awaited_once_with(pin="1234")
369
370
371async def test_stale_credentials_cleared_when_pairing_not_required() -> None:
372 """
373 A device that (no longer) requires pairing gets its leftover credentials cleared.
374
375 Covers the HomePod trap: credentials stored while a password was set keep forcing
376 the pair-verify route after the password is removed again, which the device may
377 accept without actually outputting audio. Re-running the flow must reset this.
378 """
379 collected: dict[str, Any] = {}
380
381 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
382 collected.update(values)
383 return {"player_id": "test_player"}
384
385 session, mass = _make_session(finish)
386 # flags without the PIN/legacy/password bits: no pairing required
387 player = _streaming_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS}, flags="0x4")
388
389 with patch(_PAIRING_TARGET) as pairing_cls:
390 task = asyncio.create_task(player.run_setup_flow(session))
391 await _wait_for(lambda: session.finished)
392 await task
393
394 pairing_cls.assert_not_called()
395 assert collected == {CONF_AIRPLAY_CREDENTIALS: None}
396 # nothing to ask: the flow finishes without publishing any form
397 assert not [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
398
399
400async def test_abort_mid_pairing_closes_session() -> None:
401 """Cancelling the flow while awaiting the PIN tears the live pairing session down."""
402
403 async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
404 return {"player_id": "test_player"}
405
406 session, _mass = _make_session(finish)
407 player = _streaming_player()
408 pairing = AsyncMock()
409
410 with patch(_PAIRING_TARGET, return_value=pairing):
411 task = asyncio.create_task(player.run_setup_flow(session))
412 await _wait_for(
413 lambda: session.current_step is not None and session.current_step.step_id == "pair_pin"
414 )
415 task.cancel()
416 with contextlib.suppress(asyncio.CancelledError):
417 await task
418
419 pairing.close.assert_awaited()
420
421
422async def test_pairing_start_failure_aborts_with_reason() -> None:
423 """
424 A failure starting the pairing session aborts cleanly, not as a raw crash.
425
426 The engine turns an uncaught exception into a generic ``internal_error``; the flow
427 instead raises ``AbortFlow("pairing_failed")`` so the user sees an actionable reason,
428 and the half-started session is still torn down.
429 """
430
431 async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
432 return {"player_id": "test_player"}
433
434 session, _mass = _make_session(finish)
435 player = _streaming_player()
436 pairing = AsyncMock()
437 # a device/binary/system failure surfaces from starting the session, not from finish
438 pairing.start_pairing_session = AsyncMock(
439 side_effect=RuntimeError("Unable to locate cliairplay binary")
440 )
441
442 with patch(_PAIRING_TARGET, return_value=pairing), pytest.raises(AbortFlow) as exc:
443 await player.run_setup_flow(session)
444
445 assert exc.value.reason == "pairing_failed"
446 # the half-started session is still torn down
447 pairing.close.assert_awaited()
448
449
450# --------------------------------------------------------------------------------------
451# Control player: streaming + optional Companion/MRP ("two codes")
452# --------------------------------------------------------------------------------------
453
454
455async def test_two_code_sequence_streaming_then_companion() -> None:
456 """A controlled device pairs the streaming PIN, then the optional Companion PIN."""
457 collected: dict[str, Any] = {}
458
459 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
460 collected.update(values)
461 return {"player_id": "apctl"}
462
463 session, mass = _make_session(finish, player_id="apctl")
464 player = _control_player()
465 streaming = AsyncMock()
466 streaming.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
467 companion = _pyatv_pairing("companion-creds")
468 responses: dict[str, dict[str, Any]] = {
469 "pair_pin": {CONF_PAIRING_PIN: "1234"},
470 "companion_offer": {CONF_PAIR_NOW: True},
471 "pair_companion": {CONF_COMPANION_PAIRING_PIN: "5678"},
472 # MRP is offered for an Apple TV; decline it to keep this a two-code run
473 "mrp_offer": {CONF_PAIR_NOW: False},
474 }
475
476 with (
477 patch(_PAIRING_TARGET, return_value=streaming),
478 patch(_PYATV_PAIR_TARGET, return_value=companion) as pyatv_pair,
479 ):
480 task = asyncio.create_task(player.run_setup_flow(session))
481 await _pump(session, task, lambda step: responses[step.step_id])
482
483 assert collected[CONF_AIRPLAY_CREDENTIALS] == FAKE_AP2_CREDS
484 assert collected[CONF_COMPANION_CREDENTIALS] == "companion-creds"
485 companion.pin.assert_called_once_with(5678)
486 pyatv_pair.assert_called_once()
487 step_ids = [step.step_id for step in _published_steps(mass) if step.type == FlowStepType.FORM]
488 # streaming PIN comes before the optional control offers
489 assert step_ids.index("pair_pin") < step_ids.index("companion_offer")
490 assert "pair_companion" in step_ids
491 # both PINs render as 4-digit code inputs
492 pin_entries = [
493 entry
494 for step in _published_steps(mass)
495 if step.type == FlowStepType.FORM
496 for entry in step.entries
497 if entry.key in (CONF_PAIRING_PIN, CONF_COMPANION_PAIRING_PIN)
498 ]
499 assert len(pin_entries) == 2
500 assert all(entry.type is ConfigEntryType.PAIRING_CODE for entry in pin_entries)
501 assert all(entry.format == "####" for entry in pin_entries)
502
503
504async def test_all_pairings_reoffered_and_skippable_when_already_paired() -> None:
505 """
506 A fully paired device re-offers every pairing on a re-run; declining all is a no-op.
507
508 Previously stored credentials silently skipped their steps, leaving no way to
509 redo a stale pairing from the player settings.
510 """
511 collected: dict[str, Any] = {}
512
513 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
514 collected.update(values)
515 return {"player_id": "apctl"}
516
517 session, mass = _make_session(finish, player_id="apctl")
518 player = _control_player(
519 setup_data={
520 CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS,
521 CONF_COMPANION_CREDENTIALS: "companion-creds",
522 }
523 )
524 responses: dict[str, dict[str, Any]] = {
525 "streaming_repair_offer": {CONF_PAIR_NOW: False},
526 "companion_offer": {CONF_PAIR_NOW: False},
527 "mrp_offer": {CONF_PAIR_NOW: False},
528 }
529
530 with (
531 patch(_PAIRING_TARGET) as pairing_cls,
532 patch(_PYATV_PAIR_TARGET) as pyatv_pair,
533 ):
534 task = asyncio.create_task(player.run_setup_flow(session))
535 await _pump(session, task, lambda step: responses[step.step_id])
536
537 assert collected == {}
538 pairing_cls.assert_not_called()
539 pyatv_pair.assert_not_called()
540 step_ids = [step.step_id for step in _published_steps(mass) if step.type == FlowStepType.FORM]
541 assert step_ids == ["streaming_repair_offer", "companion_offer", "mrp_offer"]
542
543
544def _password_player(
545 *, player_id: str = "test_player", setup_data: dict[str, Any] | None = None
546) -> AirPlayPlayer:
547 """Create an AirPlay 2 player that announces password protection (flags bit 0x80)."""
548 provider = MagicMock()
549 provider.dacp_id = "0123456789ABCDEF"
550 _stub_setup_data(provider, player_id, setup_data or {})
551 return AirPlayPlayer(
552 provider=provider,
553 player_id=player_id,
554 display_name="Test HomePod",
555 address="127.0.0.1",
556 manufacturer="Apple",
557 model="HomePod mini",
558 raop_discovery_info=None,
559 airplay_discovery_info=_service_info(
560 AIRPLAY_DISCOVERY_TYPE, {"features": AP2_FEATURES, "flags": "0x80"}
561 ),
562 )
563
564
565def _raop_password_player(*, player_id: str = "test_player") -> AirPlayPlayer:
566 """Create a legacy RAOP receiver that publishes the classic ``pw=true`` boolean."""
567 provider = MagicMock()
568 provider.dacp_id = "0123456789ABCDEF"
569 _stub_setup_data(provider, player_id, {})
570 return AirPlayPlayer(
571 provider=provider,
572 player_id=player_id,
573 display_name="Test Speaker",
574 address="127.0.0.1",
575 manufacturer="Denon",
576 model="AVR",
577 raop_discovery_info=_service_info("_raop._tcp.local.", {"pw": "true"}, port=5000),
578 airplay_discovery_info=None,
579 )
580
581
582async def test_streaming_password_pairing_persists_the_device_password() -> None:
583 """The entered password is stored as player config, not discarded with the form."""
584 collected: dict[str, Any] = {}
585
586 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
587 collected.update(values)
588 return {"player_id": "test_player"}
589
590 session, _mass = _make_session(finish)
591 player = _password_player()
592 pairing = AsyncMock()
593 pairing.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
594
595 with patch(_PAIRING_TARGET, return_value=pairing):
596 task = asyncio.create_task(player.run_setup_flow(session))
597 await _pump(session, task, lambda _step: {CONF_PAIRING_PASSWORD: "hunter2"})
598
599 # the pairing credentials still go to setup_data...
600 assert collected == {CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS}
601 # ...and the password itself is persisted so every stream can present it
602 player.mass.config.set_raw_player_config_value.assert_called_once_with( # type: ignore[attr-defined]
603 "test_player", CONF_PASSWORD, "hunter2"
604 )
605
606
607async def test_streaming_password_pairing_uses_a_password_form() -> None:
608 """A password-protected device is asked for a password instead of a PIN."""
609
610 async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
611 return {"player_id": "test_player"}
612
613 session, mass = _make_session(finish)
614 player = _password_player()
615 pairing = AsyncMock()
616 pairing.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
617
618 with patch(_PAIRING_TARGET, return_value=pairing):
619 task = asyncio.create_task(player.run_setup_flow(session))
620 await _pump(session, task, lambda _step: {CONF_PAIRING_PASSWORD: "hunter2"})
621
622 forms = [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
623 assert [step.step_id for step in forms] == ["pair_password"]
624 # the device shows no PIN in this flow, so no PIN pairing is started
625 pairing.start_pin_pairing.assert_not_awaited()
626
627
628async def test_raop_password_device_is_asked_for_its_password_without_pairing() -> None:
629 """A legacy RAOP receiver has no pairing to do, but still needs its password."""
630 collected: dict[str, Any] = {}
631
632 async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
633 collected.update(values)
634 return {"player_id": "test_player"}
635
636 session, mass = _make_session(finish)
637 player = _raop_password_player()
638 assert player.needs_setup is True
639 assert player.setup_reason == "password_required"
640
641 with patch(_PAIRING_TARGET) as pairing_cls:
642 task = asyncio.create_task(player.run_setup_flow(session))
643 await _pump(session, task, lambda _step: {CONF_PAIRING_PASSWORD: "hunter2"})
644
645 forms = [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
646 assert [step.step_id for step in forms] == ["pair_password"]
647 # no pairing session is ever built for a device that has nothing to pair
648 pairing_cls.assert_not_called()
649 assert collected == {}
650 player.mass.config.set_raw_player_config_value.assert_any_call( # type: ignore[attr-defined]
651 "test_player", CONF_PASSWORD, "hunter2"
652 )
653 assert player.needs_setup is False
654
655
656async def test_paired_device_with_a_rejected_password_is_asked_for_it_again() -> None:
657 """
658 Stored credentials must not skip the password step.
659
660 This is the device that gained password protection after it was set up: it is
661 already paired, so there is nothing to pair, yet it cannot stream until the
662 (new) password is entered.
663 """
664
665 async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
666 return {"player_id": "test_player"}
667
668 session, mass = _make_session(finish)
669 player = _password_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS})
670 player.set_password_invalid(True)
671 assert player.needs_setup is True
672
673 responses: dict[str, dict[str, Any]] = {
674 "streaming_repair_offer": {CONF_PAIR_NOW: False},
675 "pair_password": {CONF_PAIRING_PASSWORD: "hunter2"},
676 }
677 with patch(_PAIRING_TARGET) as pairing_cls:
678 task = asyncio.create_task(player.run_setup_flow(session))
679 await _pump(session, task, lambda step: responses[step.step_id])
680
681 forms = [step.step_id for step in _published_steps(mass) if step.type == FlowStepType.FORM]
682 # skipping the optional re-pair still leads to the password step
683 assert forms == ["streaming_repair_offer", "pair_password"]
684 pairing_cls.assert_not_called()
685 # storing a fresh password clears the reject marker, so the player is ready again
686 assert player.password_invalid is False
687 assert player.needs_setup is False
688
689
690async def test_ready_player_is_not_asked_for_a_password() -> None:
691 """A paired device with a working password only gets the optional re-pair offer."""
692
693 async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
694 return {"player_id": "test_player"}
695
696 session, mass = _make_session(finish)
697 player = _password_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS})
698 player._store_device_password("hunter2")
699
700 responses: dict[str, dict[str, Any]] = {
701 "streaming_repair_offer": {CONF_PAIR_NOW: False},
702 }
703 with patch(_PAIRING_TARGET) as pairing_cls:
704 task = asyncio.create_task(player.run_setup_flow(session))
705 await _pump(session, task, lambda step: responses[step.step_id])
706
707 forms = [step.step_id for step in _published_steps(mass) if step.type == FlowStepType.FORM]
708 # no password form for a ready player - only the skippable re-pair offer
709 assert forms == ["streaming_repair_offer"]
710 pairing_cls.assert_not_called()
711