music-assistant-server

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