music-assistant-server

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