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