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