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