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