music-assistant-server

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