music-assistant-server

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