music-assistant-server

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