music-assistant-server

1"""Tests for control-capable players in the AirPlay provider."""
2
3from __future__ import annotations
4
5import logging
6from typing import cast
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from music_assistant_models.enums import (
11    ImageType,
12    MediaType,
13    PlaybackState,
14    PlayerFeature,
15    PlayerType,
16)
17from music_assistant_models.errors import MediaNotFoundError, PlayerCommandFailed
18from music_assistant_models.media_items import MediaItemImage
19from pyatv import exceptions as pyatv_exceptions
20from pyatv.const import (
21    DeviceState,
22    FeatureName,
23    PairingRequirement,
24    PowerState,
25    Protocol,
26)
27from pyatv.const import MediaType as PyatvMediaType
28from pyatv.interface import App, AppleTV, ArtworkInfo, Playing
29from pyatv.settings import MrpTunnel
30from zeroconf import ServiceStateChange
31
32from music_assistant.models.player import PlayerMedia
33from music_assistant.providers.airplay.constants import (
34    AIRPLAY_DISCOVERY_TYPE,
35    COMPANION_DISCOVERY_TYPE,
36    CONF_COMPANION_CREDENTIALS,
37    CONF_MRP_CREDENTIALS,
38    CONF_NATIVE_MRP_CREDENTIALS,
39    EXTERNAL_ARTWORK_PATH_PREFIX,
40    MRP_DISCOVERY_TYPE,
41)
42from music_assistant.providers.airplay.control_player import AirPlayControlPlayer
43from music_assistant.providers.airplay.player import AirPlayPlayer, GenericAirPlayPlayer
44from music_assistant.providers.airplay.provider import AirPlayProvider
45
46PLAYER_ID = "apaabbccddeeff"
47DEVICE_ID = "AA:BB:CC:DD:EE:FF"
48DACP_ID = "0123456789ABCDEF"
49AP2_FEATURES = "0x4A7FDFD5,0x3C177FDE"
50
51
52def _service_info(
53    service_type: str,
54    *,
55    properties: dict[str, str] | None = None,
56    address: str = "192.168.1.10",
57) -> MagicMock:
58    """Create an mDNS service-info mock."""
59    info = MagicMock()
60    info.type = service_type
61    info.name = f"test.{service_type}"
62    info.port = 7000 if service_type == AIRPLAY_DISCOVERY_TYPE else 49152
63    info.decoded_properties = properties or {}
64    info.properties = {
65        key.encode(): value.encode() for key, value in info.decoded_properties.items()
66    }
67    info.addresses = [b"\xc0\xa8\x01\x0a"]
68    info.parsed_addresses.return_value = [address]
69    return info
70
71
72def _make_control_player(
73    *,
74    model: str = "Apple TV 4K",
75    companion_flags: str | None = "0x367A2",
76    config_values: dict[str, object] | None = None,
77    setup_data: dict[str, object] | None = None,
78) -> AirPlayControlPlayer:
79    """Create a control-capable AirPlay player with mocked provider state."""
80    provider = MagicMock()
81    provider.instance_id = "airplay"
82    provider.dacp_id = DACP_ID
83    provider.logger = logging.getLogger("test.airplay.apple")
84    config = MagicMock()
85    values = config_values or {}
86    config.get_value.side_effect = lambda key, default=None: values.get(key, default)
87    config.update.side_effect = values.update
88    provider.mass.config.get_base_player_config.return_value = config
89    provider.mass.config.save_player_config = AsyncMock()
90    # Credentials live in the player's (encrypted) setup_data; serve them through the
91    # mass.config get/set surface that Player.get_setup_value / _update_setup_data use.
92    # The same dict object is returned so tests can observe cleared/updated values.
93    stored_setup_data = setup_data if setup_data is not None else {}
94
95    def _config_get(key: str, default: object = None) -> object:
96        if key == f"players/{PLAYER_ID}/setup_data":
97            return stored_setup_data
98        if key == f"players/{PLAYER_ID}":
99            return {"player_id": PLAYER_ID}
100        return default
101
102    def _config_set(key: str, value: object, **_kwargs: object) -> None:
103        prefix = f"players/{PLAYER_ID}/setup_data/"
104        if key.startswith(prefix):
105            stored_setup_data[key[len(prefix) :]] = value
106
107    provider.mass.config.get.side_effect = _config_get
108    provider.mass.config.set.side_effect = _config_set
109    provider.mass.config.decrypt_string.side_effect = lambda value: value
110    provider.mass.config.encrypt_string.side_effect = lambda value: value
111    # raw player config values live in their own store (individual tests override
112    # this); without it they would read back as (truthy) mocks
113    provider.mass.config.get_raw_player_config_value.side_effect = (
114        lambda _player_id, _key, default=None: default
115    )
116    # Real Companion services advertise the flags under the mixed-case key
117    # "rpFl"; zeroconf preserves TXT key casing as sent on the wire.
118    companion_info = (
119        _service_info(
120            COMPANION_DISCOVERY_TYPE,
121            properties={"rpFl": companion_flags},
122        )
123        if companion_flags is not None
124        else None
125    )
126    return AirPlayControlPlayer(
127        provider=provider,
128        player_id=PLAYER_ID,
129        raop_discovery_info=None,
130        airplay_discovery_info=_service_info(
131            AIRPLAY_DISCOVERY_TYPE,
132            properties={
133                "deviceid": DEVICE_ID,
134                "features": AP2_FEATURES,
135                "model": "AppleTV11,1" if "TV" in model else "AudioAccessory5,1",
136                "osvers": "26.0",
137            },
138        ),
139        companion_discovery_info=companion_info,
140        mrp_discovery_info=None,
141        address="192.168.1.10",
142        display_name="Test Apple Device",
143        manufacturer="Apple",
144        model=model,
145        initial_volume=25,
146    )
147
148
149def test_player_models_have_distinct_types() -> None:
150    """Generic and control-capable AirPlay endpoints use distinct player types."""
151    provider = MagicMock()
152    generic = GenericAirPlayPlayer(
153        provider=provider,
154        player_id="generic",
155        raop_discovery_info=None,
156        airplay_discovery_info=None,
157        address="192.168.1.20",
158        display_name="Generic",
159        manufacturer="Receiver",
160        model="Receiver",
161    )
162    controlled = _make_control_player()
163
164    assert generic.type == PlayerType.PROTOCOL
165    assert controlled.type == PlayerType.PLAYER
166
167
168@pytest.mark.parametrize(
169    ("flags", "supported"),
170    [
171        ("0x367A2", True),
172        ("0x367A6", False),
173        ("0x62792", False),
174        ("invalid", False),
175    ],
176)
177def test_companion_pairing_follows_advertised_flags(flags: str, supported: bool) -> None:
178    """Only a Companion service advertising PIN pairing is offered for setup."""
179    player = _make_control_player(companion_flags=flags)
180
181    assert player.companion_pairing_supported is supported
182    # Pairability alone never advertises POWER: without stored credentials or a
183    # connected control channel the command could not be served.
184    assert PlayerFeature.POWER not in player.supported_features
185    assert PlayerFeature.NEXT_PREVIOUS not in player.supported_features
186
187
188def test_playback_state_churn_does_not_force_control_reconnect() -> None:
189    """A receiver toggling volatile TXT state must not change the reconnect signature."""
190    idle = _service_info(
191        AIRPLAY_DISCOVERY_TYPE,
192        properties={"deviceid": DEVICE_ID, "pk": "device-key", "flags": "0x4"},
193    )
194    # Same device while receiving a stream: only the session `flags` bit differs.
195    streaming = _service_info(
196        AIRPLAY_DISCOVERY_TYPE,
197        properties={"deviceid": DEVICE_ID, "pk": "device-key", "flags": "0x404"},
198    )
199    assert AirPlayControlPlayer._service_signature(idle) == AirPlayControlPlayer._service_signature(
200        streaming
201    )
202
203    # TXT keys are case-insensitive (RFC 6763): re-casing keys is not a change.
204    recased = _service_info(
205        AIRPLAY_DISCOVERY_TYPE,
206        properties={"DeviceID": DEVICE_ID, "PK": "device-key", "Flags": "0x4"},
207    )
208    assert AirPlayControlPlayer._service_signature(idle) == AirPlayControlPlayer._service_signature(
209        recased
210    )
211
212    # A change to a connection-relevant field still forces a reconnect.
213    rekeyed = _service_info(
214        AIRPLAY_DISCOVERY_TYPE,
215        properties={"deviceid": DEVICE_ID, "pk": "rotated-key", "flags": "0x4"},
216    )
217    assert AirPlayControlPlayer._service_signature(idle) != AirPlayControlPlayer._service_signature(
218        rekeyed
219    )
220
221
222def test_power_feature_requires_credentials_or_connection() -> None:
223    """POWER is advertised for stored Companion credentials or a live channel."""
224    paired = _make_control_player(setup_data={CONF_COMPANION_CREDENTIALS: "companion-creds"})
225    assert PlayerFeature.POWER in paired.supported_features
226
227    connected = _make_control_player()
228    device = MagicMock(spec=AppleTV)
229    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.TurnOn
230    connected._companion_device = device
231    assert PlayerFeature.POWER in connected.supported_features
232
233
234def test_power_feature_skips_transient_mrp_tunnels() -> None:
235    """A transient MRP tunnel does not count as real power control."""
236    # pyatv reports TurnOn/TurnOff as available on every MRP connection, but a HomePod
237    # reached over the transient tunnel cannot act on them and derives its power state
238    # from logicalDeviceCount, which does not count AirPlay audio sessions.
239    homepod = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
240    assert homepod._uses_transient_mrp is True
241
242    device = MagicMock(spec=AppleTV)
243    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.TurnOn
244    homepod._mrp_device = device
245
246    assert PlayerFeature.POWER not in homepod.supported_features
247
248
249async def test_power_command_refuses_transient_mrp_tunnels() -> None:
250    """A power command never reaches a tunnel that cannot serve it."""
251    homepod = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
252    device = MagicMock(spec=AppleTV)
253    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.TurnOn
254    homepod._mrp_device = device
255
256    with pytest.raises(PlayerCommandFailed):
257        await homepod.power(True)
258
259    device.power.turn_on.assert_not_called()
260
261
262def test_native_transport_features_follow_live_capabilities() -> None:
263    """External next/previous controls are advertised only while available."""
264    player = _make_control_player()
265    device = MagicMock(spec=AppleTV)
266    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.Next
267
268    player._companion_device = device
269
270    assert PlayerFeature.NEXT_PREVIOUS in player.supported_features
271
272
273async def test_pairing_moved_out_of_config_entries() -> None:
274    """Pairing and credentials are handled by the setup flow, not by config entries."""
275    # even with stored credentials, no pairing action/credential entry is emitted:
276    # the interactive setup flow owns pairing and stores creds in setup_data.
277    player = _make_control_player(
278        setup_data={
279            CONF_COMPANION_CREDENTIALS: "companion-creds",
280            CONF_MRP_CREDENTIALS: "mrp-creds",
281            CONF_NATIVE_MRP_CREDENTIALS: "native-creds",
282        }
283    )
284
285    entries = await player.get_config_entries()
286    keys = {entry.key for entry in entries}
287
288    assert CONF_COMPANION_CREDENTIALS not in keys
289    assert CONF_MRP_CREDENTIALS not in keys
290    assert CONF_NATIVE_MRP_CREDENTIALS not in keys
291    assert not any("pairing" in key for key in keys)
292    # the run_setup_flow override is what drives the pairing now
293    assert type(player).run_setup_flow is not AirPlayPlayer.run_setup_flow
294
295
296def test_mute_feature_follows_available_control_path() -> None:
297    """Mute is available only for an active stream or native absolute volume."""
298    player = _make_control_player()
299    assert PlayerFeature.VOLUME_MUTE not in player.supported_features
300
301    player.stream = MagicMock(running=True)
302    assert PlayerFeature.VOLUME_MUTE in player.supported_features
303
304    player.stream = None
305    device = MagicMock(spec=AppleTV)
306    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.SetVolume
307    player._mrp_device = device
308    assert PlayerFeature.VOLUME_MUTE in player.supported_features
309
310
311async def test_native_mute_zeros_and_restores_volume() -> None:
312    """Native mute emulation restores the previous absolute volume."""
313    player = _make_control_player()
314    player._attr_volume_level = 42
315    device = MagicMock(spec=AppleTV)
316    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.SetVolume
317    device.audio.set_volume = AsyncMock()
318    player._mrp_device = device
319
320    await player.volume_mute(True)
321    mute_state_after_mute = player.volume_muted
322    assert mute_state_after_mute is True
323    assert player.volume_level == 42
324    device.audio.set_volume.assert_awaited_once_with(0)
325
326    device.audio.set_volume.reset_mock()
327    await player.volume_mute(False)
328    mute_state_after_unmute = player.volume_muted
329    assert mute_state_after_unmute is False
330    device.audio.set_volume.assert_awaited_once_with(42)
331
332
333def test_duplicate_native_volume_update_is_ignored() -> None:
334    """Repeated native volume events do not rewrite config or player state."""
335    player = _make_control_player()
336    player._attr_volume_level = 42
337    player._attr_volume_muted = False
338
339    with patch.object(player, "update_state") as update_state:
340        player._handle_volume_update("companion", 42)
341
342    cast("MagicMock", player.mass.config).set_raw_player_config_value.assert_not_called()
343    update_state.assert_not_called()
344
345
346def test_native_volume_update_ignored_while_streaming() -> None:
347    """Native volume reports never overwrite the volume of a running stream."""
348    # volume_set routes to the stream while streaming and leaves the native device
349    # volume alone, so a report about that separate knob must not clobber (or persist)
350    # the level the user just set.
351    player = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
352    player.stream = MagicMock(running=True)
353    player._attr_volume_level = 62
354    player._attr_volume_muted = False
355
356    with patch.object(player, "update_state") as update_state:
357        player._handle_volume_update("mrp", 10)
358
359    assert player.volume_level == 62
360    assert player.volume_muted is False
361    cast("MagicMock", player.mass.config).set_raw_player_config_value.assert_not_called()
362    update_state.assert_not_called()
363
364
365def test_transient_mrp_volume_report_never_latches_mute() -> None:
366    """A transient MRP tunnel never mutes the player or rewrites its volume."""
367    # A muted player skips the stream volume command, so a phantom mute learned while
368    # idle would silently stop every later volume change from reaching the device.
369    player = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
370    assert player._uses_transient_mrp is True
371    player.stream = MagicMock(running=False)
372    player._attr_volume_level = 62
373    player._attr_volume_muted = False
374
375    with patch.object(player, "update_state") as update_state:
376        player._handle_volume_update("mrp", 0)
377
378    assert player.volume_muted is False
379    assert player.volume_level == 62
380    cast("MagicMock", player.mass.config).set_raw_player_config_value.assert_not_called()
381    update_state.assert_not_called()
382
383
384def test_native_volume_update_applies_when_idle() -> None:
385    """Native volume reports still drive player state while no stream is running."""
386    player = _make_control_player()
387    player.stream = MagicMock(running=False)
388    player._attr_volume_level = 62
389    player._attr_volume_muted = False
390
391    with patch.object(player, "update_state") as update_state:
392        player._handle_volume_update("companion", 10)
393
394    assert player.volume_level == 10
395    update_state.assert_called_once()
396
397
398def test_control_pairing_is_optional_and_never_requires_setup() -> None:
399    """Optional control pairing never marks the player as requiring setup."""
400    # needs_setup makes a player unavailable for playback, so it must only
401    # reflect the streaming requirements: an Apple TV streams fine without
402    # Companion/MRP credentials and pairing is offered in its settings instead.
403    player = _make_control_player()
404    assert player.companion_pairing_supported is True
405    assert player.mrp_pairing_supported is True
406    assert player.needs_setup is False
407
408    homepod = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
409    assert homepod.companion_pairing_supported is False
410    assert homepod.mrp_pairing_supported is False
411    assert homepod.needs_setup is False
412
413
414async def test_unpaired_apple_tv_never_attempts_control_connection() -> None:
415    """An Apple TV without stored credentials is never connected (or paired)."""
416    # An unsolicited connect would run pyatv's transient pair-setup, which makes
417    # an Apple TV display the AirPlay pairing dialog. Regression test for the
418    # spontaneous pairing popups: no credentials means no connection attempt,
419    # also while the Companion record is still undiscovered.
420    for companion_flags in ("0x367A2", None):
421        player = _make_control_player(companion_flags=companion_flags)
422        with patch(
423            "music_assistant.providers.airplay.control_player.pyatv.connect",
424        ) as connect:
425            assert await player._connect_companion() is False
426            assert await player._connect_mrp() is False
427        connect.assert_not_awaited()
428
429    # Apple TV detection reads the TXT model key case-insensitively as well.
430    player = _make_control_player(companion_flags=None)
431    assert player.airplay_discovery_info is not None
432    properties = player.airplay_discovery_info.decoded_properties
433    properties["Model"] = properties.pop("model")
434    with patch(
435        "music_assistant.providers.airplay.control_player.pyatv.connect",
436    ) as connect:
437        assert await player._connect_mrp() is False
438    connect.assert_not_awaited()
439
440
441async def test_play_media_wakes_device_before_starting_stream() -> None:
442    """A sleeping controlled device receives wake before the AirPlay stream starts."""
443    player = _make_control_player()
444    player._attr_powered = False
445    device = MagicMock(spec=AppleTV)
446    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.TurnOn
447    events: list[str] = []
448
449    async def _turn_on() -> None:
450        events.append("wake")
451        player._handle_power_update("companion", PowerState.On)
452
453    async def _play_media(_player: AirPlayPlayer, _media: PlayerMedia) -> None:
454        events.append("stream")
455
456    device.power.turn_on = AsyncMock(side_effect=_turn_on)
457    player._companion_device = device
458
459    with patch.object(AirPlayPlayer, "play_media", side_effect=_play_media, autospec=True):
460        await player.play_media(MagicMock(spec=PlayerMedia))
461
462    assert events == ["wake", "stream"]
463
464
465async def test_stop_routes_stale_stream_to_external_playback() -> None:
466    """A stopped AirPlay stream does not block native playback control."""
467    player = _make_control_player()
468    player.stream = MagicMock(running=False)
469    device = MagicMock(spec=AppleTV)
470    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.Stop
471    device.remote_control.stop = AsyncMock()
472    player._mrp_device = device
473
474    with patch.object(AirPlayPlayer, "stop", new=AsyncMock()) as stop_stream:
475        await player.stop()
476
477    stop_stream.assert_not_awaited()
478    device.remote_control.stop.assert_awaited_once()
479
480
481def test_power_off_update_tracks_sleep_state() -> None:
482    """A Companion power-off update clears stale playback state."""
483    player = _make_control_player()
484    player.stream = MagicMock(running=False)
485    player._attr_playback_state = PlaybackState.PLAYING
486    player._attr_active_source = "external"
487    player._attr_current_media = MagicMock()
488
489    with patch.object(player, "update_state") as update_state:
490        player._handle_power_update("companion", PowerState.Off)
491        assert player.powered is False
492        assert player.playback_state == PlaybackState.IDLE
493        assert player.active_source is None
494        assert player.current_media is None
495
496        update_state.assert_called_once()
497
498
499def test_power_off_update_ignored_while_streaming() -> None:
500    """A device streaming from Music Assistant is never marked powered off."""
501    # MRP derives power from logicalDeviceCount, which does not count AirPlay audio
502    # sessions, so a playing HomePod reports PowerState.Off. Acting on that would
503    # strand the player as "off" and trip the auto-ungroup in the players controller.
504    player = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
505    player.stream = MagicMock(running=True)
506    player._attr_powered = True
507    player._power_on_event.set()
508    player._attr_playback_state = PlaybackState.PLAYING
509    player._attr_active_source = "airplay"
510
511    with patch.object(player, "update_state") as update_state:
512        player._handle_power_update("mrp", PowerState.Off)
513
514    assert player.powered is True
515    assert player._power_on_event.is_set()
516    assert player.playback_state == PlaybackState.PLAYING
517    assert player.active_source == "airplay"
518    update_state.assert_not_called()
519
520
521def test_power_on_update_tracks_awake_state() -> None:
522    """A Companion power-on update confirms that the device is awake."""
523    player = _make_control_player()
524
525    with patch.object(player, "update_state") as update_state:
526        player._handle_power_update("companion", PowerState.On)
527
528    assert player.powered is True
529    assert player._power_on_event.is_set()
530    update_state.assert_called_once()
531
532
533def test_mrp_updates_external_source_and_media() -> None:
534    """MRP now-playing pushes expose the active app and external media."""
535    player = _make_control_player()
536    device = MagicMock(spec=AppleTV)
537    device.metadata.app = App("Music", "com.apple.Music")
538    player._mrp_device = device
539    playing = Playing(
540        media_type=PyatvMediaType.Music,
541        device_state=DeviceState.Playing,
542        title="External track",
543        artist="Artist",
544        album="Album",
545        total_time=180,
546        position=12,
547        content_identifier="external-track",
548    )
549
550    with patch.object(player, "update_state"):
551        player._handle_playing_update(playing)
552
553    assert player.playback_state == PlaybackState.PLAYING
554    assert player.active_source == "com.apple.Music"
555    assert player.current_media is not None
556    assert player.current_media.media_type == MediaType.TRACK
557    assert player.current_media.title == "External track"
558    assert player.current_media.elapsed_time == 12
559    assert player.source_list[0].name == "Music"
560
561
562def test_mrp_update_publishes_external_artwork() -> None:
563    """MRP artwork is exposed as a cached Music Assistant image-proxy URL."""
564    player = _make_control_player()
565    device = MagicMock(spec=AppleTV)
566    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.Artwork
567    device.metadata.app = App("NLZIET", "nl.nlziet")
568    artwork_id = "https://artwork.example/{w}x{h}/cover"
569    device.metadata.artwork_id = artwork_id
570    player._mrp_device = device
571
572    with (
573        patch.object(player, "update_state"),
574        patch.object(
575            player.mass.metadata,
576            "get_image_url",
577            return_value="http://mass/imageproxy/artwork",
578        ) as get_image_url,
579    ):
580        player._handle_playing_update(
581            Playing(device_state=DeviceState.Playing, title="De Eindmusical")
582        )
583
584    assert player.current_media is not None
585    assert player.current_media.image_url == "http://mass/imageproxy/artwork"
586    image = get_image_url.call_args.args[0]
587    assert image == MediaItemImage(
588        type=ImageType.THUMB,
589        path=f"{EXTERNAL_ARTWORK_PATH_PREFIX}/{PLAYER_ID}/{artwork_id}",
590        provider="airplay",
591        remotely_accessible=False,
592    )
593
594
595def test_mrp_update_refreshes_existing_app_name() -> None:
596    """A late app name replaces the bundle id on an existing source."""
597    player = _make_control_player()
598    device = MagicMock(spec=AppleTV)
599    device.features.in_state.return_value = False
600    device.metadata.app = App(None, "nl.nlziet")
601    player._mrp_device = device
602    playing = Playing(device_state=DeviceState.Playing, title="De Eindmusical")
603
604    with patch.object(player, "update_state"):
605        player._handle_playing_update(playing)
606        device.metadata.app = App("NLZIET", "nl.nlziet")
607        player._handle_playing_update(playing)
608
609    assert len(player.source_list) == 1
610    assert player.source_list[0].name == "NLZIET"
611
612
613async def test_airplay_provider_resolves_only_current_external_artwork() -> None:
614    """The image proxy can retrieve current MRP artwork but not stale artwork."""
615    player = _make_control_player()
616    device = MagicMock(spec=AppleTV)
617    device.features.in_state.side_effect = lambda _state, feature: feature == FeatureName.Artwork
618    artwork_id = "https://artwork.example/{w}x{h}/cover"
619    device.metadata.artwork_id = artwork_id
620    device.metadata.artwork = AsyncMock(
621        return_value=ArtworkInfo(b"artwork-bytes", "image/jpeg", 512, 288)
622    )
623    player._mrp_device = device
624    provider = AirPlayProvider.__new__(AirPlayProvider)
625    provider.mass = MagicMock()
626    provider.mass.players.get_player.return_value = player
627    artwork_path = f"{EXTERNAL_ARTWORK_PATH_PREFIX}/{PLAYER_ID}/{artwork_id}"
628
629    assert await provider.resolve_image(artwork_path) == b"artwork-bytes"
630
631    device.metadata.artwork_id = "replacement-artwork"
632    with pytest.raises(MediaNotFoundError):
633        await provider.resolve_image(artwork_path)
634
635
636def test_mrp_update_without_app_uses_generic_source() -> None:
637    """MRP playback without app metadata uses the generic Apple source."""
638    player = _make_control_player()
639    device = MagicMock(spec=AppleTV)
640    device.metadata.app = None
641    player._mrp_device = device
642
643    with patch.object(player, "update_state"):
644        player._handle_playing_update(
645            Playing(device_state=DeviceState.Playing, title="External media")
646        )
647
648    assert player.active_source == "airplay_control"
649    assert player.current_media is not None
650    assert player.current_media.title == "External media"
651    assert player.source_list[0].name == "AirPlay device"
652
653
654def test_mrp_idle_update_clears_external_media() -> None:
655    """An MRP idle update clears the previous external source and media."""
656    player = _make_control_player()
657    player._attr_playback_state = PlaybackState.PLAYING
658    player._attr_active_source = "com.apple.Music"
659    player._attr_current_media = MagicMock()
660
661    with patch.object(player, "update_state"):
662        player._handle_playing_update(Playing(device_state=DeviceState.Idle))
663
664    assert player.playback_state == PlaybackState.IDLE
665    assert player.active_source is None
666    assert player.current_media is None
667
668
669def test_mrp_loading_snapshot_without_playback_stays_idle() -> None:
670    """A Loading snapshot on an idle player maps to idle, not playing."""
671    # HomePods can hold a stale Loading session (with the cached metadata of a
672    # long-dead AirPlay sender) for days; trusting it would show an eternal
673    # "playing" at position 0.
674    player = _make_control_player()
675    device = MagicMock(spec=AppleTV)
676    device.metadata.app = App("Music", "com.apple.Music")
677    player._mrp_device = device
678    playing = Playing(
679        media_type=PyatvMediaType.Music,
680        device_state=DeviceState.Loading,
681        title="King of Spain",
682        total_time=209,
683        position=0,
684    )
685
686    with patch.object(player, "update_state"):
687        player._handle_playing_update(playing)
688
689    assert player.playback_state == PlaybackState.IDLE
690    assert player.active_source is None
691    assert player.current_media is None
692
693
694def test_mrp_loading_update_during_playback_stays_playing() -> None:
695    """A Loading update while already playing means buffering, so keep playing."""
696    player = _make_control_player()
697    device = MagicMock(spec=AppleTV)
698    device.metadata.app = App("Music", "com.apple.Music")
699    player._mrp_device = device
700    player._attr_playback_state = PlaybackState.PLAYING
701    player._attr_active_source = "com.apple.Music"
702
703    with patch.object(player, "update_state"):
704        player._handle_playing_update(Playing(device_state=DeviceState.Loading, title="Next track"))
705
706    assert player.playback_state == PlaybackState.PLAYING
707    assert player.active_source == "com.apple.Music"
708    assert player.current_media is not None
709    assert player.current_media.title == "Next track"
710
711
712async def test_disconnect_control_services_clears_external_state() -> None:
713    """A forced teardown drops the external snapshot owned by the closed MRP link."""
714    # _handle_connection_closed ignores the late close callback for a device
715    # that was already detached, so the teardown itself must clear the snapshot.
716    player = _make_control_player()
717    device = MagicMock(spec=AppleTV)
718    player._mrp_device = device
719    player._attr_playback_state = PlaybackState.PLAYING
720    player._attr_active_source = "com.apple.Music"
721    player._attr_current_media = MagicMock()
722
723    with patch.object(player, "update_state"):
724        await player._disconnect_control_services()
725
726    assert player.playback_state == PlaybackState.IDLE
727    assert player.active_source is None
728    assert player.current_media is None
729
730
731def test_lost_mrp_connection_clears_external_state() -> None:
732    """Losing MRP playback monitoring clears the last external snapshot."""
733    player = _make_control_player()
734    device = MagicMock(spec=AppleTV)
735    player._mrp_device = device
736    player._attr_playback_state = PlaybackState.PAUSED
737    player._attr_active_source = "com.netflix.Netflix"
738    player._attr_current_media = MagicMock()
739
740    with (
741        patch.object(player, "update_state"),
742        patch.object(player, "_schedule_connection") as schedule_connection,
743    ):
744        player._handle_connection_closed("mrp", device)
745
746    assert player.playback_state == PlaybackState.IDLE
747    assert player.active_source is None
748    assert player.current_media is None
749    schedule_connection.assert_called_once()
750
751
752def test_failed_mrp_push_updates_clear_external_state() -> None:
753    """A dropped MRP push updater clears the last external snapshot."""
754    player = _make_control_player()
755    device = MagicMock(spec=AppleTV)
756    player._mrp_device = device
757    player._attr_playback_state = PlaybackState.PAUSED
758    player._attr_active_source = "com.netflix.Netflix"
759    player._attr_current_media = MagicMock()
760
761    with (
762        patch.object(player, "update_state"),
763        patch.object(player, "_schedule_connection"),
764    ):
765        player._handle_push_error(device, RuntimeError("push failed"))
766
767    assert player.playback_state == PlaybackState.IDLE
768    assert player.active_source is None
769    assert player.current_media is None
770
771
772def test_lost_mrp_connection_keeps_streaming_state() -> None:
773    """A lost MRP connection never overrules the state owned by our own stream."""
774    player = _make_control_player()
775    player.stream = MagicMock(running=True)
776    device = MagicMock(spec=AppleTV)
777    player._mrp_device = device
778    player._attr_playback_state = PlaybackState.PLAYING
779    player._attr_active_source = "com.netflix.Netflix"
780    media = MagicMock()
781    player._attr_current_media = media
782
783    with (
784        patch.object(player, "update_state"),
785        patch.object(player, "_schedule_connection"),
786    ):
787        player._handle_connection_closed("mrp", device)
788
789    assert player.playback_state == PlaybackState.PLAYING
790    assert player.active_source == "com.netflix.Netflix"
791    assert player.current_media is media
792
793
794def test_lost_companion_connection_keeps_external_state() -> None:
795    """External playback comes from MRP only, so a Companion drop leaves it alone."""
796    player = _make_control_player()
797    device = MagicMock(spec=AppleTV)
798    player._companion_device = device
799    player._mrp_device = MagicMock(spec=AppleTV)
800    player._attr_playback_state = PlaybackState.PAUSED
801    player._attr_active_source = "com.netflix.Netflix"
802
803    with (
804        patch.object(player, "update_state"),
805        patch.object(player, "_schedule_connection"),
806    ):
807        player._handle_connection_closed("companion", device)
808
809    assert player.playback_state == PlaybackState.PAUSED
810    assert player.active_source == "com.netflix.Netflix"
811
812
813@pytest.mark.parametrize("source", ["companion", "mrp"])
814def test_lost_connection_closes_pyatv_facade(source: str) -> None:
815    """A dropped connection closes the facade instead of leaking its aiohttp session."""
816    player = _make_control_player()
817    device = MagicMock(spec=AppleTV)
818    if source == "companion":
819        player._companion_device = device
820    else:
821        player._mrp_device = device
822
823    with (
824        patch.object(player, "update_state"),
825        patch.object(player, "_schedule_connection"),
826    ):
827        player._handle_connection_closed(source, device)
828
829    device.close.assert_called_once_with()
830
831
832def test_close_callback_for_detached_device_leaves_connection_alone() -> None:
833    """A late callback for an already-replaced device never closes the live one."""
834    player = _make_control_player()
835    detached = MagicMock(spec=AppleTV)
836    connected = MagicMock(spec=AppleTV)
837    player._mrp_device = connected
838
839    with (
840        patch.object(player, "update_state"),
841        patch.object(player, "_schedule_connection") as schedule_connection,
842    ):
843        player._handle_connection_closed("mrp", detached)
844
845    detached.close.assert_not_called()
846    connected.close.assert_not_called()
847    assert player._mrp_device is connected
848    schedule_connection.assert_not_called()
849
850
851async def test_mrp_retry_does_not_recycle_connected_companion() -> None:
852    """A failed MRP monitor leaves an active Companion control channel intact."""
853    player = _make_control_player()
854    player._companion_device = MagicMock(spec=AppleTV)
855
856    with (
857        patch.object(player, "_connect_mrp", new=AsyncMock(return_value=True)),
858        patch.object(player, "_disconnect_control_services", new=AsyncMock()) as disconnect,
859    ):
860        assert await player._connect_control_services() is True
861
862    disconnect.assert_not_awaited()
863
864
865async def test_connection_retains_listener_references() -> None:
866    """Pyatv listeners remain strongly referenced for the connection lifetime."""
867    player = _make_control_player(setup_data={CONF_COMPANION_CREDENTIALS: "companion-creds"})
868    device = MagicMock(spec=AppleTV)
869    device.features.in_state.return_value = False
870
871    with (
872        patch(
873            "music_assistant.providers.airplay.control_player.pyatv.connect", return_value=device
874        ),
875        patch.object(player, "update_state"),
876    ):
877        assert await player._connect_companion() is False
878
879    assert player._companion_listener is not None
880    await player._disconnect_control_services()
881    assert player._companion_listener is None
882
883
884async def test_mrp_connection_uses_dedicated_pairing_credentials() -> None:
885    """Playback monitoring connects with pyatv's complete AirPlay credentials."""
886    credentials = "ltpk:ltsk:accessory-id:client-id"
887    player = _make_control_player(setup_data={CONF_MRP_CREDENTIALS: credentials})
888    device = MagicMock(spec=AppleTV)
889    device.features.in_state.side_effect = lambda _state, feature: (
890        feature == FeatureName.PushUpdates
891    )
892    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
893
894    with (
895        patch(
896            "music_assistant.providers.airplay.control_player.pyatv.connect",
897            return_value=device,
898        ) as connect,
899        patch.object(player, "update_state"),
900    ):
901        assert await player._connect_mrp() is False
902
903    assert connect.await_args is not None
904    config = connect.await_args.args[0]
905    service = config.get_service(Protocol.AirPlay)
906    assert service is not None
907    assert service.credentials == credentials
908    assert player._mrp_state_listener is not None
909    assert player._mrp_push_listener is not None
910
911
912async def test_rejected_mrp_credentials_are_cleared() -> None:
913    """Rejected MRP credentials return playback monitoring to setup state."""
914    setup_data: dict[str, object] = {CONF_MRP_CREDENTIALS: "invalid-creds"}
915    player = _make_control_player(setup_data=setup_data)
916
917    with patch(
918        "music_assistant.providers.airplay.control_player.pyatv.connect",
919        side_effect=pyatv_exceptions.InvalidCredentialsError(),
920    ):
921        assert await player._connect_mrp() is False
922
923    # the rejected credential is cleared from the player's setup_data
924    assert setup_data[CONF_MRP_CREDENTIALS] is None
925
926
927async def test_homepod_mrp_connection_uses_transient_credentials() -> None:
928    """HomePod playback monitoring connects without persisted credentials."""
929    player = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
930    device = MagicMock(spec=AppleTV)
931    device.features.in_state.side_effect = lambda _state, feature: (
932        feature == FeatureName.PushUpdates
933    )
934    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
935
936    with (
937        patch(
938            "music_assistant.providers.airplay.control_player.pyatv.connect",
939            return_value=device,
940        ) as connect,
941        patch.object(player, "update_state"),
942    ):
943        assert await player._connect_mrp() is False
944
945    assert connect.await_args is not None
946    config = connect.await_args.args[0]
947    service = config.get_service(Protocol.AirPlay)
948    assert service is not None
949    assert service.credentials is None
950
951
952async def test_third_party_mrp_tunnel_is_forced_from_capabilities() -> None:
953    """A third-party MRP tunnel is not blocked by pyatv's model allowlist."""
954    player = _make_control_player(companion_flags=None)
955    assert player.airplay_discovery_info is not None
956    player.airplay_discovery_info.decoded_properties["model"] = "ThirdPartyReceiver1,1"
957    device = MagicMock(spec=AppleTV)
958    device.features.in_state.side_effect = lambda _state, feature: (
959        feature == FeatureName.PushUpdates
960    )
961    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
962
963    with (
964        patch(
965            "music_assistant.providers.airplay.control_player.pyatv.connect",
966            return_value=device,
967        ) as connect,
968        patch.object(player, "update_state"),
969    ):
970        assert await player._connect_mrp() is False
971
972    assert connect.await_args is not None
973    config = connect.await_args.args[0]
974    storage = connect.await_args.kwargs["storage"]
975    settings = await storage.get_settings(config)
976    assert settings.protocols.airplay.mrp_tunnel == MrpTunnel.Force
977
978
979async def test_native_mrp_connection_uses_advertised_service() -> None:
980    """Native MRP monitoring connects directly to its advertised endpoint."""
981    player = _make_control_player(companion_flags=None)
982    player.mrp_discovery_info = _service_info(
983        MRP_DISCOVERY_TYPE,
984        properties={"SystemBuildVersion": "18A123"},
985        address="192.168.1.30",
986    )
987    device = MagicMock(spec=AppleTV)
988    device.features.in_state.side_effect = lambda _state, feature: (
989        feature == FeatureName.PushUpdates
990    )
991    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
992
993    with (
994        patch(
995            "music_assistant.providers.airplay.control_player.pyatv.connect",
996            return_value=device,
997        ) as connect,
998        patch.object(player, "update_state"),
999    ):
1000        assert await player._connect_mrp() is False
1001
1002    assert connect.await_args is not None
1003    config = connect.await_args.args[0]
1004    service = config.get_service(Protocol.MRP)
1005    assert service is not None
1006    assert service.port == player.mrp_discovery_info.port
1007
1008
1009def test_mrp_credentials_are_scoped_to_transport() -> None:
1010    """Native and tunneled MRP keep independent pairing credentials."""
1011    player = _make_control_player(
1012        setup_data={
1013            CONF_MRP_CREDENTIALS: "tunnel-creds",
1014            CONF_NATIVE_MRP_CREDENTIALS: "native-creds",
1015        }
1016    )
1017    assert player._mrp_credentials == "tunnel-creds"
1018
1019    player.mrp_discovery_info = _service_info(
1020        MRP_DISCOVERY_TYPE,
1021        properties={"SystemBuildVersion": "18A123"},
1022    )
1023
1024    assert player._mrp_credentials == "native-creds"
1025
1026
1027def test_protocol_config_uses_its_own_discovery_address() -> None:
1028    """MRP connects to the AirPlay endpoint rather than a Companion-only address."""
1029    player = _make_control_player()
1030    player.companion_discovery_info = _service_info(
1031        COMPANION_DISCOVERY_TYPE,
1032        address="192.168.1.10",
1033    )
1034    airplay_info = _service_info(
1035        AIRPLAY_DISCOVERY_TYPE,
1036        address="192.168.1.20",
1037    )
1038
1039    config = player._build_config(
1040        airplay_info,
1041        Protocol.AirPlay,
1042        None,
1043        PairingRequirement.NotNeeded,
1044    )
1045
1046    assert config is not None
1047    assert str(config.address) == "192.168.1.20"
1048
1049
1050@pytest.mark.parametrize(
1051    ("manufacturer", "model", "expected_type"),
1052    [
1053        ("Apple", "Apple TV 4K", AirPlayControlPlayer),
1054        ("Apple", "HomePod Mini", AirPlayControlPlayer),
1055        # Macs are multi-purpose devices, not standalone players.
1056        ("Apple", "MacBook Pro", GenericAirPlayPlayer),
1057        # Advertised control capabilities never promote a third-party receiver.
1058        ("Receiver", "Generic", GenericAirPlayPlayer),
1059    ],
1060)
1061async def test_provider_selects_player_model_from_device_identity(
1062    manufacturer: str,
1063    model: str,
1064    expected_type: type[AirPlayPlayer],
1065) -> None:
1066    """The player model follows the device identity, not discovered capabilities."""
1067    provider = AirPlayProvider.__new__(AirPlayProvider)
1068    provider.mass = MagicMock()
1069    provider.logger = logging.getLogger("test.airplay.provider")
1070    provider.config = MagicMock()
1071    provider.config.instance_id = "airplay"
1072    provider._bridge_manager = MagicMock()
1073    provider._bridge_manager.evaluate_bridge = AsyncMock()
1074    provider.dashboards = MagicMock()
1075    provider._companion_info_by_address = {
1076        "192.168.1.10": _service_info(
1077            COMPANION_DISCOVERY_TYPE,
1078            properties={"rpFl": "0x367A2"},
1079        )
1080    }
1081    provider._mrp_info_by_address = {}
1082    provider.mass.discovery.async_find_mdns_service = AsyncMock(return_value=None)
1083    provider.mass.players.register = AsyncMock()
1084    provider.mass.players.get_player.return_value = None
1085    provider.mass.config.get_raw_player_config_value.side_effect = (
1086        lambda _player_id, key, default=None: True if key == "enabled" else default
1087    )
1088    info = _service_info(
1089        AIRPLAY_DISCOVERY_TYPE,
1090        properties={"features": AP2_FEATURES},
1091    )
1092
1093    with (
1094        patch(
1095            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
1096            return_value="192.168.1.10",
1097        ),
1098        patch(
1099            "music_assistant.providers.airplay.provider.get_model_info",
1100            return_value=(manufacturer, model),
1101        ),
1102    ):
1103        await provider._setup_player("player", "Player", info)
1104
1105    player = provider.mass.players.register.await_args.args[0]
1106    assert isinstance(player, expected_type)
1107
1108
1109async def test_apple_device_setup_attaches_discovered_companion_service() -> None:
1110    """An Apple device picks up its Companion service discovered by name."""
1111    provider = AirPlayProvider.__new__(AirPlayProvider)
1112    provider.mass = MagicMock()
1113    provider.logger = logging.getLogger("test.airplay.provider")
1114    provider.config = MagicMock()
1115    provider.config.instance_id = "airplay"
1116    provider._bridge_manager = MagicMock()
1117    provider._bridge_manager.evaluate_bridge = AsyncMock()
1118    provider.dashboards = MagicMock()
1119    provider._companion_info_by_address = {}
1120    provider._mrp_info_by_address = {}
1121    companion_info = _service_info(
1122        COMPANION_DISCOVERY_TYPE,
1123        properties={"rpFl": "0x367A2"},
1124    )
1125
1126    async def _find_service(
1127        service_type: str,
1128        display_name: str,
1129        timeout: float = 3.0,
1130    ) -> MagicMock | None:
1131        _ = display_name, timeout
1132        return companion_info if service_type == COMPANION_DISCOVERY_TYPE else None
1133
1134    provider.mass.discovery.async_find_mdns_service = AsyncMock(side_effect=_find_service)
1135    provider.mass.players.register = AsyncMock()
1136    provider.mass.players.get_player.return_value = None
1137    provider.mass.config.get_raw_player_config_value.side_effect = (
1138        lambda _player_id, key, default=None: True if key == "enabled" else default
1139    )
1140    info = _service_info(
1141        AIRPLAY_DISCOVERY_TYPE,
1142        properties={"model": "AppleTV11,1"},
1143    )
1144    info.parsed_addresses.return_value = ["fd00::10", "192.168.1.10"]
1145
1146    with (
1147        patch(
1148            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
1149            return_value="fd00::10",
1150        ),
1151        patch(
1152            "music_assistant.providers.airplay.provider.get_model_info",
1153            return_value=("Apple", "Apple TV 4K"),
1154        ),
1155    ):
1156        await provider._setup_player("apple", "Apple TV", info)
1157
1158    player = provider.mass.players.register.await_args.args[0]
1159    assert isinstance(player, AirPlayControlPlayer)
1160    assert player.companion_discovery_info is companion_info
1161
1162
1163async def test_companion_discovery_is_attached_and_retained_during_sleep() -> None:
1164    """The last Companion endpoint remains available when mDNS withdraws it."""
1165    provider = AirPlayProvider.__new__(AirPlayProvider)
1166    provider.mass = MagicMock()
1167    provider._companion_info_by_address = {}
1168    player = _make_control_player(companion_flags=None)
1169    player.address = "fd00::10"
1170    assert player.airplay_discovery_info is not None
1171    cast("MagicMock", player.airplay_discovery_info).parsed_addresses.return_value = [
1172        "fd00::10",
1173        "192.168.1.10",
1174    ]
1175    set_companion_info = AsyncMock()
1176    player.set_companion_discovery_info = set_companion_info  # type: ignore[method-assign]
1177    provider.get_players = MagicMock(return_value=[player])  # type: ignore[method-assign]
1178    provider.mass.players.get_player.return_value = player
1179    info = _service_info(COMPANION_DISCOVERY_TYPE)
1180
1181    await provider._handle_companion_service_state_change(
1182        info.name,
1183        ServiceStateChange.Added,
1184        info,
1185    )
1186    set_companion_info.assert_awaited_once_with(info)
1187
1188    await provider._handle_companion_service_state_change(
1189        info.name,
1190        ServiceStateChange.Removed,
1191        None,
1192    )
1193    set_companion_info.assert_awaited_once_with(info)
1194    assert provider._companion_info_by_address["192.168.1.10"] is info
1195
1196
1197async def test_control_service_cache_prunes_addresses_no_longer_advertised() -> None:
1198    """A control service that moves to a new address leaves no stale cache entry."""
1199    provider = AirPlayProvider.__new__(AirPlayProvider)
1200    provider.mass = MagicMock()
1201    provider._companion_info_by_address = {}
1202    provider.get_players = MagicMock(return_value=[])  # type: ignore[method-assign]
1203    info = _service_info(COMPANION_DISCOVERY_TYPE)
1204    other_info = _service_info(COMPANION_DISCOVERY_TYPE, address="192.168.1.60")
1205    other_info.name = f"other.{COMPANION_DISCOVERY_TYPE}"
1206    moved_info = _service_info(COMPANION_DISCOVERY_TYPE, address="192.168.1.40")
1207
1208    for service_info in (info, other_info, moved_info):
1209        await provider._handle_companion_service_state_change(
1210            service_info.name,
1211            ServiceStateChange.Updated,
1212            service_info,
1213        )
1214
1215    assert "192.168.1.10" not in provider._companion_info_by_address
1216    assert provider._companion_info_by_address["192.168.1.40"] is moved_info
1217    assert provider._companion_info_by_address["192.168.1.60"] is other_info
1218
1219
1220async def test_mrp_discovery_is_attached_by_shared_address() -> None:
1221    """A native MRP service is attached to its controlled AirPlay endpoint."""
1222    provider = AirPlayProvider.__new__(AirPlayProvider)
1223    provider.mass = MagicMock()
1224    provider._mrp_info_by_address = {}
1225    player = _make_control_player(companion_flags=None)
1226    set_mrp_info = AsyncMock()
1227    player.set_mrp_discovery_info = set_mrp_info  # type: ignore[method-assign]
1228    provider.get_players = MagicMock(return_value=[player])  # type: ignore[method-assign]
1229    info = _service_info(MRP_DISCOVERY_TYPE)
1230
1231    await provider._handle_mrp_service_state_change(
1232        info.name,
1233        ServiceStateChange.Added,
1234        info,
1235    )
1236
1237    set_mrp_info.assert_awaited_once_with(info)
1238    assert provider._mrp_info_by_address["192.168.1.10"] is info
1239
1240
1241async def test_late_companion_discovery_never_changes_player_model() -> None:
1242    """A control service appearing for a generic endpoint never changes its model."""
1243    provider = AirPlayProvider.__new__(AirPlayProvider)
1244    provider.mass = MagicMock()
1245    provider.logger = logging.getLogger("test.airplay.provider")
1246    provider.config = MagicMock()
1247    provider.config.instance_id = "airplay"
1248    provider._companion_info_by_address = {}
1249    # a bare provider (built via __new__) lacks the bridge manager that
1250    # supported_features consults; a real one exists before any player does
1251    provider._bridge_manager = MagicMock(get_bridge=MagicMock(return_value=None))
1252    generic_player = GenericAirPlayPlayer(
1253        provider=provider,
1254        player_id=PLAYER_ID,
1255        raop_discovery_info=None,
1256        airplay_discovery_info=_service_info(AIRPLAY_DISCOVERY_TYPE),
1257        address="192.168.1.10",
1258        display_name="Controlled receiver",
1259        manufacturer="Receiver",
1260        model="Model",
1261    )
1262    provider.get_players = MagicMock(return_value=[generic_player])  # type: ignore[method-assign]
1263    provider.mass.players.unregister = AsyncMock()
1264    provider.mass.players.register = AsyncMock()
1265    info = _service_info(
1266        COMPANION_DISCOVERY_TYPE,
1267        properties={"rpFl": "0x367A2"},
1268    )
1269
1270    await provider._handle_companion_service_state_change(
1271        info.name,
1272        ServiceStateChange.Added,
1273        info,
1274    )
1275
1276    provider.mass.players.unregister.assert_not_awaited()
1277    provider.mass.players.register.assert_not_awaited()
1278
1279
1280async def test_generic_device_setup_skips_control_discovery() -> None:
1281    """A non-Apple receiver registers without Companion/MRP service lookups."""
1282    provider = AirPlayProvider.__new__(AirPlayProvider)
1283    provider.mass = MagicMock()
1284    provider.logger = logging.getLogger("test.airplay.provider")
1285    provider.config = MagicMock()
1286    provider.config.instance_id = "airplay"
1287    provider._bridge_manager = MagicMock()
1288    provider._bridge_manager.evaluate_bridge = AsyncMock()
1289    provider.dashboards = MagicMock()
1290    provider._companion_info_by_address = {}
1291    provider._mrp_info_by_address = {}
1292    provider.mass.discovery.async_find_mdns_service = AsyncMock(return_value=None)
1293    provider.mass.players.register = AsyncMock()
1294    provider.mass.players.get_player.return_value = None
1295    provider.mass.config.get_raw_player_config_value.side_effect = (
1296        lambda _player_id, key, default=None: True if key == "enabled" else default
1297    )
1298    info = _service_info(
1299        AIRPLAY_DISCOVERY_TYPE,
1300        properties={"model": "ThirdPartyReceiver1,1", "features": "0x0"},
1301    )
1302
1303    with (
1304        patch(
1305            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
1306            return_value="192.168.1.10",
1307        ),
1308        patch(
1309            "music_assistant.providers.airplay.provider.get_model_info",
1310            return_value=("Receiver", "Generic"),
1311        ),
1312    ):
1313        await provider._setup_player("player", "Player", info)
1314
1315    player = provider.mass.players.register.await_args.args[0]
1316    assert isinstance(player, GenericAirPlayPlayer)
1317    control_lookups = [
1318        lookup_call
1319        for lookup_call in provider.mass.discovery.async_find_mdns_service.await_args_list
1320        if lookup_call.args
1321        and lookup_call.args[0] in (COMPANION_DISCOVERY_TYPE, MRP_DISCOVERY_TYPE)
1322    ]
1323    assert not control_lookups
1324
1325
1326async def test_related_discovery_finds_differently_named_cached_service() -> None:
1327    """Control discovery can match a cached service by address instead of name."""
1328    provider = AirPlayProvider.__new__(AirPlayProvider)
1329    provider.mass = MagicMock()
1330    provider.mass.discovery.aiozc.zeroconf.cache.cache = {
1331        "_companion-link._tcp.local.": {},
1332        "uuid._companion-link._tcp.local.": {},
1333    }
1334    info = _service_info(COMPANION_DISCOVERY_TYPE)
1335    info.async_request = AsyncMock(return_value=True)
1336
1337    with patch(
1338        "music_assistant.providers.airplay.provider.AsyncServiceInfo",
1339        return_value=info,
1340    ) as service_info:
1341        result = await provider._find_cached_discovery_info(
1342            COMPANION_DISCOVERY_TYPE,
1343            {"192.168.1.10"},
1344        )
1345
1346    assert result is info
1347    service_info.assert_called_once_with(
1348        COMPANION_DISCOVERY_TYPE,
1349        "uuid._companion-link._tcp.local.",
1350    )
1351
1352
1353async def test_list_installed_app_ids_none_when_feature_unavailable() -> None:
1354    """No AppList feature available on the device returns None."""
1355    player = _make_control_player()
1356    with patch.object(player, "_device_for_feature", return_value=None):
1357        assert await player.async_list_installed_app_ids() is None
1358
1359
1360async def test_list_installed_app_ids_none_on_command_error() -> None:
1361    """A control-channel error while listing apps returns None."""
1362    player = _make_control_player()
1363    device = MagicMock()
1364    device.apps.app_list = AsyncMock(side_effect=pyatv_exceptions.CommandError("boom"))
1365    with patch.object(player, "_device_for_feature", return_value=device):
1366        assert await player.async_list_installed_app_ids() is None
1367
1368
1369async def test_list_installed_app_ids_returns_bundle_ids() -> None:
1370    """A successful app list returns the set of installed bundle ids."""
1371    player = _make_control_player()
1372    device = MagicMock()
1373    device.apps.app_list = AsyncMock(return_value=[App("App A", "com.a"), App("App B", "com.b")])
1374    with patch.object(player, "_device_for_feature", return_value=device):
1375        assert await player.async_list_installed_app_ids() == {"com.a", "com.b"}
1376
1377
1378async def test_launch_app_raises_when_feature_unavailable() -> None:
1379    """Launching with no LaunchApp feature raises PlayerCommandFailed."""
1380    player = _make_control_player()
1381    with (
1382        patch.object(player, "_device_for_feature", return_value=None),
1383        pytest.raises(PlayerCommandFailed),
1384    ):
1385        await player.async_launch_app("musicassistant://dashboard/show")
1386
1387
1388async def test_launch_app_wraps_command_error() -> None:
1389    """A control-channel error while launching raises PlayerCommandFailed."""
1390    player = _make_control_player()
1391    device = MagicMock()
1392    device.apps.launch_app = AsyncMock(side_effect=pyatv_exceptions.CommandError("nope"))
1393    with (
1394        patch.object(player, "_device_for_feature", return_value=device),
1395        pytest.raises(PlayerCommandFailed),
1396    ):
1397        await player.async_launch_app("musicassistant://dashboard/show")
1398
1399
1400async def test_launch_app_dispatches_to_device() -> None:
1401    """Launching on an available device dispatches the value over Companion."""
1402    player = _make_control_player()
1403    device = MagicMock()
1404    device.apps.launch_app = AsyncMock()
1405    with patch.object(player, "_device_for_feature", return_value=device):
1406        await player.async_launch_app("musicassistant://dashboard/show")
1407    device.apps.launch_app.assert_awaited_once_with("musicassistant://dashboard/show")
1408