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
813async def test_mrp_retry_does_not_recycle_connected_companion() -> None:
814    """A failed MRP monitor leaves an active Companion control channel intact."""
815    player = _make_control_player()
816    player._companion_device = MagicMock(spec=AppleTV)
817
818    with (
819        patch.object(player, "_connect_mrp", new=AsyncMock(return_value=True)),
820        patch.object(player, "_disconnect_control_services", new=AsyncMock()) as disconnect,
821    ):
822        assert await player._connect_control_services() is True
823
824    disconnect.assert_not_awaited()
825
826
827async def test_connection_retains_listener_references() -> None:
828    """Pyatv listeners remain strongly referenced for the connection lifetime."""
829    player = _make_control_player(setup_data={CONF_COMPANION_CREDENTIALS: "companion-creds"})
830    device = MagicMock(spec=AppleTV)
831    device.features.in_state.return_value = False
832
833    with (
834        patch(
835            "music_assistant.providers.airplay.control_player.pyatv.connect", return_value=device
836        ),
837        patch.object(player, "update_state"),
838    ):
839        assert await player._connect_companion() is False
840
841    assert player._companion_listener is not None
842    await player._disconnect_control_services()
843    assert player._companion_listener is None
844
845
846async def test_mrp_connection_uses_dedicated_pairing_credentials() -> None:
847    """Playback monitoring connects with pyatv's complete AirPlay credentials."""
848    credentials = "ltpk:ltsk:accessory-id:client-id"
849    player = _make_control_player(setup_data={CONF_MRP_CREDENTIALS: credentials})
850    device = MagicMock(spec=AppleTV)
851    device.features.in_state.side_effect = lambda _state, feature: (
852        feature == FeatureName.PushUpdates
853    )
854    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
855
856    with (
857        patch(
858            "music_assistant.providers.airplay.control_player.pyatv.connect",
859            return_value=device,
860        ) as connect,
861        patch.object(player, "update_state"),
862    ):
863        assert await player._connect_mrp() is False
864
865    assert connect.await_args is not None
866    config = connect.await_args.args[0]
867    service = config.get_service(Protocol.AirPlay)
868    assert service is not None
869    assert service.credentials == credentials
870    assert player._mrp_state_listener is not None
871    assert player._mrp_push_listener is not None
872
873
874async def test_rejected_mrp_credentials_are_cleared() -> None:
875    """Rejected MRP credentials return playback monitoring to setup state."""
876    setup_data: dict[str, object] = {CONF_MRP_CREDENTIALS: "invalid-creds"}
877    player = _make_control_player(setup_data=setup_data)
878
879    with patch(
880        "music_assistant.providers.airplay.control_player.pyatv.connect",
881        side_effect=pyatv_exceptions.InvalidCredentialsError(),
882    ):
883        assert await player._connect_mrp() is False
884
885    # the rejected credential is cleared from the player's setup_data
886    assert setup_data[CONF_MRP_CREDENTIALS] is None
887
888
889async def test_homepod_mrp_connection_uses_transient_credentials() -> None:
890    """HomePod playback monitoring connects without persisted credentials."""
891    player = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
892    device = MagicMock(spec=AppleTV)
893    device.features.in_state.side_effect = lambda _state, feature: (
894        feature == FeatureName.PushUpdates
895    )
896    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
897
898    with (
899        patch(
900            "music_assistant.providers.airplay.control_player.pyatv.connect",
901            return_value=device,
902        ) as connect,
903        patch.object(player, "update_state"),
904    ):
905        assert await player._connect_mrp() is False
906
907    assert connect.await_args is not None
908    config = connect.await_args.args[0]
909    service = config.get_service(Protocol.AirPlay)
910    assert service is not None
911    assert service.credentials is None
912
913
914async def test_third_party_mrp_tunnel_is_forced_from_capabilities() -> None:
915    """A third-party MRP tunnel is not blocked by pyatv's model allowlist."""
916    player = _make_control_player(companion_flags=None)
917    assert player.airplay_discovery_info is not None
918    player.airplay_discovery_info.decoded_properties["model"] = "ThirdPartyReceiver1,1"
919    device = MagicMock(spec=AppleTV)
920    device.features.in_state.side_effect = lambda _state, feature: (
921        feature == FeatureName.PushUpdates
922    )
923    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
924
925    with (
926        patch(
927            "music_assistant.providers.airplay.control_player.pyatv.connect",
928            return_value=device,
929        ) as connect,
930        patch.object(player, "update_state"),
931    ):
932        assert await player._connect_mrp() is False
933
934    assert connect.await_args is not None
935    config = connect.await_args.args[0]
936    storage = connect.await_args.kwargs["storage"]
937    settings = await storage.get_settings(config)
938    assert settings.protocols.airplay.mrp_tunnel == MrpTunnel.Force
939
940
941async def test_native_mrp_connection_uses_advertised_service() -> None:
942    """Native MRP monitoring connects directly to its advertised endpoint."""
943    player = _make_control_player(companion_flags=None)
944    player.mrp_discovery_info = _service_info(
945        MRP_DISCOVERY_TYPE,
946        properties={"SystemBuildVersion": "18A123"},
947        address="192.168.1.30",
948    )
949    device = MagicMock(spec=AppleTV)
950    device.features.in_state.side_effect = lambda _state, feature: (
951        feature == FeatureName.PushUpdates
952    )
953    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
954
955    with (
956        patch(
957            "music_assistant.providers.airplay.control_player.pyatv.connect",
958            return_value=device,
959        ) as connect,
960        patch.object(player, "update_state"),
961    ):
962        assert await player._connect_mrp() is False
963
964    assert connect.await_args is not None
965    config = connect.await_args.args[0]
966    service = config.get_service(Protocol.MRP)
967    assert service is not None
968    assert service.port == player.mrp_discovery_info.port
969
970
971def test_mrp_credentials_are_scoped_to_transport() -> None:
972    """Native and tunneled MRP keep independent pairing credentials."""
973    player = _make_control_player(
974        setup_data={
975            CONF_MRP_CREDENTIALS: "tunnel-creds",
976            CONF_NATIVE_MRP_CREDENTIALS: "native-creds",
977        }
978    )
979    assert player._mrp_credentials == "tunnel-creds"
980
981    player.mrp_discovery_info = _service_info(
982        MRP_DISCOVERY_TYPE,
983        properties={"SystemBuildVersion": "18A123"},
984    )
985
986    assert player._mrp_credentials == "native-creds"
987
988
989def test_protocol_config_uses_its_own_discovery_address() -> None:
990    """MRP connects to the AirPlay endpoint rather than a Companion-only address."""
991    player = _make_control_player()
992    player.companion_discovery_info = _service_info(
993        COMPANION_DISCOVERY_TYPE,
994        address="192.168.1.10",
995    )
996    airplay_info = _service_info(
997        AIRPLAY_DISCOVERY_TYPE,
998        address="192.168.1.20",
999    )
1000
1001    config = player._build_config(
1002        airplay_info,
1003        Protocol.AirPlay,
1004        None,
1005        PairingRequirement.NotNeeded,
1006    )
1007
1008    assert config is not None
1009    assert str(config.address) == "192.168.1.20"
1010
1011
1012@pytest.mark.parametrize(
1013    ("manufacturer", "model", "expected_type"),
1014    [
1015        ("Apple", "Apple TV 4K", AirPlayControlPlayer),
1016        ("Apple", "HomePod Mini", AirPlayControlPlayer),
1017        # Macs are multi-purpose devices, not standalone players.
1018        ("Apple", "MacBook Pro", GenericAirPlayPlayer),
1019        # Advertised control capabilities never promote a third-party receiver.
1020        ("Receiver", "Generic", GenericAirPlayPlayer),
1021    ],
1022)
1023async def test_provider_selects_player_model_from_device_identity(
1024    manufacturer: str,
1025    model: str,
1026    expected_type: type[AirPlayPlayer],
1027) -> None:
1028    """The player model follows the device identity, not discovered capabilities."""
1029    provider = AirPlayProvider.__new__(AirPlayProvider)
1030    provider.mass = MagicMock()
1031    provider.logger = logging.getLogger("test.airplay.provider")
1032    provider.config = MagicMock()
1033    provider.config.instance_id = "airplay"
1034    provider._bridge_manager = MagicMock()
1035    provider._bridge_manager.evaluate_bridge = AsyncMock()
1036    provider.dashboards = MagicMock()
1037    provider._companion_info_by_address = {
1038        "192.168.1.10": _service_info(
1039            COMPANION_DISCOVERY_TYPE,
1040            properties={"rpFl": "0x367A2"},
1041        )
1042    }
1043    provider._mrp_info_by_address = {}
1044    provider.mass.discovery.async_find_mdns_service = AsyncMock(return_value=None)
1045    provider.mass.players.register = AsyncMock()
1046    provider.mass.players.get_player.return_value = None
1047    provider.mass.config.get_raw_player_config_value.side_effect = (
1048        lambda _player_id, key, default=None: True if key == "enabled" else default
1049    )
1050    info = _service_info(
1051        AIRPLAY_DISCOVERY_TYPE,
1052        properties={"features": AP2_FEATURES},
1053    )
1054
1055    with (
1056        patch(
1057            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
1058            return_value="192.168.1.10",
1059        ),
1060        patch(
1061            "music_assistant.providers.airplay.provider.get_model_info",
1062            return_value=(manufacturer, model),
1063        ),
1064    ):
1065        await provider._setup_player("player", "Player", info)
1066
1067    player = provider.mass.players.register.await_args.args[0]
1068    assert isinstance(player, expected_type)
1069
1070
1071async def test_apple_device_setup_attaches_discovered_companion_service() -> None:
1072    """An Apple device picks up its Companion service discovered by name."""
1073    provider = AirPlayProvider.__new__(AirPlayProvider)
1074    provider.mass = MagicMock()
1075    provider.logger = logging.getLogger("test.airplay.provider")
1076    provider.config = MagicMock()
1077    provider.config.instance_id = "airplay"
1078    provider._bridge_manager = MagicMock()
1079    provider._bridge_manager.evaluate_bridge = AsyncMock()
1080    provider.dashboards = MagicMock()
1081    provider._companion_info_by_address = {}
1082    provider._mrp_info_by_address = {}
1083    companion_info = _service_info(
1084        COMPANION_DISCOVERY_TYPE,
1085        properties={"rpFl": "0x367A2"},
1086    )
1087
1088    async def _find_service(
1089        service_type: str,
1090        display_name: str,
1091        timeout: float = 3.0,
1092    ) -> MagicMock | None:
1093        _ = display_name, timeout
1094        return companion_info if service_type == COMPANION_DISCOVERY_TYPE else None
1095
1096    provider.mass.discovery.async_find_mdns_service = AsyncMock(side_effect=_find_service)
1097    provider.mass.players.register = AsyncMock()
1098    provider.mass.players.get_player.return_value = None
1099    provider.mass.config.get_raw_player_config_value.side_effect = (
1100        lambda _player_id, key, default=None: True if key == "enabled" else default
1101    )
1102    info = _service_info(
1103        AIRPLAY_DISCOVERY_TYPE,
1104        properties={"model": "AppleTV11,1"},
1105    )
1106    info.parsed_addresses.return_value = ["fd00::10", "192.168.1.10"]
1107
1108    with (
1109        patch(
1110            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
1111            return_value="fd00::10",
1112        ),
1113        patch(
1114            "music_assistant.providers.airplay.provider.get_model_info",
1115            return_value=("Apple", "Apple TV 4K"),
1116        ),
1117    ):
1118        await provider._setup_player("apple", "Apple TV", info)
1119
1120    player = provider.mass.players.register.await_args.args[0]
1121    assert isinstance(player, AirPlayControlPlayer)
1122    assert player.companion_discovery_info is companion_info
1123
1124
1125async def test_companion_discovery_is_attached_and_retained_during_sleep() -> None:
1126    """The last Companion endpoint remains available when mDNS withdraws it."""
1127    provider = AirPlayProvider.__new__(AirPlayProvider)
1128    provider.mass = MagicMock()
1129    provider._companion_info_by_address = {}
1130    player = _make_control_player(companion_flags=None)
1131    player.address = "fd00::10"
1132    assert player.airplay_discovery_info is not None
1133    cast("MagicMock", player.airplay_discovery_info).parsed_addresses.return_value = [
1134        "fd00::10",
1135        "192.168.1.10",
1136    ]
1137    set_companion_info = AsyncMock()
1138    player.set_companion_discovery_info = set_companion_info  # type: ignore[method-assign]
1139    provider.get_players = MagicMock(return_value=[player])  # type: ignore[method-assign]
1140    provider.mass.players.get_player.return_value = player
1141    info = _service_info(COMPANION_DISCOVERY_TYPE)
1142
1143    await provider._handle_companion_service_state_change(
1144        info.name,
1145        ServiceStateChange.Added,
1146        info,
1147    )
1148    set_companion_info.assert_awaited_once_with(info)
1149
1150    await provider._handle_companion_service_state_change(
1151        info.name,
1152        ServiceStateChange.Removed,
1153        None,
1154    )
1155    set_companion_info.assert_awaited_once_with(info)
1156    assert provider._companion_info_by_address["192.168.1.10"] is info
1157
1158
1159async def test_control_service_cache_prunes_addresses_no_longer_advertised() -> None:
1160    """A control service that moves to a new address leaves no stale cache entry."""
1161    provider = AirPlayProvider.__new__(AirPlayProvider)
1162    provider.mass = MagicMock()
1163    provider._companion_info_by_address = {}
1164    provider.get_players = MagicMock(return_value=[])  # type: ignore[method-assign]
1165    info = _service_info(COMPANION_DISCOVERY_TYPE)
1166    other_info = _service_info(COMPANION_DISCOVERY_TYPE, address="192.168.1.60")
1167    other_info.name = f"other.{COMPANION_DISCOVERY_TYPE}"
1168    moved_info = _service_info(COMPANION_DISCOVERY_TYPE, address="192.168.1.40")
1169
1170    for service_info in (info, other_info, moved_info):
1171        await provider._handle_companion_service_state_change(
1172            service_info.name,
1173            ServiceStateChange.Updated,
1174            service_info,
1175        )
1176
1177    assert "192.168.1.10" not in provider._companion_info_by_address
1178    assert provider._companion_info_by_address["192.168.1.40"] is moved_info
1179    assert provider._companion_info_by_address["192.168.1.60"] is other_info
1180
1181
1182async def test_mrp_discovery_is_attached_by_shared_address() -> None:
1183    """A native MRP service is attached to its controlled AirPlay endpoint."""
1184    provider = AirPlayProvider.__new__(AirPlayProvider)
1185    provider.mass = MagicMock()
1186    provider._mrp_info_by_address = {}
1187    player = _make_control_player(companion_flags=None)
1188    set_mrp_info = AsyncMock()
1189    player.set_mrp_discovery_info = set_mrp_info  # type: ignore[method-assign]
1190    provider.get_players = MagicMock(return_value=[player])  # type: ignore[method-assign]
1191    info = _service_info(MRP_DISCOVERY_TYPE)
1192
1193    await provider._handle_mrp_service_state_change(
1194        info.name,
1195        ServiceStateChange.Added,
1196        info,
1197    )
1198
1199    set_mrp_info.assert_awaited_once_with(info)
1200    assert provider._mrp_info_by_address["192.168.1.10"] is info
1201
1202
1203async def test_late_companion_discovery_never_changes_player_model() -> None:
1204    """A control service appearing for a generic endpoint never changes its model."""
1205    provider = AirPlayProvider.__new__(AirPlayProvider)
1206    provider.mass = MagicMock()
1207    provider.logger = logging.getLogger("test.airplay.provider")
1208    provider.config = MagicMock()
1209    provider.config.instance_id = "airplay"
1210    provider._companion_info_by_address = {}
1211    # a bare provider (built via __new__) lacks the bridge manager that
1212    # supported_features consults; a real one exists before any player does
1213    provider._bridge_manager = MagicMock(get_bridge=MagicMock(return_value=None))
1214    generic_player = GenericAirPlayPlayer(
1215        provider=provider,
1216        player_id=PLAYER_ID,
1217        raop_discovery_info=None,
1218        airplay_discovery_info=_service_info(AIRPLAY_DISCOVERY_TYPE),
1219        address="192.168.1.10",
1220        display_name="Controlled receiver",
1221        manufacturer="Receiver",
1222        model="Model",
1223    )
1224    provider.get_players = MagicMock(return_value=[generic_player])  # type: ignore[method-assign]
1225    provider.mass.players.unregister = AsyncMock()
1226    provider.mass.players.register = AsyncMock()
1227    info = _service_info(
1228        COMPANION_DISCOVERY_TYPE,
1229        properties={"rpFl": "0x367A2"},
1230    )
1231
1232    await provider._handle_companion_service_state_change(
1233        info.name,
1234        ServiceStateChange.Added,
1235        info,
1236    )
1237
1238    provider.mass.players.unregister.assert_not_awaited()
1239    provider.mass.players.register.assert_not_awaited()
1240
1241
1242async def test_generic_device_setup_skips_control_discovery() -> None:
1243    """A non-Apple receiver registers without Companion/MRP service lookups."""
1244    provider = AirPlayProvider.__new__(AirPlayProvider)
1245    provider.mass = MagicMock()
1246    provider.logger = logging.getLogger("test.airplay.provider")
1247    provider.config = MagicMock()
1248    provider.config.instance_id = "airplay"
1249    provider._bridge_manager = MagicMock()
1250    provider._bridge_manager.evaluate_bridge = AsyncMock()
1251    provider.dashboards = MagicMock()
1252    provider._companion_info_by_address = {}
1253    provider._mrp_info_by_address = {}
1254    provider.mass.discovery.async_find_mdns_service = AsyncMock(return_value=None)
1255    provider.mass.players.register = AsyncMock()
1256    provider.mass.players.get_player.return_value = None
1257    provider.mass.config.get_raw_player_config_value.side_effect = (
1258        lambda _player_id, key, default=None: True if key == "enabled" else default
1259    )
1260    info = _service_info(
1261        AIRPLAY_DISCOVERY_TYPE,
1262        properties={"model": "ThirdPartyReceiver1,1", "features": "0x0"},
1263    )
1264
1265    with (
1266        patch(
1267            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
1268            return_value="192.168.1.10",
1269        ),
1270        patch(
1271            "music_assistant.providers.airplay.provider.get_model_info",
1272            return_value=("Receiver", "Generic"),
1273        ),
1274    ):
1275        await provider._setup_player("player", "Player", info)
1276
1277    player = provider.mass.players.register.await_args.args[0]
1278    assert isinstance(player, GenericAirPlayPlayer)
1279    control_lookups = [
1280        lookup_call
1281        for lookup_call in provider.mass.discovery.async_find_mdns_service.await_args_list
1282        if lookup_call.args
1283        and lookup_call.args[0] in (COMPANION_DISCOVERY_TYPE, MRP_DISCOVERY_TYPE)
1284    ]
1285    assert not control_lookups
1286
1287
1288async def test_related_discovery_finds_differently_named_cached_service() -> None:
1289    """Control discovery can match a cached service by address instead of name."""
1290    provider = AirPlayProvider.__new__(AirPlayProvider)
1291    provider.mass = MagicMock()
1292    provider.mass.discovery.aiozc.zeroconf.cache.cache = {
1293        "_companion-link._tcp.local.": {},
1294        "uuid._companion-link._tcp.local.": {},
1295    }
1296    info = _service_info(COMPANION_DISCOVERY_TYPE)
1297    info.async_request = AsyncMock(return_value=True)
1298
1299    with patch(
1300        "music_assistant.providers.airplay.provider.AsyncServiceInfo",
1301        return_value=info,
1302    ) as service_info:
1303        result = await provider._find_cached_discovery_info(
1304            COMPANION_DISCOVERY_TYPE,
1305            {"192.168.1.10"},
1306        )
1307
1308    assert result is info
1309    service_info.assert_called_once_with(
1310        COMPANION_DISCOVERY_TYPE,
1311        "uuid._companion-link._tcp.local.",
1312    )
1313
1314
1315async def test_list_installed_app_ids_none_when_feature_unavailable() -> None:
1316    """No AppList feature available on the device returns None."""
1317    player = _make_control_player()
1318    with patch.object(player, "_device_for_feature", return_value=None):
1319        assert await player.async_list_installed_app_ids() is None
1320
1321
1322async def test_list_installed_app_ids_none_on_command_error() -> None:
1323    """A control-channel error while listing apps returns None."""
1324    player = _make_control_player()
1325    device = MagicMock()
1326    device.apps.app_list = AsyncMock(side_effect=pyatv_exceptions.CommandError("boom"))
1327    with patch.object(player, "_device_for_feature", return_value=device):
1328        assert await player.async_list_installed_app_ids() is None
1329
1330
1331async def test_list_installed_app_ids_returns_bundle_ids() -> None:
1332    """A successful app list returns the set of installed bundle ids."""
1333    player = _make_control_player()
1334    device = MagicMock()
1335    device.apps.app_list = AsyncMock(return_value=[App("App A", "com.a"), App("App B", "com.b")])
1336    with patch.object(player, "_device_for_feature", return_value=device):
1337        assert await player.async_list_installed_app_ids() == {"com.a", "com.b"}
1338
1339
1340async def test_launch_app_raises_when_feature_unavailable() -> None:
1341    """Launching with no LaunchApp feature raises PlayerCommandFailed."""
1342    player = _make_control_player()
1343    with (
1344        patch.object(player, "_device_for_feature", return_value=None),
1345        pytest.raises(PlayerCommandFailed),
1346    ):
1347        await player.async_launch_app("musicassistant://dashboard/show")
1348
1349
1350async def test_launch_app_wraps_command_error() -> None:
1351    """A control-channel error while launching raises PlayerCommandFailed."""
1352    player = _make_control_player()
1353    device = MagicMock()
1354    device.apps.launch_app = AsyncMock(side_effect=pyatv_exceptions.CommandError("nope"))
1355    with (
1356        patch.object(player, "_device_for_feature", return_value=device),
1357        pytest.raises(PlayerCommandFailed),
1358    ):
1359        await player.async_launch_app("musicassistant://dashboard/show")
1360
1361
1362async def test_launch_app_dispatches_to_device() -> None:
1363    """Launching on an available device dispatches the value over Companion."""
1364    player = _make_control_player()
1365    device = MagicMock()
1366    device.apps.launch_app = AsyncMock()
1367    with patch.object(player, "_device_for_feature", return_value=device):
1368        await player.async_launch_app("musicassistant://dashboard/show")
1369    device.apps.launch_app.assert_awaited_once_with("musicassistant://dashboard/show")
1370