music-assistant-server

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