music-assistant-server

1"""Unit tests for AirPlay player."""
2
3import asyncio
4import logging
5import time
6from typing import cast
7from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
8
9import pytest
10from music_assistant_models.constants import PLAYER_CONTROL_NATIVE
11from music_assistant_models.enums import (
12    ConfigEntryType,
13    ContentType,
14    CrossfadeMode,
15    MediaType,
16    PlaybackState,
17    PlayerFeature,
18    VolumeNormalizationMode,
19)
20from music_assistant_models.errors import PlayerCommandFailed
21from music_assistant_models.media_items import AudioFormat
22
23from music_assistant.constants import CONF_SYNC_ADJUST
24from music_assistant.controllers.streams.audio import StreamsAudio
25from music_assistant.providers.airplay.constants import (
26    AIRPLAY_PCM_FORMAT,
27    CONF_AIRPLAY_CREDENTIALS,
28    CONF_ENCRYPTION,
29    CONF_FORCE_RAOP,
30    CONF_IGNORE_VOLUME,
31    CONF_PASSWORD,
32    CONF_PASSWORD_INVALID,
33    CONF_RAOP_CREDENTIALS,
34    CONF_STORED_VOLUME,
35    StreamingProtocol,
36)
37from music_assistant.providers.airplay.player import AirPlayPlayer
38from music_assistant.providers.airplay.provider import AirPlayProvider
39from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
40
41# _airplay._tcp features bitmask with the AirPlay 2 feature bits set (bit 38/48).
42AP2_FEATURES = "0x4A7FDFD5,0x3C177FDE"
43# audioFormat bits as advertised in a receiver's /info format tables.
44ALAC_44100_16 = 1 << 18
45ALAC_44100_24 = 1 << 19
46ALAC_48000_24 = 1 << 21
47
48
49def _stub_raw_config(provider: MagicMock, stored: dict[str, object] | None = None) -> None:
50    """Serve raw player config values from a dict instead of an (always truthy) mock."""
51    values = stored if stored is not None else {}
52    provider.mass.config.get_raw_player_config_value.side_effect = (
53        lambda _player_id, key, default=None: values.get(key, default)
54    )
55    provider.mass.config.set_raw_player_config_value.side_effect = lambda _player_id, key, value: (
56        values.__setitem__(key, value)
57    )
58
59
60def _stub_volume_scaling(provider: MagicMock, min_volume: int = 0, max_volume: int = 100) -> None:
61    """Apply the controller's real min/max volume scaling instead of a mock."""
62    identity = (min_volume, max_volume) == (0, 100)
63    provider.mass.players.scale_volume_to_device.side_effect = lambda _player_id, logical: (
64        logical if identity else min_volume + (logical * (max_volume - min_volume)) // 100
65    )
66    provider.mass.players.scale_volume_from_device.side_effect = lambda _player_id, device: (
67        device if identity else ((device - min_volume) * 100) // (max_volume - min_volume)
68    )
69
70
71@pytest.fixture
72def airplay_player() -> AirPlayPlayer:
73    """Create a basic AirPlayPlayer with mock defaults."""
74    provider = MagicMock()
75    _stub_raw_config(provider)
76    _stub_volume_scaling(provider)
77    return AirPlayPlayer(
78        provider=provider,
79        player_id="test_player",
80        display_name="Test Player",
81        address="127.0.0.1",
82        manufacturer="Test Manufacturer",
83        model="Test Model",
84        raop_discovery_info=None,
85        airplay_discovery_info=None,
86    )
87
88
89@pytest.mark.parametrize(
90    ("manufacturer", "model", "expected"),
91    [
92        ("Apple", "MacBookPro18,3", False),
93        ("Apple Inc.", "MacBook Air (MacBookAir10,1)", False),
94        ("Apple", "iMac (iMac21,1)", False),
95        ("Apple", "Mac mini (Mac16,11)", False),
96        ("Apple", "Mac Pro (MacPro7,1)", False),
97        ("Apple", "Mac Studio (Mac14,13)", False),
98        ("Apple", "HomePod Mini", True),
99        ("Apple", "Apple TV 4K", True),
100        ("Acme", "Mac-compatible receiver", True),
101    ],
102)
103def test_macos_devices_are_disabled_by_default(
104    manufacturer: str, model: str, expected: bool
105) -> None:
106    """Macs are disabled by default without affecting other AirPlay receivers."""
107    provider = MagicMock()
108    player = AirPlayPlayer(
109        provider=provider,
110        player_id="test_player",
111        display_name="Test Player",
112        address="127.0.0.1",
113        manufacturer=manufacturer,
114        model=model,
115        raop_discovery_info=None,
116        airplay_discovery_info=None,
117    )
118
119    assert player.enabled_by_default is expected
120    assert provider.mass.config.create_default_player_config.call_args.args[-1] is expected
121
122
123@pytest.mark.parametrize(
124    ("aiplay_properties", "raop_properties", "expected"),
125    [
126        ({b"flags": b"0x200"}, None, True),
127        ({b"sf": b"0x201"}, None, True),
128        ({b"flags": b"0x4"}, None, False),
129        ({b"sf": b"0x8"}, None, True),
130        ({b"flags": b"0x9"}, None, True),
131        (None, {b"flags": "0x200"}, True),
132        (None, {b"sf": b"0x201"}, True),
133        (None, {b"flags": b"0x4"}, False),
134        (None, {b"sf": b"0x8"}, True),
135        (None, {b"flags": b"0x9"}, True),
136        # Combined flags across discovery records should be OR-ed.
137        ({b"sf": b"0x8"}, {b"sf": b"0x200"}, True),
138        ({b"sf": b"0x200"}, {b"sf": b"0x8"}, True),
139        ({b"flags": b"0x4"}, {b"flags": b"0x0"}, False),
140        ({}, {}, False),
141    ],
142)
143def test_requires_pin_pairing(
144    airplay_player: AirPlayPlayer,
145    aiplay_properties: dict[bytes, bytes] | None,
146    raop_properties: dict[bytes, bytes] | None,
147    expected: bool,
148) -> None:
149    """Test the _requires_pairing method of AirPlayPlayer."""
150    if aiplay_properties is not None:
151        aiplay_discovery_info = MagicMock()
152        aiplay_discovery_info.properties = aiplay_properties
153        airplay_player.airplay_discovery_info = aiplay_discovery_info
154    if raop_properties is not None:
155        raop_discovery_info = MagicMock()
156        raop_discovery_info.properties = raop_properties
157        airplay_player.raop_discovery_info = raop_discovery_info
158    assert airplay_player._requires_pin_pairing() == expected
159
160
161@pytest.mark.parametrize(
162    ("aiplay_properties", "raop_properties", "expected"),
163    [
164        ({b"flags": b"0x80"}, None, True),
165        ({b"sf": b"0x81"}, None, True),
166        ({b"flags": b"0x4"}, None, False),
167        ({b"sf": b"0x80"}, None, True),
168        ({b"flags": b"0x90"}, None, True),
169        ({b"flags": b"0x1000"}, None, True),
170        (None, {b"flags": "0x80"}, True),
171        (None, {b"sf": b"0x81"}, True),
172        (None, {b"flags": b"0x4"}, False),
173        (None, {b"sf": b"0x80"}, True),
174        (None, {b"flags": b"0x90"}, True),
175        ({}, {}, False),
176    ],
177)
178def test_password_required(
179    airplay_player: AirPlayPlayer,
180    aiplay_properties: dict[bytes, bytes] | None,
181    raop_properties: dict[bytes, bytes] | None,
182    expected: bool,
183) -> None:
184    """Test the flags-based password announcements (non-Apple-TV model)."""
185    if aiplay_properties is not None:
186        aiplay_discovery_info = MagicMock()
187        aiplay_discovery_info.properties = aiplay_properties
188        aiplay_discovery_info.decoded_properties = {}
189        airplay_player.airplay_discovery_info = aiplay_discovery_info
190    if raop_properties is not None:
191        raop_discovery_info = MagicMock()
192        raop_discovery_info.properties = raop_properties
193        raop_discovery_info.decoded_properties = {}
194        airplay_player.raop_discovery_info = raop_discovery_info
195    assert airplay_player.password_required == expected
196
197
198def test_build_streaming_pairing_uses_discovered_ipv4_address() -> None:
199    """HAP pairing falls back to a discovered IPv4 address when playback uses IPv6."""
200    provider = MagicMock()
201    provider.dacp_id = "test_dacp"
202    airplay_info = MagicMock()
203    airplay_info.properties = {b"flags": b"0x80"}
204    airplay_info.port = 7000
205    player = AirPlayPlayer(
206        provider=provider,
207        player_id="test_player",
208        display_name="Test Player",
209        address="2001:db8::10",
210        manufacturer="Apple",
211        model="AppleTV",
212        raop_discovery_info=None,
213        airplay_discovery_info=airplay_info,
214    )
215    pairing_instance = MagicMock()
216
217    with (
218        patch(
219            "music_assistant.providers.airplay.player.get_primary_ip_address_from_zeroconf",
220            return_value="192.168.1.50",
221        ),
222        patch(
223            "music_assistant.providers.airplay.pairing.AirPlayPairing",
224            return_value=pairing_instance,
225        ) as pairing_cls,
226    ):
227        result = player._build_streaming_pairing(StreamingProtocol.AIRPLAY2)
228
229    assert result is pairing_instance
230    assert pairing_cls.call_args.kwargs["address"] == "192.168.1.50"
231
232
233def test_build_streaming_pairing_fails_without_ipv4_address() -> None:
234    """HAP pairing reports an actionable error when discovery has no IPv4 address."""
235    provider = MagicMock()
236    provider.dacp_id = "test_dacp"
237    airplay_info = MagicMock()
238    airplay_info.properties = {b"flags": b"0x80"}
239    airplay_info.port = 7000
240    player = AirPlayPlayer(
241        provider=provider,
242        player_id="test_player",
243        display_name="Test Player",
244        address="2001:db8::10",
245        manufacturer="Apple",
246        model="AppleTV",
247        raop_discovery_info=None,
248        airplay_discovery_info=airplay_info,
249    )
250
251    with (
252        patch(
253            "music_assistant.providers.airplay.player.get_primary_ip_address_from_zeroconf",
254            return_value="2001:db8::20",
255        ),
256        pytest.raises(PlayerCommandFailed, match="requires an IPv4"),
257    ):
258        player._build_streaming_pairing(StreamingProtocol.AIRPLAY2)
259
260
261@pytest.mark.asyncio
262async def test_config_entries_include_ignore_volume(airplay_player: AirPlayPlayer) -> None:
263    """The ignore_volume setting must be offered in the player config entries."""
264    entries = await airplay_player.get_config_entries()
265    assert any(entry.key == CONF_IGNORE_VOLUME for entry in entries)
266
267
268@pytest.mark.asyncio
269async def test_config_entries_preserve_raop_encryption_setting(
270    airplay_player: AirPlayPlayer,
271) -> None:
272    """RAOP keeps its advanced encryption toggle with the secure default enabled."""
273    entries = await airplay_player.get_config_entries()
274    entry = next(entry for entry in entries if entry.key == CONF_ENCRYPTION)
275
276    assert entry.default_value is True
277    assert entry.hidden is False
278    assert entry.advanced is True
279
280
281@pytest.mark.asyncio
282async def test_config_entries_sync_adjust_is_non_advanced(airplay_player: AirPlayPlayer) -> None:
283    """AirPlay offers sync_adjust as a discoverable (non-advanced) setting."""
284    entries = await airplay_player.get_config_entries()
285    entry = next((entry for entry in entries if entry.key == CONF_SYNC_ADJUST), None)
286    assert entry is not None
287    # non-advanced so users can find it: it is the primary control for compensating
288    # a device wired to a TV / AV receiver / amplifier that adds its own audio delay
289    assert entry.advanced is False
290
291
292def _set_discovery_info(
293    player: AirPlayPlayer,
294    *,
295    raop: bool,
296    airplay: bool,
297    airplay_features: str | None = None,
298) -> None:
299    """
300    Attach discovery mocks so the device advertises the given protocols.
301
302    :param airplay_features: When set, the _airplay service advertises this
303        ``features`` bitmask (e.g. to mark the device AirPlay 2 capable).
304    """
305    if raop:
306        raop_info = MagicMock()
307        raop_info.properties = {}
308        raop_info.decoded_properties = {}
309        player.raop_discovery_info = raop_info
310    else:
311        player.raop_discovery_info = None
312    if airplay:
313        airplay_info = MagicMock()
314        airplay_info.properties = {}
315        airplay_info.decoded_properties = {"features": airplay_features} if airplay_features else {}
316        player.airplay_discovery_info = airplay_info
317    else:
318        player.airplay_discovery_info = None
319
320
321def _make_apple_player() -> AirPlayPlayer:
322    """Create an AirPlayPlayer that identifies as a genuine Apple device."""
323    return AirPlayPlayer(
324        provider=MagicMock(),
325        player_id="test_player",
326        display_name="Test Apple TV",
327        address="127.0.0.1",
328        manufacturer="Apple",
329        model="Apple TV 4K",
330        raop_discovery_info=None,
331        airplay_discovery_info=None,
332    )
333
334
335# --- Force-RAOP escape hatch: toggle visibility ---
336
337
338@pytest.mark.asyncio
339async def test_force_raop_toggle_offered_for_non_apple_airplay2(
340    airplay_player: AirPlayPlayer,
341) -> None:
342    """A non-Apple AirPlay 2 device that also speaks RAOP gets the force-RAOP escape hatch."""
343    _set_discovery_info(airplay_player, raop=True, airplay=True, airplay_features=AP2_FEATURES)
344    entries = await airplay_player.get_config_entries()
345    entry = next((entry for entry in entries if entry.key == CONF_FORCE_RAOP), None)
346    assert entry is not None
347    assert entry.type == ConfigEntryType.BOOLEAN
348    assert entry.default_value is False
349    # advanced-only: it is a workaround, not a routine protocol choice
350    assert entry.advanced is True
351
352
353@pytest.mark.asyncio
354async def test_force_raop_toggle_hidden_for_apple_airplay2() -> None:
355    """Genuine Apple AirPlay 2 devices are always AirPlay 2, so no force-RAOP toggle is offered."""
356    player = _make_apple_player()
357    _set_discovery_info(player, raop=True, airplay=True, airplay_features=AP2_FEATURES)
358    entries = await player.get_config_entries()
359    assert all(entry.key != CONF_FORCE_RAOP for entry in entries)
360
361
362@pytest.mark.asyncio
363async def test_force_raop_toggle_hidden_for_raop_only(airplay_player: AirPlayPlayer) -> None:
364    """A RAOP-only device has nothing to force, so no toggle is offered."""
365    _set_discovery_info(airplay_player, raop=True, airplay=False)
366    entries = await airplay_player.get_config_entries()
367    assert all(entry.key != CONF_FORCE_RAOP for entry in entries)
368
369
370@pytest.mark.asyncio
371async def test_force_raop_toggle_hidden_for_airplay2_only(airplay_player: AirPlayPlayer) -> None:
372    """An AirPlay-2-only device (no RAOP service) has nothing to fall back to: no toggle."""
373    _set_discovery_info(airplay_player, raop=False, airplay=True, airplay_features=AP2_FEATURES)
374    entries = await airplay_player.get_config_entries()
375    assert all(entry.key != CONF_FORCE_RAOP for entry in entries)
376
377
378# --- Protocol resolution ---
379
380
381@pytest.mark.parametrize(
382    ("airplay_props", "raop_props", "expected"),
383    [
384        # devices advertising the AirPlay 2 feature bits get AirPlay 2
385        ({"features": AP2_FEATURES}, {}, StreamingProtocol.AIRPLAY2),
386        # the _raop ft field is used as fallback when _airplay lacks features
387        ({}, {"ft": "0x445F8A00,0x1C340"}, StreamingProtocol.AIRPLAY2),
388        # legacy receivers without the AirPlay 2 feature bits stay on RAOP
389        ({"features": "0x5A7FFFF7"}, {}, StreamingProtocol.RAOP),
390        # no features advertised at all: RAOP (safe legacy default)
391        ({}, {}, StreamingProtocol.RAOP),
392    ],
393)
394def test_protocol_resolution_follows_capability(
395    airplay_props: dict[str, str], raop_props: dict[str, str], expected: StreamingProtocol
396) -> None:
397    """Without the force toggle, protocol resolution follows the advertised AirPlay 2 bits."""
398    raop_info = MagicMock()
399    raop_info.decoded_properties = raop_props
400    airplay_info = MagicMock()
401    airplay_info.decoded_properties = airplay_props
402    player = AirPlayPlayer(
403        provider=MagicMock(),
404        player_id="test_player",
405        display_name="Test Player",
406        address="127.0.0.1",
407        manufacturer="Test Manufacturer",
408        model="Test Model",
409        raop_discovery_info=raop_info,
410        airplay_discovery_info=airplay_info,
411    )
412    _configure_player(player, {CONF_FORCE_RAOP: False})
413    assert player.protocol == expected
414
415
416def test_protocol_resolution_airplay_service_only() -> None:
417    """A device advertising only the _airplay service is AirPlay 2 even without features."""
418    airplay_info = MagicMock()
419    airplay_info.decoded_properties = {}
420    player = AirPlayPlayer(
421        provider=MagicMock(),
422        player_id="test_player",
423        display_name="Test Player",
424        address="127.0.0.1",
425        manufacturer="Test Manufacturer",
426        model="Test Model",
427        raop_discovery_info=None,
428        airplay_discovery_info=airplay_info,
429    )
430    _configure_player(player, {CONF_FORCE_RAOP: False})
431    assert player.protocol == StreamingProtocol.AIRPLAY2
432
433
434def test_force_raop_resolves_to_raop_on_non_apple_airplay2(airplay_player: AirPlayPlayer) -> None:
435    """Enabling the toggle on an eligible device forces RAOP for both resolution and stream args."""
436    _set_discovery_info(airplay_player, raop=True, airplay=True, airplay_features=AP2_FEATURES)
437    _configure_player(airplay_player, {CONF_FORCE_RAOP: True})
438    assert airplay_player.protocol == StreamingProtocol.RAOP
439    assert airplay_player.protocol_override == StreamingProtocol.RAOP
440
441
442def test_force_raop_ignored_on_apple_airplay2() -> None:
443    """A stray persisted force_raop is ignored on Apple AirPlay 2 devices (never eligible)."""
444    player = _make_apple_player()
445    _set_discovery_info(player, raop=True, airplay=True, airplay_features=AP2_FEATURES)
446    _configure_player(player, {CONF_FORCE_RAOP: True})
447    assert player.protocol == StreamingProtocol.AIRPLAY2
448    assert player.protocol_override is None
449
450
451@pytest.mark.parametrize(
452    ("stored_config", "expected"),
453    [
454        # no credentials at all: pairing is required before the player is usable
455        ({}, True),
456        # a legacy RAOP pairing keeps the player usable after the device
457        # resolves to AirPlay 2 (the binary streams RAOP-compat with the secret)
458        ({CONF_RAOP_CREDENTIALS: "clientid:secret"}, False),
459        # AirPlay 2 credentials obviously suffice as well
460        ({CONF_AIRPLAY_CREDENTIALS: "a" * 192}, False),
461    ],
462)
463def test_needs_setup_accepts_credentials_for_either_protocol(
464    airplay_player: AirPlayPlayer, stored_config: dict[str, str], expected: bool
465) -> None:
466    """A PIN-pairing device needs setup only when no credentials are stored at all."""
467    # PIN-required device that resolves to AirPlay 2 (Apple TV-like)
468    airplay_info = MagicMock()
469    airplay_info.properties = {b"flags": b"0x8"}
470    airplay_info.decoded_properties = {"features": "0x4A7FDFD5,0x3C177FDE"}
471    airplay_player.airplay_discovery_info = airplay_info
472    # credentials now live in the player's setup_data, read via get_setup_value
473    airplay_player.get_setup_value = (  # type: ignore[method-assign]
474        lambda key, default=None: stored_config.get(key, default)
475    )
476    assert airplay_player.needs_setup is expected
477
478
479# --- Hi-res playback tests ---
480
481
482def _configure_player(player: AirPlayPlayer, values: dict[str, object]) -> None:
483    """Stub the player config to return the given values."""
484    player.config.get_value.side_effect = (  # type: ignore[attr-defined]
485        lambda key, default=None: values.get(key, default)
486    )
487
488
489@pytest.mark.parametrize(
490    ("advertised_audio_formats", "force_raop", "airplay2_capable", "expected"),
491    [
492        # 24-bit advertised on the realtime stream
493        (ALAC_44100_24, False, True, [(44100, 24), (48000, 24)]),
494        # the Apple TV advertises 24-bit for its buffered stream only
495        (ALAC_48000_24, False, True, [(44100, 24), (48000, 24)]),
496        # forced RAOP cannot do 24-bit: falls back to the 16-bit base
497        (ALAC_44100_24, True, True, [(44100, 16)]),
498        # a receiver that streams RAOP never gets 24-bit, whatever it advertises
499        (ALAC_44100_24, False, False, [(44100, 16)]),
500        # only 16-bit advertised: the 16-bit default
501        (ALAC_44100_16, False, True, [(44100, 16)]),
502        # nothing advertised (unreachable device or no format tables)
503        (0, False, True, [(44100, 16)]),
504    ],
505)
506def test_hires_supported_sample_rates(
507    airplay_player: AirPlayPlayer,
508    advertised_audio_formats: int,
509    force_raop: bool,
510    airplay2_capable: bool,
511    expected: list[tuple[int, int]],
512) -> None:
513    """The formats the device advertises drive the advertised sample rates."""
514    _set_discovery_info(
515        airplay_player,
516        raop=True,
517        airplay=True,
518        airplay_features=AP2_FEATURES if airplay2_capable else None,
519    )
520    airplay_player.advertised_audio_formats = advertised_audio_formats
521    _configure_player(airplay_player, {CONF_FORCE_RAOP: force_raop})
522    assert airplay_player.supported_sample_rates == expected
523
524
525def test_get_stream_pcm_format_hires(airplay_player: AirPlayPlayer) -> None:
526    """For a 24-bit capable device the stream format is 24-bit in a s32le container."""
527    _set_discovery_info(airplay_player, raop=True, airplay=True, airplay_features=AP2_FEATURES)
528    airplay_player.advertised_audio_formats = ALAC_44100_24
529    _configure_player(airplay_player, {CONF_FORCE_RAOP: False})
530
531    session_format = AudioFormat(
532        content_type=ContentType.PCM_F32LE, sample_rate=48000, bit_depth=32
533    )
534    stream_format = airplay_player.get_stream_pcm_format(session_format)
535    # the binary expects raw s32le on stdin for --bitdepth 24
536    assert stream_format.content_type == ContentType.PCM_S32LE
537    assert stream_format.sample_rate == 48000
538    assert stream_format.bit_depth == 24
539
540    # an unsupported session rate falls back to the 44.1 kHz base
541    session_format = AudioFormat(
542        content_type=ContentType.PCM_F32LE, sample_rate=96000, bit_depth=32
543    )
544    stream_format = airplay_player.get_stream_pcm_format(session_format)
545    assert stream_format.sample_rate == 44100
546    assert stream_format.bit_depth == 24
547
548
549def test_get_stream_pcm_format_default(airplay_player: AirPlayPlayer) -> None:
550    """Without a 24-bit capable device the stream format is the 44.1/16 default."""
551    _set_discovery_info(airplay_player, raop=True, airplay=True)
552    _configure_player(airplay_player, {CONF_FORCE_RAOP: False})
553    session_format = AudioFormat(
554        content_type=ContentType.PCM_F32LE, sample_rate=48000, bit_depth=32
555    )
556    assert airplay_player.get_stream_pcm_format(session_format) == AIRPLAY_PCM_FORMAT
557
558
559@pytest.mark.asyncio
560async def test_session_pcm_format_selection(airplay_player: AirPlayPlayer) -> None:
561    """AirPlay delegates the complete session format decision to the shared selector."""
562    selected_format = AudioFormat(
563        content_type=ContentType.PCM_S24LE,
564        sample_rate=48000,
565        bit_depth=24,
566    )
567    streams_audio = cast("MagicMock", airplay_player.mass.streams.audio)
568    streams_audio.select_flow_pcm_format = AsyncMock(return_value=selected_format)
569    cast("MagicMock", airplay_player.mass.player_queues.get).return_value = None
570    media = MagicMock()
571    media.source_id = "queue1"
572    media.queue_item_id = "item1"
573    queue_item = MagicMock()
574    queue_item.streamdetails.audio_format.sample_rate = 48000
575    airplay_player.mass.player_queues.get_item.return_value = queue_item  # type: ignore[attr-defined]
576    sync_clients = [airplay_player, airplay_player]
577
578    fmt = await airplay_player._get_session_pcm_format(sync_clients, media)
579
580    assert fmt is selected_format
581    streams_audio.select_flow_pcm_format.assert_awaited_once_with(
582        airplay_player,
583        start_streamdetails=queue_item.streamdetails,
584        crossfade_enabled=False,
585        overlay_active=False,
586        fallback_sample_rate=AIRPLAY_PCM_FORMAT.sample_rate,
587        output_players=sync_clients,
588    )
589
590
591@pytest.mark.asyncio
592@pytest.mark.parametrize(
593    ("normalization_mode", "expected_content_type", "expected_bit_depth"),
594    [
595        (VolumeNormalizationMode.DISABLED, ContentType.PCM_S24LE, 24),
596        (VolumeNormalizationMode.MEASUREMENT_ONLY, ContentType.PCM_F32LE, 32),
597    ],
598)
599async def test_session_pcm_format_selects_processing_depth(
600    airplay_player: AirPlayPlayer,
601    normalization_mode: VolumeNormalizationMode,
602    expected_content_type: ContentType,
603    expected_bit_depth: int,
604) -> None:
605    """An AirPlay session only uses float PCM when processing needs headroom."""
606    _set_discovery_info(airplay_player, raop=True, airplay=True, airplay_features=AP2_FEATURES)
607    airplay_player.advertised_audio_formats = ALAC_48000_24
608    _configure_player(airplay_player, {CONF_FORCE_RAOP: False})
609    airplay_player.mass.streams.audio = StreamsAudio(airplay_player.mass)
610    cast("MagicMock", airplay_player.mass.config.get_player_dsp_config).return_value = MagicMock(
611        enabled=False
612    )
613    cast(
614        "MagicMock", airplay_player.mass.streams.get_crossfade_mode
615    ).return_value = CrossfadeMode.DISABLED
616
617    streamdetails = MagicMock()
618    streamdetails.audio_format = AudioFormat(
619        content_type=ContentType.FLAC,
620        sample_rate=48000,
621        bit_depth=24,
622    )
623    streamdetails.media_type = MediaType.TRACK
624    streamdetails.volume_normalization_mode = normalization_mode
625    queue_item = MagicMock(streamdetails=streamdetails)
626    queue = MagicMock(crossfade_enabled=False, overlay_enabled=False, overlay_source=None)
627    cast("MagicMock", airplay_player.mass.player_queues.get).return_value = queue
628    cast("MagicMock", airplay_player.mass.player_queues.get_item).return_value = queue_item
629    media = MagicMock(source_id="queue1", queue_item_id="item1", media_type=MediaType.TRACK)
630
631    fmt = await airplay_player._get_session_pcm_format([airplay_player], media)
632
633    assert fmt.content_type == expected_content_type
634    assert fmt.sample_rate == 48000
635    assert fmt.bit_depth == expected_bit_depth
636
637
638# --- Volume and Mute tests ---
639
640
641def _setup_running_stream(player: AirPlayPlayer) -> AsyncMock:
642    """Attach a mock running stream to the player and return the send_cli_command mock."""
643    stream = MagicMock()
644    stream.running = True
645    # every streaming player has a session; this one is playing, not parked
646    stream.session = MagicMock(parked=False)
647    send_cmd = AsyncMock()
648    stream.send_cli_command = send_cmd
649    player.stream = stream
650    return send_cmd
651
652
653@pytest.mark.asyncio
654async def test_volume_mute_sends_zero(airplay_player: AirPlayPlayer) -> None:
655    """Muting with a running stream should send VOLUME=0."""
656    send_cmd = _setup_running_stream(airplay_player)
657    airplay_player._attr_volume_level = 75
658
659    await airplay_player.volume_mute(True)
660
661    send_cmd.assert_called_once_with("VOLUME=0")
662    assert airplay_player._attr_volume_muted is True
663
664
665@pytest.mark.asyncio
666async def test_volume_set_skipped_while_muted(airplay_player: AirPlayPlayer) -> None:
667    """Volume changes while muted should NOT send a CLI command."""
668    send_cmd = _setup_running_stream(airplay_player)
669    airplay_player._attr_volume_muted = True
670
671    await airplay_player.volume_set(60)
672
673    send_cmd.assert_not_called()
674    assert airplay_player._attr_volume_level == 60
675
676
677@pytest.mark.asyncio
678async def test_volume_set_records_level_before_sending(airplay_player: AirPlayPlayer) -> None:
679    """A resync reading the level mid-send must observe the new volume, not the old one."""
680    send_cmd = _setup_running_stream(airplay_player)
681    airplay_player._attr_volume_level = 20
682    observed: list[int | None] = []
683
684    async def read_level_while_sending(_command: str) -> bool:
685        # stands in for the connect-time volume resync, which reads the player's
686        # level while this send is still suspended
687        await asyncio.sleep(0)
688        observed.append(airplay_player.volume_level)
689        return True
690
691    send_cmd.side_effect = read_level_while_sending
692
693    await airplay_player.volume_set(80)
694
695    assert observed == [80]
696    assert airplay_player.volume_level == 80
697
698
699@pytest.mark.asyncio
700async def test_volume_set_records_level_when_the_send_fails(
701    airplay_player: AirPlayPlayer,
702) -> None:
703    """A dropped command must not lose the requested level; the resync repairs the device."""
704    send_cmd = _setup_running_stream(airplay_player)
705    send_cmd.return_value = False
706    airplay_player._attr_volume_level = 20
707
708    await airplay_player.volume_set(80)
709
710    send_cmd.assert_awaited_once_with("VOLUME=80")
711    assert airplay_player.volume_level == 80
712
713
714@pytest.mark.asyncio
715async def test_volume_unmute_restores_volume(airplay_player: AirPlayPlayer) -> None:
716    """Unmuting with a running stream should send VOLUME={current_volume}."""
717    send_cmd = _setup_running_stream(airplay_player)
718    airplay_player._attr_volume_level = 42
719    airplay_player._attr_volume_muted = True
720
721    await airplay_player.volume_mute(False)
722
723    send_cmd.assert_called_once_with("VOLUME=42")
724    assert airplay_player._attr_volume_muted is False
725
726
727@pytest.mark.asyncio
728async def test_volume_mute_no_stream(airplay_player: AirPlayPlayer) -> None:
729    """Muting without a running stream should update state without CLI commands."""
730    airplay_player.stream = None
731
732    with patch.object(AirPlayPlayer, "update_state") as mock_update:
733        await airplay_player.volume_mute(True)
734
735        assert airplay_player._attr_volume_muted is True
736        mock_update.assert_called_once()
737
738
739def test_owns_volume_true_without_protocol_parent(airplay_player: AirPlayPlayer) -> None:
740    """A standalone AirPlay player always owns its own volume."""
741    assert airplay_player.owns_volume is True
742
743
744def test_owns_volume_true_when_parent_unresolvable(airplay_player: AirPlayPlayer) -> None:
745    """A protocol parent that no longer resolves cannot own the volume either."""
746    airplay_player.mass.players.get_player.return_value = None  # type: ignore[attr-defined]
747    airplay_player.set_protocol_parent_id("parent")
748
749    assert airplay_player.owns_volume is True
750
751
752def test_owns_volume_true_when_parent_control_is_self(airplay_player: AirPlayPlayer) -> None:
753    """This output owns the volume when the parent's control resolves to it directly."""
754    parent = MagicMock()
755    parent.volume_control_for_output.return_value = "test_player"
756    airplay_player.mass.players.get_player.return_value = parent  # type: ignore[attr-defined]
757    airplay_player.set_protocol_parent_id("parent")
758
759    assert airplay_player.owns_volume is True
760    # the control must be resolved against this player as the rendering output
761    parent.volume_control_for_output.assert_called_once_with(airplay_player.player_id)
762
763
764def test_owns_volume_true_when_parent_control_is_bridge_on_self(
765    airplay_player: AirPlayPlayer,
766) -> None:
767    """This output owns the volume when the control is a bridge riding on it."""
768    parent = MagicMock()
769    parent.volume_control_for_output.return_value = "sendspin_bridge"
770    bridge = MagicMock()
771    bridge.underlying_player_id = "test_player"
772    airplay_player.mass.players.get_player.side_effect = {  # type: ignore[attr-defined]
773        "parent": parent,
774        "sendspin_bridge": bridge,
775    }.get
776    airplay_player.set_protocol_parent_id("parent")
777
778    assert airplay_player.owns_volume is True
779
780
781@pytest.mark.parametrize("control", ["dlna_player", PLAYER_CONTROL_NATIVE])
782def test_owns_volume_false_when_another_control_owns_it(
783    airplay_player: AirPlayPlayer, control: str
784) -> None:
785    """Another control (a sibling interface, or the receiver's own native control) owns it."""
786    parent = MagicMock()
787    parent.volume_control_for_output.return_value = control
788    airplay_player.mass.players.get_player.side_effect = {  # type: ignore[attr-defined]
789        "parent": parent,
790    }.get
791    airplay_player.set_protocol_parent_id("parent")
792
793    assert airplay_player.owns_volume is False
794
795
796def test_update_volume_from_device_keeps_native_parent_feedback(
797    airplay_player: AirPlayPlayer,
798) -> None:
799    """Use DACP feedback to keep the child AirPlay volume current."""
800    parent = MagicMock()
801    parent.state.volume_level = 42
802    parent.volume_control = PLAYER_CONTROL_NATIVE
803    airplay_player.mass.players.get_player.return_value = parent  # type: ignore[attr-defined]
804    airplay_player.config.get_value.return_value = False  # type: ignore[attr-defined]
805    airplay_player.set_protocol_parent_id("parent")
806    airplay_player._attr_volume_level = 57
807    airplay_player.last_command_sent = time.time()
808
809    with patch.object(AirPlayPlayer, "update_state") as mock_update:
810        airplay_player.update_volume_from_device(57)
811
812    assert airplay_player._attr_volume_level == 57
813    airplay_player.mass.config.set_raw_player_config_value.assert_called_once_with(  # type: ignore[attr-defined]
814        airplay_player.player_id, CONF_STORED_VOLUME, 57
815    )
816    mock_update.assert_called_once()
817
818
819def test_release_foreign_mute_latch_clears_mute_owned_by_other_control(
820    airplay_player: AirPlayPlayer,
821) -> None:
822    """
823    Clear our mute when it is owned by a control that doesn't render this stream.
824
825    The mute is a latch that only an explicit unmute clears, so a mute applied while
826    a sibling interface owned the parent would otherwise start this stream silent and
827    swallow every volume command after it.
828    """
829    parent = MagicMock()
830    parent.mute_control_for_output.return_value = "cast_player"
831    cast_player = MagicMock()
832    cast_player.underlying_player_id = None
833    airplay_player.mass.players.get_player.side_effect = {  # type: ignore[attr-defined]
834        "parent": parent,
835        "cast_player": cast_player,
836    }.get
837    airplay_player.set_protocol_parent_id("parent")
838    airplay_player._attr_volume_muted = True
839
840    with patch.object(AirPlayPlayer, "update_state") as mock_update:
841        airplay_player.release_foreign_mute_latch()
842
843    assert airplay_player._attr_volume_muted is False
844    mock_update.assert_called_once()
845    # the control must be resolved against this player as the rendering output
846    parent.mute_control_for_output.assert_called_once_with(airplay_player.player_id)
847
848
849def test_release_foreign_mute_latch_keeps_mute_owned_by_this_output(
850    airplay_player: AirPlayPlayer,
851) -> None:
852    """Keep our own mute when this player owns the parent's mute."""
853    parent = MagicMock()
854    parent.mute_control_for_output.return_value = "test_player"
855    airplay_player.mass.players.get_player.return_value = parent  # type: ignore[attr-defined]
856    airplay_player.set_protocol_parent_id("parent")
857    airplay_player._attr_volume_muted = True
858
859    with patch.object(AirPlayPlayer, "update_state") as mock_update:
860        airplay_player.release_foreign_mute_latch()
861
862    assert airplay_player._attr_volume_muted is True
863    mock_update.assert_not_called()
864
865
866def test_release_foreign_mute_latch_does_nothing_when_not_muted(
867    airplay_player: AirPlayPlayer,
868) -> None:
869    """Never having latched a mute is not something to act on."""
870    parent = MagicMock()
871    airplay_player.mass.players.get_player.return_value = parent  # type: ignore[attr-defined]
872    airplay_player.set_protocol_parent_id("parent")
873    airplay_player._attr_volume_muted = False
874
875    with patch.object(AirPlayPlayer, "update_state") as mock_update:
876        airplay_player.release_foreign_mute_latch()
877
878    assert airplay_player._attr_volume_muted is False
879    parent.mute_control_for_output.assert_not_called()
880    mock_update.assert_not_called()
881
882
883# --- Pause / stop dispatch tests ---
884
885
886def test_supported_features_always_includes_pause(airplay_player: AirPlayPlayer) -> None:
887    """
888    PAUSE stays advertised whether or not the player is grouped.
889
890    Keeping PAUSE keeps the AirPlay player itself as the pause control target, so a
891    grouped pause parks the complete session (see pause()) instead of the players
892    controller falling through to a linked native player's pause, which would only
893    pause the sync leader while the other members keep playing.
894    """
895    airplay_player._attr_group_members = []
896    assert PlayerFeature.PAUSE in airplay_player.supported_features
897    # sync leader: still advertises PAUSE
898    airplay_player._attr_group_members = ["test_player", "child"]
899    assert PlayerFeature.PAUSE in airplay_player.supported_features
900
901
902def test_bridged_player_advertises_announcements_only_while_streaming(
903    airplay_player: AirPlayPlayer,
904) -> None:
905    """
906    A Sendspin-bridged player advertises PLAY_ANNOUNCEMENT only while streaming.
907
908    The bridge's live stream is a regular AirPlayStream the clip mixes into, so
909    the feature stays available then. An idle bridged player must not advertise
910    it - a dedicated announcement session would fight the bridge for the
911    device, so those announcements keep their existing routing.
912    """
913    bridge_manager = cast("AirPlayProvider", airplay_player.provider).bridge_manager
914    with patch.object(bridge_manager, "get_bridge", return_value=None):
915        assert PlayerFeature.PLAY_ANNOUNCEMENT in airplay_player.supported_features
916    streaming_bridge = MagicMock(owns_airplay_stream=True)
917    with patch.object(bridge_manager, "get_bridge", return_value=streaming_bridge):
918        assert PlayerFeature.PLAY_ANNOUNCEMENT in airplay_player.supported_features
919    idle_bridge = MagicMock(owns_airplay_stream=False)
920    with patch.object(bridge_manager, "get_bridge", return_value=idle_bridge):
921        assert PlayerFeature.PLAY_ANNOUNCEMENT not in airplay_player.supported_features
922        # ... but a configured-yet-idle bridge never hides the feature while the
923        # player runs its own session-backed stream (playing over AirPlay itself)
924        airplay_player.stream = MagicMock(running=True, session=MagicMock())
925        assert PlayerFeature.PLAY_ANNOUNCEMENT in airplay_player.supported_features
926        airplay_player.stream = None
927
928
929@pytest.mark.asyncio
930async def test_single_player_play_sends_action_play(airplay_player: AirPlayPlayer) -> None:
931    """An unsynced player resumes its paused stream in place with ACTION=PLAY."""
932    airplay_player._attr_group_members = []
933    send_cmd = _setup_running_stream(airplay_player)
934    send_cmd.return_value = True
935
936    await airplay_player.play()
937
938    send_cmd.assert_awaited_once_with("ACTION=PLAY")
939    # resume re-anchors the binary, so the tracked re-anchor shift is reset to match
940    cast("MagicMock", airplay_player.stream).reset_reanchor_shift.assert_called_once_with()
941
942
943@pytest.mark.asyncio
944async def test_single_player_play_keeps_shift_when_resume_not_delivered(
945    airplay_player: AirPlayPlayer,
946) -> None:
947    """An undelivered ACTION=PLAY leaves the tracked re-anchor shift untouched."""
948    airplay_player._attr_group_members = []
949    send_cmd = _setup_running_stream(airplay_player)
950    send_cmd.return_value = False
951
952    await airplay_player.play()
953
954    send_cmd.assert_awaited_once_with("ACTION=PLAY")
955    cast("MagicMock", airplay_player.stream).reset_reanchor_shift.assert_not_called()
956
957
958@pytest.mark.asyncio
959async def test_grouped_play_resumes_active_native_queue(airplay_player: AirPlayPlayer) -> None:
960    """A linked AirPlay group resumes the queue owned by its native parent."""
961    airplay_player._attr_group_members = ["test_player", "child"]
962    send_cmd = _setup_running_stream(airplay_player)
963    active_queue = MagicMock(queue_id="native_parent")
964
965    with (
966        patch.object(
967            airplay_player.mass.players, "get_active_queue", return_value=active_queue
968        ) as get_active_queue,
969        patch.object(
970            airplay_player.mass.player_queues, "resume", new_callable=AsyncMock
971        ) as resume_queue,
972    ):
973        await airplay_player.play()
974
975    get_active_queue.assert_called_once_with(airplay_player)
976    resume_queue.assert_awaited_once_with("native_parent", fade_in=False)
977    send_cmd.assert_not_awaited()
978
979
980@pytest.mark.asyncio
981async def test_single_player_pause_sends_action_pause(airplay_player: AirPlayPlayer) -> None:
982    """An unsynced player pauses the stream in place with ACTION=PAUSE."""
983    airplay_player._attr_group_members = []
984    airplay_player.mass.players.iter_players.return_value = []  # type: ignore[attr-defined]
985    send_cmd = _setup_running_stream(airplay_player)
986
987    with patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as mock_stop:
988        await airplay_player.pause()
989
990    send_cmd.assert_called_once_with("ACTION=PAUSE")
991    mock_stop.assert_not_called()
992
993
994@pytest.mark.asyncio
995async def test_grouped_leader_pause_parks_session(airplay_player: AirPlayPlayer) -> None:
996    """A sync leader pauses by parking the session (standby), never sending ACTION=PAUSE."""
997    airplay_player._attr_group_members = ["test_player", "child"]
998    send_cmd = _setup_running_stream(airplay_player)
999    assert airplay_player.stream is not None
1000    session = cast("MagicMock", airplay_player.stream.session)
1001    session.standby = AsyncMock(return_value=True)
1002
1003    with patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as mock_stop:
1004        await airplay_player.pause()
1005
1006    session.standby.assert_awaited_once()
1007    mock_stop.assert_not_called()
1008    send_cmd.assert_not_called()
1009
1010
1011@pytest.mark.asyncio
1012async def test_grouped_pause_falls_back_to_stop(airplay_player: AirPlayPlayer) -> None:
1013    """When a member cannot be parked, grouped pause stops the session."""
1014    airplay_player._attr_group_members = ["test_player", "child"]
1015    send_cmd = _setup_running_stream(airplay_player)
1016    assert airplay_player.stream is not None
1017    session = cast("MagicMock", airplay_player.stream.session)
1018    session.standby = AsyncMock(return_value=False)
1019
1020    with patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as mock_stop:
1021        await airplay_player.pause()
1022
1023    mock_stop.assert_called_once()
1024    send_cmd.assert_not_called()
1025
1026
1027@pytest.mark.asyncio
1028async def test_synced_child_pause_parks_session(airplay_player: AirPlayPlayer) -> None:
1029    """A synced child also pauses by parking the shared session, never ACTION=PAUSE."""
1030    airplay_player._attr_group_members = []
1031    send_cmd = _setup_running_stream(airplay_player)
1032    assert airplay_player.stream is not None
1033    session = cast("MagicMock", airplay_player.stream.session)
1034    session.standby = AsyncMock(return_value=True)
1035
1036    with (
1037        patch.object(AirPlayPlayer, "synced_to", new_callable=PropertyMock, return_value="parent"),
1038        patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as mock_stop,
1039    ):
1040        await airplay_player.pause()
1041
1042    session.standby.assert_awaited_once()
1043    mock_stop.assert_not_called()
1044    send_cmd.assert_not_called()
1045
1046
1047# --- Automatic group re-join after unexpected stream loss ---
1048
1049
1050def _make_idle_player(player_id: str = "test_player") -> AirPlayPlayer:
1051    """Create an idle, ungrouped AirPlayPlayer wired for the re-join tests."""
1052    player = AirPlayPlayer(
1053        provider=MagicMock(),
1054        player_id=player_id,
1055        display_name=f"Player {player_id}",
1056        address="127.0.0.1",
1057        manufacturer="Test Manufacturer",
1058        model="Test Model",
1059        raop_discovery_info=None,
1060        airplay_discovery_info=None,
1061    )
1062    # the synced_to property scans all players of the provider
1063    _players_mock(player).iter_players.return_value = []
1064    player._attr_group_members = []
1065    player._attr_playback_state = PlaybackState.IDLE
1066    player.stream = None
1067    return player
1068
1069
1070def _make_playing_leader(player_id: str = "leader") -> AirPlayPlayer:
1071    """Create an AirPlayPlayer that looks like the playing leader of a live session."""
1072    leader = _make_idle_player(player_id)
1073    leader._attr_playback_state = PlaybackState.PLAYING
1074    stream = MagicMock()
1075    stream.running = True
1076    stream.session = MagicMock(parked=False)
1077    leader.stream = stream
1078    return leader
1079
1080
1081def _players_mock(player: AirPlayPlayer) -> MagicMock:
1082    """Return the mocked players controller of the given player."""
1083    return cast("MagicMock", player.mass.players)
1084
1085
1086def _attach_running_session(player: AirPlayPlayer, sync_clients: list[AirPlayPlayer]) -> None:
1087    """Attach a mock running stream whose session carries the given members."""
1088    stream = MagicMock()
1089    stream.running = True
1090    stream.session = MagicMock(parked=False)
1091    stream.session.sync_clients = sync_clients
1092    player.stream = stream
1093
1094
1095_NO_DELAYS = "music_assistant.providers.airplay.player.AIRPLAY_REJOIN_ATTEMPT_DELAYS"
1096
1097
1098@pytest.mark.asyncio
1099async def test_rejoin_succeeds_on_first_attempt() -> None:
1100    """A re-join attempt joins the player back to the playing leader's session."""
1101    player = _make_idle_player()
1102    leader = _make_playing_leader()
1103    players_mock = _players_mock(player)
1104    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1105
1106    async def cmd_group(player_id: str, target_id: str) -> None:
1107        assert player_id == player.player_id
1108        assert target_id == leader.player_id
1109        _attach_running_session(player, [leader, player])
1110
1111    players_mock.cmd_group = AsyncMock(side_effect=cmd_group)
1112    players_mock.cmd_ungroup = AsyncMock()
1113
1114    with patch(_NO_DELAYS, (0, 0, 0)):
1115        await player._group_rejoin_attempts(["leader"])
1116
1117    assert players_mock.cmd_group.await_count == 1
1118    players_mock.cmd_ungroup.assert_not_awaited()
1119
1120
1121@pytest.mark.asyncio
1122async def test_rejoin_retries_after_failure_then_succeeds() -> None:
1123    """A failed attempt (device still unreachable) is retried with backoff."""
1124    player = _make_idle_player()
1125    leader = _make_playing_leader()
1126    players_mock = _players_mock(player)
1127    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1128    attempts: list[int] = []
1129
1130    async def cmd_group(_player_id: str, _target_id: str) -> None:
1131        attempts.append(1)
1132        if len(attempts) == 1:
1133            raise PlayerCommandFailed("device unreachable")
1134        _attach_running_session(player, [leader, player])
1135
1136    players_mock.cmd_group = AsyncMock(side_effect=cmd_group)
1137
1138    with patch(_NO_DELAYS, (0, 0, 0)):
1139        await player._group_rejoin_attempts(["leader"])
1140
1141    assert len(attempts) == 2
1142
1143
1144@pytest.mark.asyncio
1145async def test_rejoin_gives_up_after_all_attempts() -> None:
1146    """Without a playing group to re-join, the attempts run out and stop cleanly."""
1147    player = _make_idle_player()
1148    players_mock = _players_mock(player)
1149    players_mock.get_player.return_value = None
1150    players_mock.cmd_group = AsyncMock()
1151
1152    with patch(_NO_DELAYS, (0, 0, 0)):
1153        await player._group_rejoin_attempts(["leader"])
1154
1155    players_mock.cmd_group.assert_not_awaited()
1156
1157
1158@pytest.mark.asyncio
1159async def test_rejoin_aborts_when_player_used_meanwhile() -> None:
1160    """A player that was grouped or repurposed meanwhile is left alone."""
1161    player = _make_idle_player()
1162    leader = _make_playing_leader()
1163    players_mock = _players_mock(player)
1164    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1165    players_mock.cmd_group = AsyncMock()
1166    # the user started something else on the player during the backoff
1167    stream = MagicMock()
1168    stream.running = True
1169    player.stream = stream
1170
1171    with patch(_NO_DELAYS, (0, 0, 0)):
1172        await player._group_rejoin_attempts(["leader"])
1173
1174    players_mock.cmd_group.assert_not_awaited()
1175
1176
1177@pytest.mark.asyncio
1178async def test_rejoin_undoes_dangling_membership() -> None:
1179    """A join that yields no running stream is rolled back before the next attempt."""
1180    player = _make_idle_player()
1181    leader = _make_playing_leader()
1182    players_mock = _players_mock(player)
1183    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1184    # cmd_group "succeeds" but the late-join failed internally: no stream appears
1185    players_mock.cmd_group = AsyncMock()
1186    players_mock.cmd_ungroup = AsyncMock()
1187
1188    with patch(_NO_DELAYS, (0, 0, 0)):
1189        await player._group_rejoin_attempts(["leader"])
1190
1191    # every attempt rolled its dangling membership back
1192    assert players_mock.cmd_group.await_count == 3
1193    assert players_mock.cmd_ungroup.await_count == 3
1194
1195
1196@pytest.mark.asyncio
1197async def test_rejoin_cancelled_when_player_unavailable() -> None:
1198    """An offline player abandons the re-join right away instead of attempting."""
1199    player = _make_idle_player()
1200    leader = _make_playing_leader()
1201    players_mock = _players_mock(player)
1202    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1203    players_mock.cmd_group = AsyncMock()
1204    player._attr_available = False
1205
1206    # the later long delays prove the loop returns on the first pass
1207    with patch(_NO_DELAYS, (0, 60, 60)):
1208        await asyncio.wait_for(player._group_rejoin_attempts(["leader"]), timeout=5)
1209
1210    players_mock.cmd_group.assert_not_awaited()
1211    players_mock.get_player.assert_not_called()
1212
1213
1214@pytest.mark.asyncio
1215async def test_rejoin_aborts_when_synced_into_foreign_group() -> None:
1216    """A player the user grouped elsewhere meanwhile is left alone."""
1217    player = _make_idle_player()
1218    leader = _make_playing_leader()
1219    foreign_leader = MagicMock()
1220    foreign_leader.player_id = "other"
1221    foreign_leader.group_members = ["other", player.player_id]
1222    # the player reports it is now synced to a leader outside the original group
1223    _players_mock(player).iter_players.return_value = [foreign_leader]
1224    players_mock = _players_mock(player)
1225    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1226    players_mock.cmd_group = AsyncMock()
1227
1228    with patch(_NO_DELAYS, (0, 60, 60)):
1229        await asyncio.wait_for(player._group_rejoin_attempts(["leader"]), timeout=5)
1230
1231    players_mock.cmd_group.assert_not_awaited()
1232
1233
1234def test_resolve_rejoin_target_finds_promoted_sibling() -> None:
1235    """When the old leader is gone, a promoted (now leading) sibling is the target."""
1236    player = _make_idle_player()
1237    sibling = _make_playing_leader("sibling")
1238    sibling._attr_group_members = ["sibling", "other_member"]
1239    _players_mock(player).get_player.side_effect = lambda player_id: {"sibling": sibling}.get(
1240        player_id
1241    )
1242
1243    assert player._resolve_rejoin_target(["old_leader", "sibling"]) is sibling
1244
1245
1246def test_resolve_rejoin_target_skips_candidate_in_foreign_group() -> None:
1247    """A candidate absorbed into another group is never followed there."""
1248    player = _make_idle_player()
1249    foreign_leader = _make_playing_leader("foreign")
1250    foreign_leader._attr_group_members = ["foreign", "old_leader"]
1251    old_leader = _make_playing_leader("old_leader")
1252    # the old leader reports it is now synced to the foreign leader
1253    _players_mock(old_leader).iter_players.return_value = [foreign_leader]
1254    _players_mock(player).get_player.side_effect = lambda player_id: {
1255        "old_leader": old_leader,
1256        "foreign": foreign_leader,
1257    }.get(player_id)
1258
1259    assert player._resolve_rejoin_target(["old_leader"]) is None
1260
1261
1262@pytest.mark.parametrize(
1263    ("playback_state", "stream_running", "available", "expected"),
1264    [
1265        (PlaybackState.PLAYING, True, True, True),
1266        # a parked (paused) session has no live timeline to late-join
1267        (PlaybackState.PAUSED, True, True, False),
1268        (PlaybackState.IDLE, False, True, False),
1269        # target device itself dropped off the network
1270        (PlaybackState.PLAYING, True, False, False),
1271    ],
1272)
1273def test_resolve_rejoin_target_requires_playing_session(
1274    playback_state: PlaybackState, stream_running: bool, available: bool, expected: bool
1275) -> None:
1276    """Only an available target with an actively playing session is accepted."""
1277    player = _make_idle_player()
1278    leader = _make_playing_leader()
1279    leader._attr_playback_state = playback_state
1280    cast("MagicMock", leader.stream).running = stream_running
1281    leader._attr_available = available
1282    _players_mock(player).get_player.side_effect = lambda player_id: {"leader": leader}.get(
1283        player_id
1284    )
1285
1286    target = player._resolve_rejoin_target(["leader"])
1287    assert (target is leader) is expected
1288
1289
1290@pytest.mark.asyncio
1291async def test_rejoin_heals_session_when_membership_survived() -> None:
1292    """A player still holding sync membership (static group) heals the session only."""
1293    player = _make_idle_player()
1294    leader = _make_playing_leader()
1295    # the sync membership survived the stream loss: the player is still listed
1296    # as a member of (and synced to) the leader
1297    leader._attr_group_members = ["leader", player.player_id]
1298    _players_mock(player).iter_players.return_value = [leader]
1299    players_mock = _players_mock(player)
1300    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1301    players_mock.cmd_group = AsyncMock()
1302    players_mock.cmd_ungroup = AsyncMock()
1303    session = cast("MagicMock", leader.stream).session
1304
1305    async def add_client(joiner: AirPlayPlayer) -> None:
1306        assert joiner is player
1307        _attach_running_session(player, [leader, player])
1308
1309    session.add_client = AsyncMock(side_effect=add_client)
1310
1311    with patch(_NO_DELAYS, (0,)):
1312        await player._group_rejoin_attempts(["leader"])
1313
1314    session.add_client.assert_awaited_once()
1315    players_mock.cmd_group.assert_not_awaited()
1316    players_mock.cmd_ungroup.assert_not_awaited()
1317
1318
1319@pytest.mark.asyncio
1320async def test_rejoin_session_heal_failure_keeps_membership() -> None:
1321    """A failed session heal never touches the (configured) group membership."""
1322    player = _make_idle_player()
1323    leader = _make_playing_leader()
1324    leader._attr_group_members = ["leader", player.player_id]
1325    _players_mock(player).iter_players.return_value = [leader]
1326    players_mock = _players_mock(player)
1327    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1328    players_mock.cmd_group = AsyncMock()
1329    players_mock.cmd_ungroup = AsyncMock()
1330    session = cast("MagicMock", leader.stream).session
1331    # the late-join fails internally: no stream appears on the player
1332    session.add_client = AsyncMock()
1333
1334    with patch(_NO_DELAYS, (0,)):
1335        await player._group_rejoin_attempts(["leader"])
1336
1337    session.add_client.assert_awaited_once()
1338    players_mock.cmd_ungroup.assert_not_awaited()
1339
1340
1341@pytest.mark.asyncio
1342async def test_schedule_and_cancel_group_rejoin() -> None:
1343    """Scheduling replaces a pending task and cancelling stops it."""
1344    player = _make_idle_player()
1345    _players_mock(player).get_player.return_value = None
1346    cast("MagicMock", player.mass).create_task = lambda coro: (
1347        asyncio.get_running_loop().create_task(coro)
1348    )
1349
1350    with patch(_NO_DELAYS, (60, 60, 60)):
1351        player.schedule_group_rejoin(["leader"])
1352        first_task = player._rejoin_task
1353        assert first_task is not None
1354        # a second death replaces the pending schedule
1355        player.schedule_group_rejoin(["leader"])
1356        second_task = player._rejoin_task
1357        assert second_task is not None
1358        assert second_task is not first_task
1359        await asyncio.sleep(0)
1360        assert first_task.cancelled()
1361        # any deliberate use of the player cancels the pending re-join
1362        player.cancel_group_rejoin()
1363        assert player._rejoin_task is None
1364        await asyncio.sleep(0)
1365        assert second_task.cancelled()
1366
1367
1368@pytest.mark.asyncio
1369async def test_stop_cancels_pending_rejoin() -> None:
1370    """An explicit stop command on the player drops the pending re-join."""
1371    player = _make_idle_player()
1372    _players_mock(player).get_player.return_value = None
1373    cast("MagicMock", player.mass).create_task = lambda coro: (
1374        asyncio.get_running_loop().create_task(coro)
1375    )
1376
1377    with (
1378        patch(_NO_DELAYS, (60,)),
1379        patch.object(AirPlayPlayer, "update_state"),
1380    ):
1381        player.schedule_group_rejoin(["leader"])
1382        rejoin_task = player._rejoin_task
1383        assert rejoin_task is not None
1384        await player.stop()
1385        assert player._rejoin_task is None
1386        await asyncio.sleep(0)
1387        assert rejoin_task.cancelled()
1388
1389
1390# --- Group membership and the leader's stream session ---
1391
1392
1393@pytest.mark.asyncio
1394async def test_set_members_adds_the_child_to_the_running_session() -> None:
1395    """A member joining a leader with a live session is added to that session."""
1396    leader = _make_playing_leader()
1397    child = _make_idle_player("child")
1398    _attach_running_session(leader, [leader])
1399    session = cast("MagicMock", leader.stream).session
1400    session.add_client = AsyncMock()
1401    _players_mock(leader).get_player.side_effect = lambda player_id: {"child": child}.get(player_id)
1402
1403    await leader.set_members(player_ids_to_add=["child"])
1404
1405    session.add_client.assert_awaited_once_with(child)
1406    assert leader.group_members == ["leader", "child"]
1407
1408
1409def test_live_session_members_reports_who_the_session_actually_feeds() -> None:
1410    """Group membership outlives the session, so only the session itself can answer."""
1411    leader = _make_playing_leader()
1412    leader._attr_group_members = ["leader", "child"]
1413    # the session dropped the child (e.g. its receiver never answered our clock)
1414    _attach_running_session(leader, [leader])
1415
1416    assert leader.live_session_members == ["leader"]
1417
1418    # no session means nobody is being rendered with, whatever the group says
1419    stream = cast("MagicMock", leader.stream)
1420    stream.running = False
1421    assert leader.live_session_members == []
1422    stream.running = True
1423    stream.session = None
1424    assert leader.live_session_members == []
1425    leader.stream = None
1426    assert leader.live_session_members == []
1427
1428
1429@pytest.mark.asyncio
1430async def test_set_members_warns_when_the_leader_has_no_session(
1431    caplog: pytest.LogCaptureFixture,
1432) -> None:
1433    """A member joining a leader that renders through a protocol gets no audio, loudly."""
1434    leader = _make_idle_player("leader")
1435    leader.logger = logging.getLogger("test.airplay.player")
1436    child = _make_idle_player("child")
1437    _players_mock(leader).get_player.side_effect = lambda player_id: {"child": child}.get(player_id)
1438    # the leader hands its audio to one of its output protocols, so it has no session
1439    leader.set_active_output_protocol("bridge_leader")
1440
1441    with caplog.at_level(logging.WARNING):
1442        await leader.set_members(player_ids_to_add=["child"])
1443
1444    assert leader.group_members == ["leader", "child"]
1445    assert "no stream session to join" in caplog.text
1446
1447
1448def _attach_live_stream(player: AirPlayPlayer, session: AirPlayStreamSession) -> MagicMock:
1449    """Attach a mock stream that is connected and fed by the given session."""
1450    stream = MagicMock()
1451    stream.running = True
1452    stream.connected = True
1453    stream.session = session
1454    stream.send_cli_command = AsyncMock(return_value=True)
1455    stream.stop = AsyncMock()
1456    player.stream = stream
1457    return stream
1458
1459
1460@pytest.mark.asyncio
1461async def test_play_after_ungrouping_a_parked_group_resumes_via_the_queue() -> None:
1462    """
1463    Breaking a parked group up leaves the remaining player alone with the park.
1464
1465    Its binary is held at standby with nothing being fed, so the resume still has
1466    to re-anchor through the queue: ACTION=PLAY carries no anchor and would
1467    report playback over silence.
1468    """
1469    leader = _make_idle_player("leader")
1470    child = _make_idle_player("child")
1471    session = AirPlayStreamSession(
1472        MagicMock(mass=leader.mass), [leader, child], AIRPLAY_PCM_FORMAT, MagicMock()
1473    )
1474    leader_stream = _attach_live_stream(leader, session)
1475    child_stream = _attach_live_stream(child, session)
1476    leader._attr_group_members = ["leader", "child"]
1477    players = _players_mock(leader)
1478    players.get_player.side_effect = lambda player_id: {"leader": leader, "child": child}.get(
1479        player_id
1480    )
1481    players.get_active_queue.return_value = MagicMock(queue_id="leader")
1482    resume_queue = AsyncMock()
1483    cast("MagicMock", leader.mass).player_queues.resume = resume_queue
1484
1485    await leader.pause()
1486    await leader.set_members(player_ids_to_remove=["child"])
1487    leader_stream.send_cli_command.reset_mock()
1488    await leader.play()
1489
1490    # the removal stops only the child; the leader keeps its parked session
1491    child_stream.stop.assert_awaited_once()
1492    leader_stream.stop.assert_not_awaited()
1493    assert leader.group_members == []
1494    assert session.sync_clients == [leader]
1495    assert session.parked is True
1496    resume_queue.assert_awaited_once_with("leader", fade_in=False)
1497    leader_stream.send_cli_command.assert_not_awaited()
1498
1499
1500@pytest.mark.asyncio
1501async def test_leader_stepping_out_alone_keeps_the_session_for_the_others() -> None:
1502    """
1503    A leader that only removes itself hands the live session to the members left behind.
1504
1505    The leader is not asked to take the others with it, so tearing the session
1506    down here would cut off members that are still supposed to be playing.
1507    """
1508    leader = _make_idle_player("leader")
1509    child = _make_idle_player("child")
1510    session = AirPlayStreamSession(
1511        MagicMock(mass=leader.mass), [leader, child], AIRPLAY_PCM_FORMAT, MagicMock()
1512    )
1513    leader_stream = _attach_live_stream(leader, session)
1514    child_stream = _attach_live_stream(child, session)
1515    leader._attr_group_members = ["leader", "child"]
1516    leader._attr_playback_state = PlaybackState.PLAYING
1517    child._attr_playback_state = PlaybackState.PLAYING
1518    lookup = {"leader": leader, "child": child}
1519    for player in (leader, child):
1520        players = _players_mock(player)
1521        players.get_player.side_effect = lookup.get
1522        players.iter_players.return_value = [leader, child]
1523
1524    await leader.set_members(player_ids_to_remove=["leader"])
1525
1526    leader_stream.stop.assert_awaited_once()
1527    child_stream.stop.assert_not_awaited()
1528    assert session.sync_clients == [child]
1529    assert leader.group_members == []
1530    # nothing claims the remaining member anymore: its caller picks the new leader
1531    assert child.synced_to is None
1532
1533
1534@pytest.mark.asyncio
1535async def test_ungroup_on_a_sync_leader_dissolves_the_whole_group() -> None:
1536    """
1537    Ungrouping a sync leader must release its members, not just the leader itself.
1538
1539    A leader lists itself in group_members, so the default ungroup asks to remove
1540    the leader AND every member in one call.
1541    """
1542    leader = _make_idle_player("leader")
1543    child = _make_idle_player("child")
1544    session = AirPlayStreamSession(
1545        MagicMock(mass=leader.mass), [leader, child], AIRPLAY_PCM_FORMAT, MagicMock()
1546    )
1547    leader_stream = _attach_live_stream(leader, session)
1548    child_stream = _attach_live_stream(child, session)
1549    leader._attr_group_members = ["leader", "child"]
1550    leader._attr_playback_state = PlaybackState.PLAYING
1551    child._attr_playback_state = PlaybackState.PLAYING
1552    lookup = {"leader": leader, "child": child}
1553    for player in (leader, child):
1554        players = _players_mock(player)
1555        players.get_player.side_effect = lookup.get
1556        players.iter_players.return_value = [leader, child]
1557
1558    await leader.ungroup()
1559
1560    assert leader.group_members == []
1561    leader_stream.stop.assert_awaited_once()
1562    child_stream.stop.assert_awaited_once()
1563    assert session.sync_clients == []
1564
1565
1566# --- Device password ---
1567
1568
1569def _set_password_discovery(
1570    player: AirPlayPlayer,
1571    *,
1572    flags: str = "0x0",
1573    pw: str = "",
1574    password: str | None = None,
1575    paired: bool = False,
1576) -> None:
1577    """
1578    Attach an AirPlay 2 + RAOP device announcing password protection.
1579
1580    :param flags: The _airplay service sf/flags bitmask (0x80 marks a password).
1581    :param pw: The legacy ``pw`` boolean published by the _raop service.
1582    :param password: The device password stored in the player config, if any.
1583    :param paired: Whether AirPlay 2 pairing credentials are stored for the device.
1584    """
1585    airplay_info = MagicMock()
1586    airplay_info.decoded_properties = {"features": AP2_FEATURES, "flags": flags}
1587    airplay_info.properties = {b"flags": flags.encode()}
1588    player.airplay_discovery_info = airplay_info
1589    raop_info = MagicMock()
1590    raop_info.decoded_properties = {"pw": pw} if pw else {}
1591    raop_info.properties = {}
1592    player.raop_discovery_info = raop_info
1593    _configure_player(player, {CONF_FORCE_RAOP: False, CONF_PASSWORD: password})
1594    credentials = {CONF_AIRPLAY_CREDENTIALS: "a" * 192} if paired else {}
1595    player.get_setup_value = (  # type: ignore[method-assign]
1596        lambda key, default=None: credentials.get(key, default)
1597    )
1598
1599
1600@pytest.mark.asyncio
1601async def test_password_entry_is_never_offered_in_the_settings(
1602    airplay_player: AirPlayPlayer,
1603) -> None:
1604    """The password is storage only: the setup flow is the sole way to enter it."""
1605    _set_password_discovery(airplay_player, flags="0x80")
1606    assert airplay_player.password_required is True
1607
1608    entries = await airplay_player.get_config_entries()
1609    entry = next(entry for entry in entries if entry.key == CONF_PASSWORD)
1610    assert entry.hidden is True
1611
1612
1613@pytest.mark.parametrize(
1614    ("flags", "pw"),
1615    [
1616        # AirPlay 2 announces password protection through the flags bit...
1617        ("0x80", ""),
1618        # ...a legacy RAOP receiver through the classic pw boolean
1619        ("0x0", "true"),
1620    ],
1621)
1622def test_announced_password_without_one_stored_needs_setup(
1623    airplay_player: AirPlayPlayer, flags: str, pw: str
1624) -> None:
1625    """A device that asks for a password it never got must be set up first."""
1626    _set_password_discovery(airplay_player, flags=flags, pw=pw)
1627
1628    assert airplay_player.password_required is True
1629    assert airplay_player.needs_setup is True
1630    assert airplay_player.setup_reason == "password_required"
1631
1632
1633def test_apple_tv_password_bit_alone_does_not_need_setup(airplay_player: AirPlayPlayer) -> None:
1634    """Apple TVs raise the generic password bit at all times; it means nothing there."""
1635    # a paired Apple TV without a password set must not be sent back into setup
1636    _set_password_discovery(airplay_player, flags="0x4c4", paired=True)
1637    airplay_player.device_info.model = "AppleTV14,1"
1638
1639    assert airplay_player.password_required is False
1640    assert airplay_player.needs_setup is False
1641
1642
1643def test_apple_tv_with_a_password_set_needs_setup(airplay_player: AirPlayPlayer) -> None:
1644    """The tvOS-specific flags bit is the Apple TV's only password announcement."""
1645    _set_password_discovery(airplay_player, flags="0x14c4")
1646    airplay_player.device_info.model = "AppleTV11,1"
1647
1648    assert airplay_player.password_required is True
1649    assert airplay_player.needs_setup is True
1650    assert airplay_player.setup_reason == "password_required"
1651
1652
1653@pytest.mark.parametrize(
1654    ("flags", "pw", "paired"),
1655    [
1656        # AirPlay 2 collects the password as part of pairing, so it ends up with both
1657        ("0x80", "", True),
1658        # a legacy RAOP receiver has no pairing at all: the password is enough
1659        ("0x0", "true", False),
1660    ],
1661)
1662def test_stored_password_clears_the_setup_requirement(
1663    airplay_player: AirPlayPlayer, flags: str, pw: str, paired: bool
1664) -> None:
1665    """Once the password is stored the player is ready to use again."""
1666    _set_password_discovery(airplay_player, flags=flags, pw=pw, password="hunter2", paired=paired)
1667
1668    assert airplay_player.needs_setup is False
1669    assert airplay_player.setup_reason is None
1670
1671
1672def test_rejected_password_marker_forces_setup(airplay_player: AirPlayPlayer) -> None:
1673    """A password the device rejected sends an otherwise ready player back into setup."""
1674    # the migration case: a paired device that gained password protection later
1675    _set_password_discovery(airplay_player, flags="0x80", password="wrong", paired=True)
1676    ready_before = airplay_player.needs_setup
1677
1678    airplay_player.set_password_invalid(True)
1679
1680    assert ready_before is False
1681    assert airplay_player.password_invalid is True
1682    assert airplay_player.needs_setup is True
1683    assert airplay_player.setup_reason == "password_required"
1684
1685
1686def test_rejected_password_marker_survives_a_restart(airplay_player: AirPlayPlayer) -> None:
1687    """The marker is persisted as a raw player config value, not just in memory."""
1688    _set_password_discovery(airplay_player, flags="0x80", password="wrong", paired=True)
1689
1690    airplay_player.set_password_invalid(True)
1691
1692    airplay_player.mass.config.set_raw_player_config_value.assert_called_once_with(  # type: ignore[attr-defined]
1693        "test_player", CONF_PASSWORD_INVALID, True
1694    )
1695
1696
1697def test_clearing_the_marker_only_writes_when_it_was_set(airplay_player: AirPlayPlayer) -> None:
1698    """Every successful connect clears the marker, but must not write the config."""
1699    _set_password_discovery(airplay_player, flags="0x80", password="hunter2", paired=True)
1700    set_raw = airplay_player.mass.config.set_raw_player_config_value
1701
1702    airplay_player.set_password_invalid(False)
1703    set_raw.assert_not_called()  # type: ignore[attr-defined]
1704
1705    airplay_player.set_password_invalid(True)
1706    assert airplay_player.needs_setup is True
1707    airplay_player.set_password_invalid(False)
1708
1709    assert airplay_player.password_invalid is False
1710    assert airplay_player.needs_setup is False
1711
1712
1713def test_rejected_password_is_published_to_clients(airplay_player: AirPlayPlayer) -> None:
1714    """The new setup requirement must reach the wire state, not just the property."""
1715    _set_password_discovery(airplay_player, flags="0x80", password="wrong", paired=True)
1716    airplay_player.update_state()
1717    before = airplay_player.state
1718    assert before.needs_setup is False
1719    assert before.available is True
1720
1721    airplay_player.set_password_invalid(True)
1722
1723    # needs_setup/setup_reason are part of the player's own state inputs, so the
1724    # update is neither short-circuited nor left to the next unrelated update
1725    after = airplay_player.state
1726    assert after.needs_setup is True
1727    assert after.setup_reason == "password_required"
1728    assert after.available is False
1729    changed = airplay_player.mass.players.signal_player_state_update.call_args[0][1]  # type: ignore[attr-defined]
1730    assert "needs_setup" in changed
1731
1732
1733def test_pin_pairing_keeps_its_own_setup_reason(airplay_player: AirPlayPlayer) -> None:
1734    """A device that only needs PIN pairing is not reported as a password problem."""
1735    airplay_info = MagicMock()
1736    airplay_info.decoded_properties = {"features": AP2_FEATURES}
1737    airplay_info.properties = {b"flags": b"0x8"}
1738    airplay_player.airplay_discovery_info = airplay_info
1739    airplay_player.get_setup_value = lambda key, default=None: default  # type: ignore[method-assign]  # noqa: ARG005
1740
1741    assert airplay_player.needs_setup is True
1742    assert airplay_player.setup_reason == "pairing_required"
1743