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