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