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    streamdetails.media_type = MediaType.TRACK
679    streamdetails.volume_normalization_mode = normalization_mode
680    queue_item = MagicMock(streamdetails=streamdetails)
681    queue = MagicMock(crossfade_enabled=False, overlay_enabled=False, overlay_source=None)
682    cast("MagicMock", airplay_player.mass.player_queues.get).return_value = queue
683    cast("MagicMock", airplay_player.mass.player_queues.get_item).return_value = queue_item
684    media = MagicMock(source_id="queue1", queue_item_id="item1", media_type=MediaType.TRACK)
685
686    fmt = await airplay_player._get_session_pcm_format([airplay_player], media)
687
688    assert fmt.content_type == expected_content_type
689    assert fmt.sample_rate == 48000
690    assert fmt.bit_depth == expected_bit_depth
691
692
693# --- Volume and Mute tests ---
694
695
696def _setup_running_stream(player: AirPlayPlayer) -> AsyncMock:
697    """Attach a mock running stream to the player and return the send_cli_command mock."""
698    stream = MagicMock()
699    stream.running = True
700    # every streaming player has a session; this one is playing, not parked
701    stream.session = MagicMock(parked=False)
702    send_cmd = AsyncMock()
703    stream.send_cli_command = send_cmd
704    player.stream = stream
705    return send_cmd
706
707
708@pytest.mark.asyncio
709async def test_volume_mute_sends_zero(airplay_player: AirPlayPlayer) -> None:
710    """Muting with a running stream should send VOLUME=0."""
711    send_cmd = _setup_running_stream(airplay_player)
712    airplay_player._attr_volume_level = 75
713
714    await airplay_player.volume_mute(True)
715
716    send_cmd.assert_called_once_with("VOLUME=0")
717    assert airplay_player._attr_volume_muted is True
718
719
720@pytest.mark.asyncio
721async def test_volume_set_skipped_while_muted(airplay_player: AirPlayPlayer) -> None:
722    """Volume changes while muted should NOT send a CLI command."""
723    send_cmd = _setup_running_stream(airplay_player)
724    airplay_player._attr_volume_muted = True
725
726    await airplay_player.volume_set(60)
727
728    send_cmd.assert_not_called()
729    assert airplay_player._attr_volume_level == 60
730
731
732@pytest.mark.asyncio
733async def test_volume_set_records_level_before_sending(airplay_player: AirPlayPlayer) -> None:
734    """A resync reading the level mid-send must observe the new volume, not the old one."""
735    send_cmd = _setup_running_stream(airplay_player)
736    airplay_player._attr_volume_level = 20
737    observed: list[int | None] = []
738
739    async def read_level_while_sending(_command: str) -> bool:
740        # stands in for the connect-time volume resync, which reads the player's
741        # level while this send is still suspended
742        await asyncio.sleep(0)
743        observed.append(airplay_player.volume_level)
744        return True
745
746    send_cmd.side_effect = read_level_while_sending
747
748    await airplay_player.volume_set(80)
749
750    assert observed == [80]
751    assert airplay_player.volume_level == 80
752
753
754@pytest.mark.asyncio
755async def test_volume_set_records_level_when_the_send_fails(
756    airplay_player: AirPlayPlayer,
757) -> None:
758    """A dropped command must not lose the requested level; the resync repairs the device."""
759    send_cmd = _setup_running_stream(airplay_player)
760    send_cmd.return_value = False
761    airplay_player._attr_volume_level = 20
762
763    await airplay_player.volume_set(80)
764
765    send_cmd.assert_awaited_once_with("VOLUME=80")
766    assert airplay_player.volume_level == 80
767
768
769@pytest.mark.asyncio
770async def test_volume_unmute_restores_volume(airplay_player: AirPlayPlayer) -> None:
771    """Unmuting with a running stream should send VOLUME={current_volume}."""
772    send_cmd = _setup_running_stream(airplay_player)
773    airplay_player._attr_volume_level = 42
774    airplay_player._attr_volume_muted = True
775
776    await airplay_player.volume_mute(False)
777
778    send_cmd.assert_called_once_with("VOLUME=42")
779    assert airplay_player._attr_volume_muted is False
780
781
782@pytest.mark.asyncio
783async def test_volume_mute_no_stream(airplay_player: AirPlayPlayer) -> None:
784    """Muting without a running stream should update state without CLI commands."""
785    airplay_player.stream = None
786
787    with patch.object(AirPlayPlayer, "update_state") as mock_update:
788        await airplay_player.volume_mute(True)
789
790        assert airplay_player._attr_volume_muted is True
791        mock_update.assert_called_once()
792
793
794def test_owns_volume_true_without_protocol_parent(airplay_player: AirPlayPlayer) -> None:
795    """A standalone AirPlay player always owns its own volume."""
796    assert airplay_player.owns_volume is True
797
798
799def test_owns_volume_true_when_parent_unresolvable(airplay_player: AirPlayPlayer) -> None:
800    """A protocol parent that no longer resolves cannot own the volume either."""
801    airplay_player.mass.players.get_player.return_value = None  # type: ignore[attr-defined]
802    airplay_player.set_protocol_parent_id("parent")
803
804    assert airplay_player.owns_volume is True
805
806
807def test_owns_volume_true_when_parent_control_is_self(airplay_player: AirPlayPlayer) -> None:
808    """This output owns the volume when the parent's control resolves to it directly."""
809    parent = MagicMock()
810    parent.volume_control_for_output.return_value = "test_player"
811    airplay_player.mass.players.get_player.return_value = parent  # type: ignore[attr-defined]
812    airplay_player.set_protocol_parent_id("parent")
813
814    assert airplay_player.owns_volume is True
815    # the control must be resolved against this player as the rendering output
816    parent.volume_control_for_output.assert_called_once_with(airplay_player.player_id)
817
818
819def test_owns_volume_true_when_parent_control_is_bridge_on_self(
820    airplay_player: AirPlayPlayer,
821) -> None:
822    """This output owns the volume when the control is a bridge riding on it."""
823    parent = MagicMock()
824    parent.volume_control_for_output.return_value = "sendspin_bridge"
825    bridge = MagicMock()
826    bridge.underlying_player_id = "test_player"
827    airplay_player.mass.players.get_player.side_effect = {  # type: ignore[attr-defined]
828        "parent": parent,
829        "sendspin_bridge": bridge,
830    }.get
831    airplay_player.set_protocol_parent_id("parent")
832
833    assert airplay_player.owns_volume is True
834
835
836@pytest.mark.parametrize("control", ["dlna_player", PLAYER_CONTROL_NATIVE])
837def test_owns_volume_false_when_another_control_owns_it(
838    airplay_player: AirPlayPlayer, control: str
839) -> None:
840    """Another control (a sibling interface, or the receiver's own native control) owns it."""
841    parent = MagicMock()
842    parent.volume_control_for_output.return_value = control
843    airplay_player.mass.players.get_player.side_effect = {  # type: ignore[attr-defined]
844        "parent": parent,
845    }.get
846    airplay_player.set_protocol_parent_id("parent")
847
848    assert airplay_player.owns_volume is False
849
850
851def test_update_volume_from_device_keeps_native_parent_feedback(
852    airplay_player: AirPlayPlayer,
853) -> None:
854    """Use DACP feedback to keep the child AirPlay volume current."""
855    parent = MagicMock()
856    parent.state.volume_level = 42
857    parent.volume_control = PLAYER_CONTROL_NATIVE
858    airplay_player.mass.players.get_player.return_value = parent  # type: ignore[attr-defined]
859    airplay_player.config.get_value.return_value = False  # type: ignore[attr-defined]
860    airplay_player.set_protocol_parent_id("parent")
861    airplay_player._attr_volume_level = 57
862    airplay_player.last_command_sent = time.time()
863
864    with patch.object(AirPlayPlayer, "update_state") as mock_update:
865        airplay_player.update_volume_from_device(57)
866
867    assert airplay_player._attr_volume_level == 57
868    airplay_player.mass.config.set_raw_player_config_value.assert_called_once_with(  # type: ignore[attr-defined]
869        airplay_player.player_id, CONF_STORED_VOLUME, 57
870    )
871    mock_update.assert_called_once()
872
873
874def test_release_foreign_mute_latch_clears_mute_owned_by_other_control(
875    airplay_player: AirPlayPlayer,
876) -> None:
877    """
878    Clear our mute when it is owned by a control that doesn't render this stream.
879
880    The mute is a latch that only an explicit unmute clears, so a mute applied while
881    a sibling interface owned the parent would otherwise start this stream silent and
882    swallow every volume command after it.
883    """
884    parent = MagicMock()
885    parent.mute_control_for_output.return_value = "cast_player"
886    cast_player = MagicMock()
887    cast_player.underlying_player_id = None
888    airplay_player.mass.players.get_player.side_effect = {  # type: ignore[attr-defined]
889        "parent": parent,
890        "cast_player": cast_player,
891    }.get
892    airplay_player.set_protocol_parent_id("parent")
893    airplay_player._attr_volume_muted = True
894
895    with patch.object(AirPlayPlayer, "update_state") as mock_update:
896        airplay_player.release_foreign_mute_latch()
897
898    assert airplay_player._attr_volume_muted is False
899    mock_update.assert_called_once()
900    # the control must be resolved against this player as the rendering output
901    parent.mute_control_for_output.assert_called_once_with(airplay_player.player_id)
902
903
904def test_release_foreign_mute_latch_keeps_mute_owned_by_this_output(
905    airplay_player: AirPlayPlayer,
906) -> None:
907    """Keep our own mute when this player owns the parent's mute."""
908    parent = MagicMock()
909    parent.mute_control_for_output.return_value = "test_player"
910    airplay_player.mass.players.get_player.return_value = parent  # type: ignore[attr-defined]
911    airplay_player.set_protocol_parent_id("parent")
912    airplay_player._attr_volume_muted = True
913
914    with patch.object(AirPlayPlayer, "update_state") as mock_update:
915        airplay_player.release_foreign_mute_latch()
916
917    assert airplay_player._attr_volume_muted is True
918    mock_update.assert_not_called()
919
920
921def test_release_foreign_mute_latch_does_nothing_when_not_muted(
922    airplay_player: AirPlayPlayer,
923) -> None:
924    """Never having latched a mute is not something to act on."""
925    parent = MagicMock()
926    airplay_player.mass.players.get_player.return_value = parent  # type: ignore[attr-defined]
927    airplay_player.set_protocol_parent_id("parent")
928    airplay_player._attr_volume_muted = False
929
930    with patch.object(AirPlayPlayer, "update_state") as mock_update:
931        airplay_player.release_foreign_mute_latch()
932
933    assert airplay_player._attr_volume_muted is False
934    parent.mute_control_for_output.assert_not_called()
935    mock_update.assert_not_called()
936
937
938# --- Pause / stop dispatch tests ---
939
940
941def test_supported_features_always_includes_pause(airplay_player: AirPlayPlayer) -> None:
942    """
943    PAUSE stays advertised whether or not the player is grouped.
944
945    Keeping PAUSE keeps the AirPlay player itself as the pause control target, so a
946    grouped pause parks the complete session (see pause()) instead of the players
947    controller falling through to a linked native player's pause, which would only
948    pause the sync leader while the other members keep playing.
949    """
950    airplay_player._attr_group_members = []
951    assert PlayerFeature.PAUSE in airplay_player.supported_features
952    # sync leader: still advertises PAUSE
953    airplay_player._attr_group_members = ["test_player", "child"]
954    assert PlayerFeature.PAUSE in airplay_player.supported_features
955
956
957def test_bridged_player_advertises_announcements_only_while_streaming(
958    airplay_player: AirPlayPlayer,
959) -> None:
960    """
961    A Sendspin-bridged player advertises PLAY_ANNOUNCEMENT only while streaming.
962
963    The bridge's live stream is a regular AirPlayStream the clip mixes into, so
964    the feature stays available then. An idle bridged player must not advertise
965    it - a dedicated announcement session would fight the bridge for the
966    device, so those announcements keep their existing routing.
967    """
968    bridge_manager = cast("AirPlayProvider", airplay_player.provider).bridge_manager
969    with patch.object(bridge_manager, "get_bridge", return_value=None):
970        assert PlayerFeature.PLAY_ANNOUNCEMENT in airplay_player.supported_features
971    streaming_bridge = MagicMock(owns_airplay_stream=True)
972    with patch.object(bridge_manager, "get_bridge", return_value=streaming_bridge):
973        assert PlayerFeature.PLAY_ANNOUNCEMENT in airplay_player.supported_features
974    idle_bridge = MagicMock(owns_airplay_stream=False)
975    with patch.object(bridge_manager, "get_bridge", return_value=idle_bridge):
976        assert PlayerFeature.PLAY_ANNOUNCEMENT not in airplay_player.supported_features
977        # ... but a configured-yet-idle bridge never hides the feature while the
978        # player runs its own session-backed stream (playing over AirPlay itself)
979        airplay_player.stream = MagicMock(running=True, session=MagicMock())
980        assert PlayerFeature.PLAY_ANNOUNCEMENT in airplay_player.supported_features
981        airplay_player.stream = None
982
983
984@pytest.mark.asyncio
985async def test_single_player_play_sends_action_play(airplay_player: AirPlayPlayer) -> None:
986    """An unsynced player resumes its paused stream in place with ACTION=PLAY."""
987    airplay_player._attr_group_members = []
988    send_cmd = _setup_running_stream(airplay_player)
989    send_cmd.return_value = True
990
991    await airplay_player.play()
992
993    send_cmd.assert_awaited_once_with("ACTION=PLAY")
994    # resume re-anchors the binary, so the tracked re-anchor shift is reset to match
995    cast("MagicMock", airplay_player.stream).reset_reanchor_shift.assert_called_once_with()
996
997
998@pytest.mark.asyncio
999async def test_single_player_play_keeps_shift_when_resume_not_delivered(
1000    airplay_player: AirPlayPlayer,
1001) -> None:
1002    """An undelivered ACTION=PLAY leaves the tracked re-anchor shift untouched."""
1003    airplay_player._attr_group_members = []
1004    send_cmd = _setup_running_stream(airplay_player)
1005    send_cmd.return_value = False
1006
1007    await airplay_player.play()
1008
1009    send_cmd.assert_awaited_once_with("ACTION=PLAY")
1010    cast("MagicMock", airplay_player.stream).reset_reanchor_shift.assert_not_called()
1011
1012
1013@pytest.mark.asyncio
1014async def test_grouped_play_resumes_active_native_queue(airplay_player: AirPlayPlayer) -> None:
1015    """A linked AirPlay group resumes the queue owned by its native parent."""
1016    airplay_player._attr_group_members = ["test_player", "child"]
1017    send_cmd = _setup_running_stream(airplay_player)
1018    active_queue = MagicMock(queue_id="native_parent")
1019
1020    with (
1021        patch.object(
1022            airplay_player.mass.players, "get_active_queue", return_value=active_queue
1023        ) as get_active_queue,
1024        patch.object(
1025            airplay_player.mass.player_queues, "resume", new_callable=AsyncMock
1026        ) as resume_queue,
1027    ):
1028        await airplay_player.play()
1029
1030    get_active_queue.assert_called_once_with(airplay_player)
1031    resume_queue.assert_awaited_once_with("native_parent", fade_in=False)
1032    send_cmd.assert_not_awaited()
1033
1034
1035@pytest.mark.asyncio
1036async def test_single_player_pause_sends_action_pause(airplay_player: AirPlayPlayer) -> None:
1037    """An unsynced player pauses the stream in place with ACTION=PAUSE."""
1038    airplay_player._attr_group_members = []
1039    airplay_player.mass.players.iter_players.return_value = []  # type: ignore[attr-defined]
1040    send_cmd = _setup_running_stream(airplay_player)
1041
1042    with patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as mock_stop:
1043        await airplay_player.pause()
1044
1045    send_cmd.assert_called_once_with("ACTION=PAUSE")
1046    mock_stop.assert_not_called()
1047
1048
1049@pytest.mark.asyncio
1050async def test_grouped_leader_pause_parks_session(airplay_player: AirPlayPlayer) -> None:
1051    """A sync leader pauses by parking the session (standby), never sending ACTION=PAUSE."""
1052    airplay_player._attr_group_members = ["test_player", "child"]
1053    send_cmd = _setup_running_stream(airplay_player)
1054    assert airplay_player.stream is not None
1055    session = cast("MagicMock", airplay_player.stream.session)
1056    session.standby = AsyncMock(return_value=True)
1057
1058    with patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as mock_stop:
1059        await airplay_player.pause()
1060
1061    session.standby.assert_awaited_once()
1062    mock_stop.assert_not_called()
1063    send_cmd.assert_not_called()
1064
1065
1066@pytest.mark.asyncio
1067async def test_grouped_pause_falls_back_to_stop(airplay_player: AirPlayPlayer) -> None:
1068    """When a member cannot be parked, grouped pause stops the session."""
1069    airplay_player._attr_group_members = ["test_player", "child"]
1070    send_cmd = _setup_running_stream(airplay_player)
1071    assert airplay_player.stream is not None
1072    session = cast("MagicMock", airplay_player.stream.session)
1073    session.standby = AsyncMock(return_value=False)
1074
1075    with patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as mock_stop:
1076        await airplay_player.pause()
1077
1078    mock_stop.assert_called_once()
1079    send_cmd.assert_not_called()
1080
1081
1082@pytest.mark.asyncio
1083async def test_synced_child_pause_parks_session(airplay_player: AirPlayPlayer) -> None:
1084    """A synced child also pauses by parking the shared session, never ACTION=PAUSE."""
1085    airplay_player._attr_group_members = []
1086    send_cmd = _setup_running_stream(airplay_player)
1087    assert airplay_player.stream is not None
1088    session = cast("MagicMock", airplay_player.stream.session)
1089    session.standby = AsyncMock(return_value=True)
1090
1091    with (
1092        patch.object(AirPlayPlayer, "synced_to", new_callable=PropertyMock, return_value="parent"),
1093        patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as mock_stop,
1094    ):
1095        await airplay_player.pause()
1096
1097    session.standby.assert_awaited_once()
1098    mock_stop.assert_not_called()
1099    send_cmd.assert_not_called()
1100
1101
1102# --- Automatic group re-join after unexpected stream loss ---
1103
1104
1105def _make_idle_player(player_id: str = "test_player") -> AirPlayPlayer:
1106    """Create an idle, ungrouped AirPlayPlayer wired for the re-join tests."""
1107    player = AirPlayPlayer(
1108        provider=MagicMock(),
1109        player_id=player_id,
1110        display_name=f"Player {player_id}",
1111        address="127.0.0.1",
1112        manufacturer="Test Manufacturer",
1113        model="Test Model",
1114        raop_discovery_info=None,
1115        airplay_discovery_info=None,
1116    )
1117    # the synced_to property scans all players of the provider
1118    _players_mock(player).iter_players.return_value = []
1119    player._attr_group_members = []
1120    player._attr_playback_state = PlaybackState.IDLE
1121    player.stream = None
1122    return player
1123
1124
1125def _make_playing_leader(player_id: str = "leader") -> AirPlayPlayer:
1126    """Create an AirPlayPlayer that looks like the playing leader of a live session."""
1127    leader = _make_idle_player(player_id)
1128    leader._attr_playback_state = PlaybackState.PLAYING
1129    stream = MagicMock()
1130    stream.running = True
1131    stream.session = MagicMock(parked=False)
1132    leader.stream = stream
1133    return leader
1134
1135
1136def _players_mock(player: AirPlayPlayer) -> MagicMock:
1137    """Return the mocked players controller of the given player."""
1138    return cast("MagicMock", player.mass.players)
1139
1140
1141def _attach_running_session(player: AirPlayPlayer, sync_clients: list[AirPlayPlayer]) -> None:
1142    """Attach a mock running stream whose session carries the given members."""
1143    stream = MagicMock()
1144    stream.running = True
1145    stream.session = MagicMock(parked=False)
1146    stream.session.sync_clients = sync_clients
1147    player.stream = stream
1148
1149
1150_NO_DELAYS = "music_assistant.providers.airplay.player.AIRPLAY_REJOIN_ATTEMPT_DELAYS"
1151
1152
1153@pytest.mark.asyncio
1154async def test_rejoin_succeeds_on_first_attempt() -> None:
1155    """A re-join attempt joins the player back to the playing leader's session."""
1156    player = _make_idle_player()
1157    leader = _make_playing_leader()
1158    players_mock = _players_mock(player)
1159    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1160
1161    async def cmd_group(player_id: str, target_id: str) -> None:
1162        assert player_id == player.player_id
1163        assert target_id == leader.player_id
1164        _attach_running_session(player, [leader, player])
1165
1166    players_mock.cmd_group = AsyncMock(side_effect=cmd_group)
1167    players_mock.cmd_ungroup = AsyncMock()
1168
1169    with patch(_NO_DELAYS, (0, 0, 0)):
1170        await player._group_rejoin_attempts(["leader"])
1171
1172    assert players_mock.cmd_group.await_count == 1
1173    players_mock.cmd_ungroup.assert_not_awaited()
1174
1175
1176@pytest.mark.asyncio
1177async def test_rejoin_retries_after_failure_then_succeeds() -> None:
1178    """A failed attempt (device still unreachable) is retried with backoff."""
1179    player = _make_idle_player()
1180    leader = _make_playing_leader()
1181    players_mock = _players_mock(player)
1182    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1183    attempts: list[int] = []
1184
1185    async def cmd_group(_player_id: str, _target_id: str) -> None:
1186        attempts.append(1)
1187        if len(attempts) == 1:
1188            raise PlayerCommandFailed("device unreachable")
1189        _attach_running_session(player, [leader, player])
1190
1191    players_mock.cmd_group = AsyncMock(side_effect=cmd_group)
1192
1193    with patch(_NO_DELAYS, (0, 0, 0)):
1194        await player._group_rejoin_attempts(["leader"])
1195
1196    assert len(attempts) == 2
1197
1198
1199@pytest.mark.asyncio
1200async def test_rejoin_gives_up_after_all_attempts() -> None:
1201    """Without a playing group to re-join, the attempts run out and stop cleanly."""
1202    player = _make_idle_player()
1203    players_mock = _players_mock(player)
1204    players_mock.get_player.return_value = None
1205    players_mock.cmd_group = AsyncMock()
1206
1207    with patch(_NO_DELAYS, (0, 0, 0)):
1208        await player._group_rejoin_attempts(["leader"])
1209
1210    players_mock.cmd_group.assert_not_awaited()
1211
1212
1213@pytest.mark.asyncio
1214async def test_rejoin_aborts_when_player_used_meanwhile() -> None:
1215    """A player that was grouped or repurposed meanwhile is left alone."""
1216    player = _make_idle_player()
1217    leader = _make_playing_leader()
1218    players_mock = _players_mock(player)
1219    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1220    players_mock.cmd_group = AsyncMock()
1221    # the user started something else on the player during the backoff
1222    stream = MagicMock()
1223    stream.running = True
1224    player.stream = stream
1225
1226    with patch(_NO_DELAYS, (0, 0, 0)):
1227        await player._group_rejoin_attempts(["leader"])
1228
1229    players_mock.cmd_group.assert_not_awaited()
1230
1231
1232@pytest.mark.asyncio
1233async def test_rejoin_undoes_dangling_membership() -> None:
1234    """A join that yields no running stream is rolled back before the next attempt."""
1235    player = _make_idle_player()
1236    leader = _make_playing_leader()
1237    players_mock = _players_mock(player)
1238    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1239    # cmd_group "succeeds" but the late-join failed internally: no stream appears
1240    players_mock.cmd_group = AsyncMock()
1241    players_mock.cmd_ungroup = AsyncMock()
1242
1243    with patch(_NO_DELAYS, (0, 0, 0)):
1244        await player._group_rejoin_attempts(["leader"])
1245
1246    # every attempt rolled its dangling membership back
1247    assert players_mock.cmd_group.await_count == 3
1248    assert players_mock.cmd_ungroup.await_count == 3
1249
1250
1251@pytest.mark.asyncio
1252async def test_rejoin_cancelled_when_player_unavailable() -> None:
1253    """An offline player abandons the re-join right away instead of attempting."""
1254    player = _make_idle_player()
1255    leader = _make_playing_leader()
1256    players_mock = _players_mock(player)
1257    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1258    players_mock.cmd_group = AsyncMock()
1259    player._attr_available = False
1260
1261    # the later long delays prove the loop returns on the first pass
1262    with patch(_NO_DELAYS, (0, 60, 60)):
1263        await asyncio.wait_for(player._group_rejoin_attempts(["leader"]), timeout=5)
1264
1265    players_mock.cmd_group.assert_not_awaited()
1266    players_mock.get_player.assert_not_called()
1267
1268
1269@pytest.mark.asyncio
1270async def test_rejoin_aborts_when_synced_into_foreign_group() -> None:
1271    """A player the user grouped elsewhere meanwhile is left alone."""
1272    player = _make_idle_player()
1273    leader = _make_playing_leader()
1274    foreign_leader = MagicMock()
1275    foreign_leader.player_id = "other"
1276    foreign_leader.group_members = ["other", player.player_id]
1277    # the player reports it is now synced to a leader outside the original group
1278    _players_mock(player).iter_players.return_value = [foreign_leader]
1279    players_mock = _players_mock(player)
1280    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1281    players_mock.cmd_group = AsyncMock()
1282
1283    with patch(_NO_DELAYS, (0, 60, 60)):
1284        await asyncio.wait_for(player._group_rejoin_attempts(["leader"]), timeout=5)
1285
1286    players_mock.cmd_group.assert_not_awaited()
1287
1288
1289def test_resolve_rejoin_target_finds_promoted_sibling() -> None:
1290    """When the old leader is gone, a promoted (now leading) sibling is the target."""
1291    player = _make_idle_player()
1292    sibling = _make_playing_leader("sibling")
1293    sibling._attr_group_members = ["sibling", "other_member"]
1294    _players_mock(player).get_player.side_effect = lambda player_id: {"sibling": sibling}.get(
1295        player_id
1296    )
1297
1298    assert player._resolve_rejoin_target(["old_leader", "sibling"]) is sibling
1299
1300
1301def test_resolve_rejoin_target_skips_candidate_in_foreign_group() -> None:
1302    """A candidate absorbed into another group is never followed there."""
1303    player = _make_idle_player()
1304    foreign_leader = _make_playing_leader("foreign")
1305    foreign_leader._attr_group_members = ["foreign", "old_leader"]
1306    old_leader = _make_playing_leader("old_leader")
1307    # the old leader reports it is now synced to the foreign leader
1308    _players_mock(old_leader).iter_players.return_value = [foreign_leader]
1309    _players_mock(player).get_player.side_effect = lambda player_id: {
1310        "old_leader": old_leader,
1311        "foreign": foreign_leader,
1312    }.get(player_id)
1313
1314    assert player._resolve_rejoin_target(["old_leader"]) is None
1315
1316
1317@pytest.mark.parametrize(
1318    ("playback_state", "stream_running", "available", "expected"),
1319    [
1320        (PlaybackState.PLAYING, True, True, True),
1321        # a parked (paused) session has no live timeline to late-join
1322        (PlaybackState.PAUSED, True, True, False),
1323        (PlaybackState.IDLE, False, True, False),
1324        # target device itself dropped off the network
1325        (PlaybackState.PLAYING, True, False, False),
1326    ],
1327)
1328def test_resolve_rejoin_target_requires_playing_session(
1329    playback_state: PlaybackState, stream_running: bool, available: bool, expected: bool
1330) -> None:
1331    """Only an available target with an actively playing session is accepted."""
1332    player = _make_idle_player()
1333    leader = _make_playing_leader()
1334    leader._attr_playback_state = playback_state
1335    cast("MagicMock", leader.stream).running = stream_running
1336    leader._attr_available = available
1337    _players_mock(player).get_player.side_effect = lambda player_id: {"leader": leader}.get(
1338        player_id
1339    )
1340
1341    target = player._resolve_rejoin_target(["leader"])
1342    assert (target is leader) is expected
1343
1344
1345@pytest.mark.asyncio
1346async def test_rejoin_heals_session_when_membership_survived() -> None:
1347    """A player still holding sync membership (static group) heals the session only."""
1348    player = _make_idle_player()
1349    leader = _make_playing_leader()
1350    # the sync membership survived the stream loss: the player is still listed
1351    # as a member of (and synced to) the leader
1352    leader._attr_group_members = ["leader", player.player_id]
1353    _players_mock(player).iter_players.return_value = [leader]
1354    players_mock = _players_mock(player)
1355    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1356    players_mock.cmd_group = AsyncMock()
1357    players_mock.cmd_ungroup = AsyncMock()
1358    session = cast("MagicMock", leader.stream).session
1359
1360    async def add_client(joiner: AirPlayPlayer) -> None:
1361        assert joiner is player
1362        _attach_running_session(player, [leader, player])
1363
1364    session.add_client = AsyncMock(side_effect=add_client)
1365
1366    with patch(_NO_DELAYS, (0,)):
1367        await player._group_rejoin_attempts(["leader"])
1368
1369    session.add_client.assert_awaited_once()
1370    players_mock.cmd_group.assert_not_awaited()
1371    players_mock.cmd_ungroup.assert_not_awaited()
1372
1373
1374@pytest.mark.asyncio
1375async def test_rejoin_session_heal_failure_keeps_membership() -> None:
1376    """A failed session heal never touches the (configured) group membership."""
1377    player = _make_idle_player()
1378    leader = _make_playing_leader()
1379    leader._attr_group_members = ["leader", player.player_id]
1380    _players_mock(player).iter_players.return_value = [leader]
1381    players_mock = _players_mock(player)
1382    players_mock.get_player.side_effect = lambda player_id: {"leader": leader}.get(player_id)
1383    players_mock.cmd_group = AsyncMock()
1384    players_mock.cmd_ungroup = AsyncMock()
1385    session = cast("MagicMock", leader.stream).session
1386    # the late-join fails internally: no stream appears on the player
1387    session.add_client = AsyncMock()
1388
1389    with patch(_NO_DELAYS, (0,)):
1390        await player._group_rejoin_attempts(["leader"])
1391
1392    session.add_client.assert_awaited_once()
1393    players_mock.cmd_ungroup.assert_not_awaited()
1394
1395
1396@pytest.mark.asyncio
1397async def test_schedule_and_cancel_group_rejoin() -> None:
1398    """Scheduling replaces a pending task and cancelling stops it."""
1399    player = _make_idle_player()
1400    _players_mock(player).get_player.return_value = None
1401    cast("MagicMock", player.mass).create_task = lambda coro: (
1402        asyncio.get_running_loop().create_task(coro)
1403    )
1404
1405    with patch(_NO_DELAYS, (60, 60, 60)):
1406        player.schedule_group_rejoin(["leader"])
1407        first_task = player._rejoin_task
1408        assert first_task is not None
1409        # a second death replaces the pending schedule
1410        player.schedule_group_rejoin(["leader"])
1411        second_task = player._rejoin_task
1412        assert second_task is not None
1413        assert second_task is not first_task
1414        await asyncio.sleep(0)
1415        assert first_task.cancelled()
1416        # any deliberate use of the player cancels the pending re-join
1417        player.cancel_group_rejoin()
1418        assert player._rejoin_task is None
1419        await asyncio.sleep(0)
1420        assert second_task.cancelled()
1421
1422
1423@pytest.mark.asyncio
1424async def test_stop_cancels_pending_rejoin() -> None:
1425    """An explicit stop command on the player drops the pending re-join."""
1426    player = _make_idle_player()
1427    _players_mock(player).get_player.return_value = None
1428    cast("MagicMock", player.mass).create_task = lambda coro: (
1429        asyncio.get_running_loop().create_task(coro)
1430    )
1431
1432    with (
1433        patch(_NO_DELAYS, (60,)),
1434        patch.object(AirPlayPlayer, "update_state"),
1435    ):
1436        player.schedule_group_rejoin(["leader"])
1437        rejoin_task = player._rejoin_task
1438        assert rejoin_task is not None
1439        await player.stop()
1440        assert player._rejoin_task is None
1441        await asyncio.sleep(0)
1442        assert rejoin_task.cancelled()
1443
1444
1445# --- Group membership and the leader's stream session ---
1446
1447
1448@pytest.mark.asyncio
1449async def test_set_members_adds_the_child_to_the_running_session() -> None:
1450    """A member joining a leader with a live session is added to that session."""
1451    leader = _make_playing_leader()
1452    child = _make_idle_player("child")
1453    _attach_running_session(leader, [leader])
1454    session = cast("MagicMock", leader.stream).session
1455    session.add_client = AsyncMock()
1456    _players_mock(leader).get_player.side_effect = lambda player_id: {"child": child}.get(player_id)
1457
1458    await leader.set_members(player_ids_to_add=["child"])
1459
1460    session.add_client.assert_awaited_once_with(child)
1461    assert leader.group_members == ["leader", "child"]
1462
1463
1464def test_live_session_members_reports_who_the_session_actually_feeds() -> None:
1465    """Group membership outlives the session, so only the session itself can answer."""
1466    leader = _make_playing_leader()
1467    leader._attr_group_members = ["leader", "child"]
1468    # the session dropped the child (e.g. its receiver never answered our clock)
1469    _attach_running_session(leader, [leader])
1470
1471    assert leader.live_session_members == ["leader"]
1472
1473    # no session means nobody is being rendered with, whatever the group says
1474    stream = cast("MagicMock", leader.stream)
1475    stream.running = False
1476    assert leader.live_session_members == []
1477    stream.running = True
1478    stream.session = None
1479    assert leader.live_session_members == []
1480    leader.stream = None
1481    assert leader.live_session_members == []
1482
1483
1484@pytest.mark.asyncio
1485async def test_set_members_warns_when_the_leader_has_no_session(
1486    caplog: pytest.LogCaptureFixture,
1487) -> None:
1488    """A member joining a leader that renders through a protocol gets no audio, loudly."""
1489    leader = _make_idle_player("leader")
1490    leader.logger = logging.getLogger("test.airplay.player")
1491    child = _make_idle_player("child")
1492    _players_mock(leader).get_player.side_effect = lambda player_id: {"child": child}.get(player_id)
1493    # the leader hands its audio to one of its output protocols, so it has no session
1494    leader.set_active_output_protocol("bridge_leader")
1495
1496    with caplog.at_level(logging.WARNING):
1497        await leader.set_members(player_ids_to_add=["child"])
1498
1499    assert leader.group_members == ["leader", "child"]
1500    assert "no stream session to join" in caplog.text
1501
1502
1503def _attach_live_stream(player: AirPlayPlayer, session: AirPlayStreamSession) -> MagicMock:
1504    """Attach a mock stream that is connected and fed by the given session."""
1505    stream = MagicMock()
1506    stream.running = True
1507    stream.connected = True
1508    stream.session = session
1509    stream.send_cli_command = AsyncMock(return_value=True)
1510    stream.stop = AsyncMock()
1511    player.stream = stream
1512    return stream
1513
1514
1515@pytest.mark.asyncio
1516async def test_play_after_ungrouping_a_parked_group_resumes_via_the_queue() -> None:
1517    """
1518    Breaking a parked group up leaves the remaining player alone with the park.
1519
1520    Its binary is held at standby with nothing being fed, so the resume still has
1521    to re-anchor through the queue: ACTION=PLAY carries no anchor and would
1522    report playback over silence.
1523    """
1524    leader = _make_idle_player("leader")
1525    child = _make_idle_player("child")
1526    session = AirPlayStreamSession(
1527        MagicMock(mass=leader.mass), [leader, child], AIRPLAY_PCM_FORMAT, MagicMock()
1528    )
1529    leader_stream = _attach_live_stream(leader, session)
1530    child_stream = _attach_live_stream(child, session)
1531    leader._attr_group_members = ["leader", "child"]
1532    players = _players_mock(leader)
1533    players.get_player.side_effect = lambda player_id: {"leader": leader, "child": child}.get(
1534        player_id
1535    )
1536    players.get_active_queue.return_value = MagicMock(queue_id="leader")
1537    resume_queue = AsyncMock()
1538    cast("MagicMock", leader.mass).player_queues.resume = resume_queue
1539
1540    await leader.pause()
1541    await leader.set_members(player_ids_to_remove=["child"])
1542    leader_stream.send_cli_command.reset_mock()
1543    await leader.play()
1544
1545    # the removal stops only the child; the leader keeps its parked session
1546    child_stream.stop.assert_awaited_once()
1547    leader_stream.stop.assert_not_awaited()
1548    assert leader.group_members == []
1549    assert session.sync_clients == [leader]
1550    assert session.parked is True
1551    resume_queue.assert_awaited_once_with("leader", fade_in=False)
1552    leader_stream.send_cli_command.assert_not_awaited()
1553
1554
1555@pytest.mark.asyncio
1556async def test_leader_stepping_out_alone_keeps_the_session_for_the_others() -> None:
1557    """
1558    A leader that only removes itself hands the live session to the members left behind.
1559
1560    The leader is not asked to take the others with it, so tearing the session
1561    down here would cut off members that are still supposed to be playing.
1562    """
1563    leader = _make_idle_player("leader")
1564    child = _make_idle_player("child")
1565    session = AirPlayStreamSession(
1566        MagicMock(mass=leader.mass), [leader, child], AIRPLAY_PCM_FORMAT, MagicMock()
1567    )
1568    leader_stream = _attach_live_stream(leader, session)
1569    child_stream = _attach_live_stream(child, session)
1570    leader._attr_group_members = ["leader", "child"]
1571    leader._attr_playback_state = PlaybackState.PLAYING
1572    child._attr_playback_state = PlaybackState.PLAYING
1573    lookup = {"leader": leader, "child": child}
1574    for player in (leader, child):
1575        players = _players_mock(player)
1576        players.get_player.side_effect = lookup.get
1577        players.iter_players.return_value = [leader, child]
1578
1579    await leader.set_members(player_ids_to_remove=["leader"])
1580
1581    leader_stream.stop.assert_awaited_once()
1582    child_stream.stop.assert_not_awaited()
1583    assert session.sync_clients == [child]
1584    assert leader.group_members == []
1585    # nothing claims the remaining member anymore: its caller picks the new leader
1586    assert child.synced_to is None
1587
1588
1589@pytest.mark.asyncio
1590async def test_ungroup_on_a_sync_leader_dissolves_the_whole_group() -> None:
1591    """
1592    Ungrouping a sync leader must release its members, not just the leader itself.
1593
1594    A leader lists itself in group_members, so the default ungroup asks to remove
1595    the leader AND every member in one call.
1596    """
1597    leader = _make_idle_player("leader")
1598    child = _make_idle_player("child")
1599    session = AirPlayStreamSession(
1600        MagicMock(mass=leader.mass), [leader, child], AIRPLAY_PCM_FORMAT, MagicMock()
1601    )
1602    leader_stream = _attach_live_stream(leader, session)
1603    child_stream = _attach_live_stream(child, session)
1604    leader._attr_group_members = ["leader", "child"]
1605    leader._attr_playback_state = PlaybackState.PLAYING
1606    child._attr_playback_state = PlaybackState.PLAYING
1607    lookup = {"leader": leader, "child": child}
1608    for player in (leader, child):
1609        players = _players_mock(player)
1610        players.get_player.side_effect = lookup.get
1611        players.iter_players.return_value = [leader, child]
1612
1613    await leader.ungroup()
1614
1615    assert leader.group_members == []
1616    leader_stream.stop.assert_awaited_once()
1617    child_stream.stop.assert_awaited_once()
1618    assert session.sync_clients == []
1619
1620
1621# --- Device password ---
1622
1623
1624def _set_password_discovery(
1625    player: AirPlayPlayer,
1626    *,
1627    flags: str = "0x0",
1628    pw: str = "",
1629    password: str | None = None,
1630    paired: bool = False,
1631) -> None:
1632    """
1633    Attach an AirPlay 2 + RAOP device announcing password protection.
1634
1635    :param flags: The _airplay service sf/flags bitmask (0x80 marks a password).
1636    :param pw: The legacy ``pw`` boolean published by the _raop service.
1637    :param password: The device password stored in the player config, if any.
1638    :param paired: Whether AirPlay 2 pairing credentials are stored for the device.
1639    """
1640    airplay_info = MagicMock()
1641    airplay_info.decoded_properties = {"features": AP2_FEATURES, "flags": flags}
1642    airplay_info.properties = {b"flags": flags.encode()}
1643    player.airplay_discovery_info = airplay_info
1644    raop_info = MagicMock()
1645    raop_info.decoded_properties = {"pw": pw} if pw else {}
1646    raop_info.properties = {}
1647    player.raop_discovery_info = raop_info
1648    _configure_player(player, {CONF_PASSWORD: password})
1649    credentials = {CONF_AIRPLAY_CREDENTIALS: "a" * 192} if paired else {}
1650    player.get_setup_value = (  # type: ignore[method-assign]
1651        lambda key, default=None: credentials.get(key, default)
1652    )
1653
1654
1655@pytest.mark.asyncio
1656async def test_password_entry_is_never_offered_in_the_settings(
1657    airplay_player: AirPlayPlayer,
1658) -> None:
1659    """The password is storage only: the setup flow is the sole way to enter it."""
1660    _set_password_discovery(airplay_player, flags="0x80")
1661    assert airplay_player.password_required is True
1662
1663    entries = await airplay_player.get_config_entries()
1664    entry = next(entry for entry in entries if entry.key == CONF_PASSWORD)
1665    assert entry.hidden is True
1666
1667
1668@pytest.mark.parametrize(
1669    ("flags", "pw"),
1670    [
1671        # AirPlay 2 announces password protection through the flags bit...
1672        ("0x80", ""),
1673        # ...a legacy RAOP receiver through the classic pw boolean
1674        ("0x0", "true"),
1675    ],
1676)
1677def test_announced_password_without_one_stored_needs_setup(
1678    airplay_player: AirPlayPlayer, flags: str, pw: str
1679) -> None:
1680    """A device that asks for a password it never got must be set up first."""
1681    _set_password_discovery(airplay_player, flags=flags, pw=pw)
1682
1683    assert airplay_player.password_required is True
1684    assert airplay_player.needs_setup is True
1685    assert airplay_player.setup_reason == "password_required"
1686
1687
1688def test_apple_tv_without_a_password_does_not_need_setup(airplay_player: AirPlayPlayer) -> None:
1689    """An Apple TV that announces no password must not be sent into setup."""
1690    # flags as published by tvOS with "Require Password" off
1691    _set_password_discovery(airplay_player, flags="0x644", paired=True)
1692    airplay_player.device_info.manufacturer = "Apple"
1693    airplay_player.device_info.model = "Apple TV 4K Gen2"
1694
1695    assert airplay_player.password_required is False
1696    assert airplay_player.needs_setup is False
1697
1698
1699def test_apple_tv_with_a_password_set_needs_setup(airplay_player: AirPlayPlayer) -> None:
1700    """An Apple TV announces its password through the same bit as every other receiver."""
1701    # the same device with "Require Password" on: the password bit replaces the pairing bit
1702    _set_password_discovery(airplay_player, flags="0x4c4")
1703    airplay_player.device_info.manufacturer = "Apple"
1704    airplay_player.device_info.model = "Apple TV 4K Gen2"
1705
1706    assert airplay_player.password_required is True
1707    assert airplay_player.needs_setup is True
1708    assert airplay_player.setup_reason == "password_required"
1709
1710
1711def test_silent_primary_bit_is_not_a_password_announcement(airplay_player: AirPlayPlayer) -> None:
1712    """The SilentPrimary flags bit says nothing about a password and must not force setup."""
1713    _set_password_discovery(airplay_player, flags="0x1644", paired=True)
1714
1715    assert airplay_player.password_required is False
1716    assert airplay_player.needs_setup is False
1717
1718
1719@pytest.mark.parametrize(
1720    ("flags", "pw", "paired"),
1721    [
1722        # AirPlay 2 collects the password as part of pairing, so it ends up with both
1723        ("0x80", "", True),
1724        # a legacy RAOP receiver has no pairing at all: the password is enough
1725        ("0x0", "true", False),
1726    ],
1727)
1728def test_stored_password_clears_the_setup_requirement(
1729    airplay_player: AirPlayPlayer, flags: str, pw: str, paired: bool
1730) -> None:
1731    """Once the password is stored the player is ready to use again."""
1732    _set_password_discovery(airplay_player, flags=flags, pw=pw, password="hunter2", paired=paired)
1733
1734    assert airplay_player.needs_setup is False
1735    assert airplay_player.setup_reason is None
1736
1737
1738def test_rejected_password_marker_forces_setup(airplay_player: AirPlayPlayer) -> None:
1739    """A password the device rejected sends an otherwise ready player back into setup."""
1740    # the migration case: a paired device that gained password protection later
1741    _set_password_discovery(airplay_player, flags="0x80", password="wrong", paired=True)
1742    ready_before = airplay_player.needs_setup
1743
1744    airplay_player.set_password_invalid(True)
1745
1746    assert ready_before is False
1747    assert airplay_player.password_invalid is True
1748    assert airplay_player.needs_setup is True
1749    assert airplay_player.setup_reason == "password_required"
1750
1751
1752def test_rejected_password_marker_survives_a_restart(airplay_player: AirPlayPlayer) -> None:
1753    """The marker is persisted as a raw player config value, not just in memory."""
1754    _set_password_discovery(airplay_player, flags="0x80", password="wrong", paired=True)
1755
1756    airplay_player.set_password_invalid(True)
1757
1758    airplay_player.mass.config.set_raw_player_config_value.assert_called_once_with(  # type: ignore[attr-defined]
1759        "test_player", CONF_PASSWORD_INVALID, True
1760    )
1761
1762
1763def test_clearing_the_marker_only_writes_when_it_was_set(airplay_player: AirPlayPlayer) -> None:
1764    """Every successful connect clears the marker, but must not write the config."""
1765    _set_password_discovery(airplay_player, flags="0x80", password="hunter2", paired=True)
1766    set_raw = airplay_player.mass.config.set_raw_player_config_value
1767
1768    airplay_player.set_password_invalid(False)
1769    set_raw.assert_not_called()  # type: ignore[attr-defined]
1770
1771    airplay_player.set_password_invalid(True)
1772    assert airplay_player.needs_setup is True
1773    airplay_player.set_password_invalid(False)
1774
1775    assert airplay_player.password_invalid is False
1776    assert airplay_player.needs_setup is False
1777
1778
1779def test_rejected_password_is_published_to_clients(airplay_player: AirPlayPlayer) -> None:
1780    """The new setup requirement must reach the wire state, not just the property."""
1781    _set_password_discovery(airplay_player, flags="0x80", password="wrong", paired=True)
1782    airplay_player.update_state()
1783    before = airplay_player.state
1784    assert before.needs_setup is False
1785    assert before.available is True
1786
1787    airplay_player.set_password_invalid(True)
1788
1789    # needs_setup/setup_reason are part of the player's own state inputs, so the
1790    # update is neither short-circuited nor left to the next unrelated update
1791    after = airplay_player.state
1792    assert after.needs_setup is True
1793    assert after.setup_reason == "password_required"
1794    assert after.available is False
1795    changed = airplay_player.mass.players.signal_player_state_update.call_args[0][1]  # type: ignore[attr-defined]
1796    assert "needs_setup" in changed
1797
1798
1799def test_pin_pairing_keeps_its_own_setup_reason(airplay_player: AirPlayPlayer) -> None:
1800    """A device that only needs PIN pairing is not reported as a password problem."""
1801    airplay_info = MagicMock()
1802    airplay_info.decoded_properties = {"features": AP2_FEATURES}
1803    airplay_info.properties = {b"flags": b"0x8"}
1804    airplay_player.airplay_discovery_info = airplay_info
1805    airplay_player.get_setup_value = lambda key, default=None: default  # type: ignore[method-assign]  # noqa: ARG005
1806
1807    assert airplay_player.needs_setup is True
1808    assert airplay_player.setup_reason == "pairing_required"
1809