music-assistant-server

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 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
260
261async def test_streaming_pin_pairing_uses_raop_credentials_key() -> None:
262    """A legacy RAOP device stores its pairing secret under the RAOP-specific key."""
263    collected: dict[str, Any] = {}
264
265    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
266        collected.update(values)
267        return {"player_id": "test_player"}
268
269    session, _mass = _make_session(finish)
270    player = _streaming_player(raop=True)
271    pairing = AsyncMock()
272    pairing.finish_pairing = AsyncMock(return_value=FAKE_RAOP_CREDS)
273
274    with patch(_PAIRING_TARGET, return_value=pairing):
275        task = asyncio.create_task(player.run_setup_flow(session))
276        await _pump(session, task, lambda _step: {CONF_PAIRING_PIN: "4321"})
277
278    assert collected == {CONF_RAOP_CREDENTIALS: FAKE_RAOP_CREDS}
279
280
281async def test_streaming_pin_pairing_retries_on_wrong_pin() -> None:
282    """A rejected PIN re-renders the form (with an error) and a fresh pairing is started."""
283    collected: dict[str, Any] = {}
284
285    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
286        collected.update(values)
287        return {"player_id": "test_player"}
288
289    session, mass = _make_session(finish)
290    player = _streaming_player()
291    rejected = AsyncMock()
292    rejected.finish_pairing = AsyncMock(side_effect=PlayerCommandFailed("wrong pin"))
293    accepted = AsyncMock()
294    accepted.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
295    attempts = [rejected, accepted]
296    pins = iter(["0000", "1234"])
297
298    # snapshot each step's errors at publish time: a successful submit later clears
299    # them on the (same) step object, so the live object cannot be inspected afterwards
300    published: list[tuple[str, dict[str, str]]] = []
301    mass.signal_event.side_effect = lambda *_a, **kwargs: published.append(
302        (kwargs["data"].step_id, dict(kwargs["data"].errors))
303    )
304
305    with patch(_PAIRING_TARGET, side_effect=lambda **_kwargs: attempts.pop(0)):
306        task = asyncio.create_task(player.run_setup_flow(session))
307        await _pump(session, task, lambda _step: {CONF_PAIRING_PIN: next(pins)})
308
309    assert collected == {CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS}
310    # each attempt uses (and tears down) its own live pairing session
311    rejected.close.assert_awaited()
312    accepted.close.assert_awaited()
313    pin_form_errors = [errors for step_id, errors in published if step_id == "pair_pin"]
314    assert len(pin_form_errors) == 2
315    # first attempt shows a clean form, the retry surfaces the failure
316    assert pin_form_errors[0] == {}
317    assert pin_form_errors[1].get("base")
318
319
320async def test_streaming_repair_offer_declined_keeps_stored_pairing() -> None:
321    """Re-running the flow when already paired offers re-pairing; declining changes nothing."""
322    finished_values: dict[str, Any] = {}
323
324    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
325        finished_values.update({"values": values})
326        return {"player_id": "test_player"}
327
328    session, mass = _make_session(finish)
329    player = _streaming_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS})
330
331    with patch(_PAIRING_TARGET) as pairing_cls:
332        task = asyncio.create_task(player.run_setup_flow(session))
333        await _pump(session, task, lambda _step: {CONF_PAIR_NOW: False})
334
335    pairing_cls.assert_not_called()
336    assert finished_values["values"] == {}
337    forms = [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
338    assert [step.step_id for step in forms] == ["streaming_repair_offer"]
339
340
341async def test_streaming_repair_offer_accepted_replaces_credentials() -> None:
342    """Accepting the re-pair offer runs a fresh pairing and stores the new credentials."""
343    new_creds = "cd" * 96
344    collected: dict[str, Any] = {}
345
346    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
347        collected.update(values)
348        return {"player_id": "test_player"}
349
350    session, _mass = _make_session(finish)
351    player = _streaming_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS})
352    pairing = AsyncMock()
353    pairing.finish_pairing = AsyncMock(return_value=new_creds)
354    responses: dict[str, dict[str, Any]] = {
355        "streaming_repair_offer": {CONF_PAIR_NOW: True},
356        "pair_pin": {CONF_PAIRING_PIN: "1234"},
357    }
358
359    with patch(_PAIRING_TARGET, return_value=pairing):
360        task = asyncio.create_task(player.run_setup_flow(session))
361        await _pump(session, task, lambda step: responses[step.step_id])
362
363    assert collected == {CONF_AIRPLAY_CREDENTIALS: new_creds}
364    pairing.finish_pairing.assert_awaited_once_with(pin="1234")
365
366
367async def test_stale_credentials_cleared_when_pairing_not_required() -> None:
368    """
369    A device that (no longer) requires pairing gets its leftover credentials cleared.
370
371    Covers the HomePod trap: credentials stored while a password was set keep forcing
372    the pair-verify route after the password is removed again, which the device may
373    accept without actually outputting audio. Re-running the flow must reset this.
374    """
375    collected: dict[str, Any] = {}
376
377    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
378        collected.update(values)
379        return {"player_id": "test_player"}
380
381    session, mass = _make_session(finish)
382    # flags without the PIN/legacy/password bits: no pairing required
383    player = _streaming_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS}, flags="0x4")
384
385    with patch(_PAIRING_TARGET) as pairing_cls:
386        task = asyncio.create_task(player.run_setup_flow(session))
387        await _wait_for(lambda: session.finished)
388        await task
389
390    pairing_cls.assert_not_called()
391    assert collected == {CONF_AIRPLAY_CREDENTIALS: None}
392    # nothing to ask: the flow finishes without publishing any form
393    assert not [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
394
395
396async def test_abort_mid_pairing_closes_session() -> None:
397    """Cancelling the flow while awaiting the PIN tears the live pairing session down."""
398
399    async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
400        return {"player_id": "test_player"}
401
402    session, _mass = _make_session(finish)
403    player = _streaming_player()
404    pairing = AsyncMock()
405
406    with patch(_PAIRING_TARGET, return_value=pairing):
407        task = asyncio.create_task(player.run_setup_flow(session))
408        await _wait_for(
409            lambda: session.current_step is not None and session.current_step.step_id == "pair_pin"
410        )
411        task.cancel()
412        with contextlib.suppress(asyncio.CancelledError):
413            await task
414
415    pairing.close.assert_awaited()
416
417
418async def test_pairing_start_failure_aborts_with_reason() -> None:
419    """
420    A failure starting the pairing session aborts cleanly, not as a raw crash.
421
422    The engine turns an uncaught exception into a generic ``internal_error``; the flow
423    instead raises ``AbortFlow("pairing_failed")`` so the user sees an actionable reason,
424    and the half-started session is still torn down.
425    """
426
427    async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
428        return {"player_id": "test_player"}
429
430    session, _mass = _make_session(finish)
431    player = _streaming_player()
432    pairing = AsyncMock()
433    # a device/binary/system failure surfaces from starting the session, not from finish
434    pairing.start_pairing_session = AsyncMock(
435        side_effect=RuntimeError("Unable to locate cliairplay binary")
436    )
437
438    with patch(_PAIRING_TARGET, return_value=pairing), pytest.raises(AbortFlow) as exc:
439        await player.run_setup_flow(session)
440
441    assert exc.value.reason == "pairing_failed"
442    # the half-started session is still torn down
443    pairing.close.assert_awaited()
444
445
446# --------------------------------------------------------------------------------------
447# Control player: streaming + optional Companion/MRP ("two codes")
448# --------------------------------------------------------------------------------------
449
450
451async def test_two_code_sequence_streaming_then_companion() -> None:
452    """A controlled device pairs the streaming PIN, then the optional Companion PIN."""
453    collected: dict[str, Any] = {}
454
455    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
456        collected.update(values)
457        return {"player_id": "apctl"}
458
459    session, mass = _make_session(finish, player_id="apctl")
460    player = _control_player()
461    streaming = AsyncMock()
462    streaming.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
463    companion = _pyatv_pairing("companion-creds")
464    responses: dict[str, dict[str, Any]] = {
465        "pair_pin": {CONF_PAIRING_PIN: "1234"},
466        "companion_offer": {CONF_PAIR_NOW: True},
467        "pair_companion": {CONF_COMPANION_PAIRING_PIN: "5678"},
468        # MRP is offered for an Apple TV; decline it to keep this a two-code run
469        "mrp_offer": {CONF_PAIR_NOW: False},
470    }
471
472    with (
473        patch(_PAIRING_TARGET, return_value=streaming),
474        patch(_PYATV_PAIR_TARGET, return_value=companion) as pyatv_pair,
475    ):
476        task = asyncio.create_task(player.run_setup_flow(session))
477        await _pump(session, task, lambda step: responses[step.step_id])
478
479    assert collected[CONF_AIRPLAY_CREDENTIALS] == FAKE_AP2_CREDS
480    assert collected[CONF_COMPANION_CREDENTIALS] == "companion-creds"
481    companion.pin.assert_called_once_with(5678)
482    pyatv_pair.assert_called_once()
483    step_ids = [step.step_id for step in _published_steps(mass) if step.type == FlowStepType.FORM]
484    # streaming PIN comes before the optional control offers
485    assert step_ids.index("pair_pin") < step_ids.index("companion_offer")
486    assert "pair_companion" in step_ids
487
488
489async def test_all_pairings_reoffered_and_skippable_when_already_paired() -> None:
490    """
491    A fully paired device re-offers every pairing on a re-run; declining all is a no-op.
492
493    Previously stored credentials silently skipped their steps, leaving no way to
494    redo a stale pairing from the player settings.
495    """
496    collected: dict[str, Any] = {}
497
498    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
499        collected.update(values)
500        return {"player_id": "apctl"}
501
502    session, mass = _make_session(finish, player_id="apctl")
503    player = _control_player(
504        setup_data={
505            CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS,
506            CONF_COMPANION_CREDENTIALS: "companion-creds",
507        }
508    )
509    responses: dict[str, dict[str, Any]] = {
510        "streaming_repair_offer": {CONF_PAIR_NOW: False},
511        "companion_offer": {CONF_PAIR_NOW: False},
512        "mrp_offer": {CONF_PAIR_NOW: False},
513    }
514
515    with (
516        patch(_PAIRING_TARGET) as pairing_cls,
517        patch(_PYATV_PAIR_TARGET) as pyatv_pair,
518    ):
519        task = asyncio.create_task(player.run_setup_flow(session))
520        await _pump(session, task, lambda step: responses[step.step_id])
521
522    assert collected == {}
523    pairing_cls.assert_not_called()
524    pyatv_pair.assert_not_called()
525    step_ids = [step.step_id for step in _published_steps(mass) if step.type == FlowStepType.FORM]
526    assert step_ids == ["streaming_repair_offer", "companion_offer", "mrp_offer"]
527
528
529def _password_player(
530    *, player_id: str = "test_player", setup_data: dict[str, Any] | None = None
531) -> AirPlayPlayer:
532    """Create an AirPlay 2 player that announces password protection (flags bit 0x80)."""
533    provider = MagicMock()
534    provider.dacp_id = "0123456789ABCDEF"
535    _stub_setup_data(provider, player_id, setup_data or {})
536    return AirPlayPlayer(
537        provider=provider,
538        player_id=player_id,
539        display_name="Test HomePod",
540        address="127.0.0.1",
541        manufacturer="Apple",
542        model="HomePod mini",
543        raop_discovery_info=None,
544        airplay_discovery_info=_service_info(
545            AIRPLAY_DISCOVERY_TYPE, {"features": AP2_FEATURES, "flags": "0x80"}
546        ),
547    )
548
549
550def _raop_password_player(*, player_id: str = "test_player") -> AirPlayPlayer:
551    """Create a legacy RAOP receiver that publishes the classic ``pw=true`` boolean."""
552    provider = MagicMock()
553    provider.dacp_id = "0123456789ABCDEF"
554    _stub_setup_data(provider, player_id, {})
555    return AirPlayPlayer(
556        provider=provider,
557        player_id=player_id,
558        display_name="Test Speaker",
559        address="127.0.0.1",
560        manufacturer="Denon",
561        model="AVR",
562        raop_discovery_info=_service_info("_raop._tcp.local.", {"pw": "true"}, port=5000),
563        airplay_discovery_info=None,
564    )
565
566
567async def test_streaming_password_pairing_persists_the_device_password() -> None:
568    """The entered password is stored as player config, not discarded with the form."""
569    collected: dict[str, Any] = {}
570
571    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
572        collected.update(values)
573        return {"player_id": "test_player"}
574
575    session, _mass = _make_session(finish)
576    player = _password_player()
577    pairing = AsyncMock()
578    pairing.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
579
580    with patch(_PAIRING_TARGET, return_value=pairing):
581        task = asyncio.create_task(player.run_setup_flow(session))
582        await _pump(session, task, lambda _step: {CONF_PAIRING_PASSWORD: "hunter2"})
583
584    # the pairing credentials still go to setup_data...
585    assert collected == {CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS}
586    # ...and the password itself is persisted so every stream can present it
587    player.mass.config.set_raw_player_config_value.assert_called_once_with(  # type: ignore[attr-defined]
588        "test_player", CONF_PASSWORD, "hunter2"
589    )
590
591
592async def test_streaming_password_pairing_uses_a_password_form() -> None:
593    """A password-protected device is asked for a password instead of a PIN."""
594
595    async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
596        return {"player_id": "test_player"}
597
598    session, mass = _make_session(finish)
599    player = _password_player()
600    pairing = AsyncMock()
601    pairing.finish_pairing = AsyncMock(return_value=FAKE_AP2_CREDS)
602
603    with patch(_PAIRING_TARGET, return_value=pairing):
604        task = asyncio.create_task(player.run_setup_flow(session))
605        await _pump(session, task, lambda _step: {CONF_PAIRING_PASSWORD: "hunter2"})
606
607    forms = [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
608    assert [step.step_id for step in forms] == ["pair_password"]
609    # the device shows no PIN in this flow, so no PIN pairing is started
610    pairing.start_pin_pairing.assert_not_awaited()
611
612
613async def test_raop_password_device_is_asked_for_its_password_without_pairing() -> None:
614    """A legacy RAOP receiver has no pairing to do, but still needs its password."""
615    collected: dict[str, Any] = {}
616
617    async def finish(_session: SetupSession, values: dict[str, Any]) -> dict[str, str]:
618        collected.update(values)
619        return {"player_id": "test_player"}
620
621    session, mass = _make_session(finish)
622    player = _raop_password_player()
623    assert player.needs_setup is True
624    assert player.setup_reason == "password_required"
625
626    with patch(_PAIRING_TARGET) as pairing_cls:
627        task = asyncio.create_task(player.run_setup_flow(session))
628        await _pump(session, task, lambda _step: {CONF_PAIRING_PASSWORD: "hunter2"})
629
630    forms = [step for step in _published_steps(mass) if step.type == FlowStepType.FORM]
631    assert [step.step_id for step in forms] == ["pair_password"]
632    # no pairing session is ever built for a device that has nothing to pair
633    pairing_cls.assert_not_called()
634    assert collected == {}
635    player.mass.config.set_raw_player_config_value.assert_any_call(  # type: ignore[attr-defined]
636        "test_player", CONF_PASSWORD, "hunter2"
637    )
638    assert player.needs_setup is False
639
640
641async def test_paired_device_with_a_rejected_password_is_asked_for_it_again() -> None:
642    """
643    Stored credentials must not skip the password step.
644
645    This is the device that gained password protection after it was set up: it is
646    already paired, so there is nothing to pair, yet it cannot stream until the
647    (new) password is entered.
648    """
649
650    async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
651        return {"player_id": "test_player"}
652
653    session, mass = _make_session(finish)
654    player = _password_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS})
655    player.set_password_invalid(True)
656    assert player.needs_setup is True
657
658    responses: dict[str, dict[str, Any]] = {
659        "streaming_repair_offer": {CONF_PAIR_NOW: False},
660        "pair_password": {CONF_PAIRING_PASSWORD: "hunter2"},
661    }
662    with patch(_PAIRING_TARGET) as pairing_cls:
663        task = asyncio.create_task(player.run_setup_flow(session))
664        await _pump(session, task, lambda step: responses[step.step_id])
665
666    forms = [step.step_id for step in _published_steps(mass) if step.type == FlowStepType.FORM]
667    # skipping the optional re-pair still leads to the password step
668    assert forms == ["streaming_repair_offer", "pair_password"]
669    pairing_cls.assert_not_called()
670    # storing a fresh password clears the reject marker, so the player is ready again
671    assert player.password_invalid is False
672    assert player.needs_setup is False
673
674
675async def test_ready_player_is_not_asked_for_a_password() -> None:
676    """A paired device with a working password only gets the optional re-pair offer."""
677
678    async def finish(_session: SetupSession, _values: dict[str, Any]) -> dict[str, str]:
679        return {"player_id": "test_player"}
680
681    session, mass = _make_session(finish)
682    player = _password_player(setup_data={CONF_AIRPLAY_CREDENTIALS: FAKE_AP2_CREDS})
683    player._store_device_password("hunter2")
684
685    responses: dict[str, dict[str, Any]] = {
686        "streaming_repair_offer": {CONF_PAIR_NOW: False},
687    }
688    with patch(_PAIRING_TARGET) as pairing_cls:
689        task = asyncio.create_task(player.run_setup_flow(session))
690        await _pump(session, task, lambda step: responses[step.step_id])
691
692    forms = [step.step_id for step in _published_steps(mass) if step.type == FlowStepType.FORM]
693    # no password form for a ready player - only the skippable re-pair offer
694    assert forms == ["streaming_repair_offer"]
695    pairing_cls.assert_not_called()
696