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_lost_mrp_connection_clears_external_state() -> None:
670    """Losing MRP playback monitoring clears the last external snapshot."""
671    player = _make_control_player()
672    device = MagicMock(spec=AppleTV)
673    player._mrp_device = device
674    player._attr_playback_state = PlaybackState.PAUSED
675    player._attr_active_source = "com.netflix.Netflix"
676    player._attr_current_media = MagicMock()
677
678    with (
679        patch.object(player, "update_state"),
680        patch.object(player, "_schedule_connection") as schedule_connection,
681    ):
682        player._handle_connection_closed("mrp", device)
683
684    assert player.playback_state == PlaybackState.IDLE
685    assert player.active_source is None
686    assert player.current_media is None
687    schedule_connection.assert_called_once()
688
689
690def test_failed_mrp_push_updates_clear_external_state() -> None:
691    """A dropped MRP push updater clears the last external snapshot."""
692    player = _make_control_player()
693    device = MagicMock(spec=AppleTV)
694    player._mrp_device = device
695    player._attr_playback_state = PlaybackState.PAUSED
696    player._attr_active_source = "com.netflix.Netflix"
697    player._attr_current_media = MagicMock()
698
699    with (
700        patch.object(player, "update_state"),
701        patch.object(player, "_schedule_connection"),
702    ):
703        player._handle_push_error(device, RuntimeError("push failed"))
704
705    assert player.playback_state == PlaybackState.IDLE
706    assert player.active_source is None
707    assert player.current_media is None
708
709
710def test_lost_mrp_connection_keeps_streaming_state() -> None:
711    """A lost MRP connection never overrules the state owned by our own stream."""
712    player = _make_control_player()
713    player.stream = MagicMock(running=True)
714    device = MagicMock(spec=AppleTV)
715    player._mrp_device = device
716    player._attr_playback_state = PlaybackState.PLAYING
717    player._attr_active_source = "com.netflix.Netflix"
718    media = MagicMock()
719    player._attr_current_media = media
720
721    with (
722        patch.object(player, "update_state"),
723        patch.object(player, "_schedule_connection"),
724    ):
725        player._handle_connection_closed("mrp", device)
726
727    assert player.playback_state == PlaybackState.PLAYING
728    assert player.active_source == "com.netflix.Netflix"
729    assert player.current_media is media
730
731
732def test_lost_companion_connection_keeps_external_state() -> None:
733    """External playback comes from MRP only, so a Companion drop leaves it alone."""
734    player = _make_control_player()
735    device = MagicMock(spec=AppleTV)
736    player._companion_device = device
737    player._mrp_device = MagicMock(spec=AppleTV)
738    player._attr_playback_state = PlaybackState.PAUSED
739    player._attr_active_source = "com.netflix.Netflix"
740
741    with (
742        patch.object(player, "update_state"),
743        patch.object(player, "_schedule_connection"),
744    ):
745        player._handle_connection_closed("companion", device)
746
747    assert player.playback_state == PlaybackState.PAUSED
748    assert player.active_source == "com.netflix.Netflix"
749
750
751async def test_mrp_retry_does_not_recycle_connected_companion() -> None:
752    """A failed MRP monitor leaves an active Companion control channel intact."""
753    player = _make_control_player()
754    player._companion_device = MagicMock(spec=AppleTV)
755
756    with (
757        patch.object(player, "_connect_mrp", new=AsyncMock(return_value=True)),
758        patch.object(player, "_disconnect_control_services", new=AsyncMock()) as disconnect,
759    ):
760        assert await player._connect_control_services() is True
761
762    disconnect.assert_not_awaited()
763
764
765async def test_connection_retains_listener_references() -> None:
766    """Pyatv listeners remain strongly referenced for the connection lifetime."""
767    player = _make_control_player(setup_data={CONF_COMPANION_CREDENTIALS: "companion-creds"})
768    device = MagicMock(spec=AppleTV)
769    device.features.in_state.return_value = False
770
771    with (
772        patch(
773            "music_assistant.providers.airplay.control_player.pyatv.connect", return_value=device
774        ),
775        patch.object(player, "update_state"),
776    ):
777        assert await player._connect_companion() is False
778
779    assert player._companion_listener is not None
780    await player._disconnect_control_services()
781    assert player._companion_listener is None
782
783
784async def test_mrp_connection_uses_dedicated_pairing_credentials() -> None:
785    """Playback monitoring connects with pyatv's complete AirPlay credentials."""
786    credentials = "ltpk:ltsk:accessory-id:client-id"
787    player = _make_control_player(setup_data={CONF_MRP_CREDENTIALS: credentials})
788    device = MagicMock(spec=AppleTV)
789    device.features.in_state.side_effect = lambda _state, feature: (
790        feature == FeatureName.PushUpdates
791    )
792    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
793
794    with (
795        patch(
796            "music_assistant.providers.airplay.control_player.pyatv.connect",
797            return_value=device,
798        ) as connect,
799        patch.object(player, "update_state"),
800    ):
801        assert await player._connect_mrp() is False
802
803    assert connect.await_args is not None
804    config = connect.await_args.args[0]
805    service = config.get_service(Protocol.AirPlay)
806    assert service is not None
807    assert service.credentials == credentials
808    assert player._mrp_state_listener is not None
809    assert player._mrp_push_listener is not None
810
811
812async def test_rejected_mrp_credentials_are_cleared() -> None:
813    """Rejected MRP credentials return playback monitoring to setup state."""
814    setup_data: dict[str, object] = {CONF_MRP_CREDENTIALS: "invalid-creds"}
815    player = _make_control_player(setup_data=setup_data)
816
817    with patch(
818        "music_assistant.providers.airplay.control_player.pyatv.connect",
819        side_effect=pyatv_exceptions.InvalidCredentialsError(),
820    ):
821        assert await player._connect_mrp() is False
822
823    # the rejected credential is cleared from the player's setup_data
824    assert setup_data[CONF_MRP_CREDENTIALS] is None
825
826
827async def test_homepod_mrp_connection_uses_transient_credentials() -> None:
828    """HomePod playback monitoring connects without persisted credentials."""
829    player = _make_control_player(model="HomePod Mini", companion_flags="0x62792")
830    device = MagicMock(spec=AppleTV)
831    device.features.in_state.side_effect = lambda _state, feature: (
832        feature == FeatureName.PushUpdates
833    )
834    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
835
836    with (
837        patch(
838            "music_assistant.providers.airplay.control_player.pyatv.connect",
839            return_value=device,
840        ) as connect,
841        patch.object(player, "update_state"),
842    ):
843        assert await player._connect_mrp() is False
844
845    assert connect.await_args is not None
846    config = connect.await_args.args[0]
847    service = config.get_service(Protocol.AirPlay)
848    assert service is not None
849    assert service.credentials is None
850
851
852async def test_third_party_mrp_tunnel_is_forced_from_capabilities() -> None:
853    """A third-party MRP tunnel is not blocked by pyatv's model allowlist."""
854    player = _make_control_player(companion_flags=None)
855    assert player.airplay_discovery_info is not None
856    player.airplay_discovery_info.decoded_properties["model"] = "ThirdPartyReceiver1,1"
857    device = MagicMock(spec=AppleTV)
858    device.features.in_state.side_effect = lambda _state, feature: (
859        feature == FeatureName.PushUpdates
860    )
861    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
862
863    with (
864        patch(
865            "music_assistant.providers.airplay.control_player.pyatv.connect",
866            return_value=device,
867        ) as connect,
868        patch.object(player, "update_state"),
869    ):
870        assert await player._connect_mrp() is False
871
872    assert connect.await_args is not None
873    config = connect.await_args.args[0]
874    storage = connect.await_args.kwargs["storage"]
875    settings = await storage.get_settings(config)
876    assert settings.protocols.airplay.mrp_tunnel == MrpTunnel.Force
877
878
879async def test_native_mrp_connection_uses_advertised_service() -> None:
880    """Native MRP monitoring connects directly to its advertised endpoint."""
881    player = _make_control_player(companion_flags=None)
882    player.mrp_discovery_info = _service_info(
883        MRP_DISCOVERY_TYPE,
884        properties={"SystemBuildVersion": "18A123"},
885        address="192.168.1.30",
886    )
887    device = MagicMock(spec=AppleTV)
888    device.features.in_state.side_effect = lambda _state, feature: (
889        feature == FeatureName.PushUpdates
890    )
891    device.metadata.playing = AsyncMock(return_value=Playing(device_state=DeviceState.Idle))
892
893    with (
894        patch(
895            "music_assistant.providers.airplay.control_player.pyatv.connect",
896            return_value=device,
897        ) as connect,
898        patch.object(player, "update_state"),
899    ):
900        assert await player._connect_mrp() is False
901
902    assert connect.await_args is not None
903    config = connect.await_args.args[0]
904    service = config.get_service(Protocol.MRP)
905    assert service is not None
906    assert service.port == player.mrp_discovery_info.port
907
908
909def test_mrp_credentials_are_scoped_to_transport() -> None:
910    """Native and tunneled MRP keep independent pairing credentials."""
911    player = _make_control_player(
912        setup_data={
913            CONF_MRP_CREDENTIALS: "tunnel-creds",
914            CONF_NATIVE_MRP_CREDENTIALS: "native-creds",
915        }
916    )
917    assert player._mrp_credentials == "tunnel-creds"
918
919    player.mrp_discovery_info = _service_info(
920        MRP_DISCOVERY_TYPE,
921        properties={"SystemBuildVersion": "18A123"},
922    )
923
924    assert player._mrp_credentials == "native-creds"
925
926
927def test_protocol_config_uses_its_own_discovery_address() -> None:
928    """MRP connects to the AirPlay endpoint rather than a Companion-only address."""
929    player = _make_control_player()
930    player.companion_discovery_info = _service_info(
931        COMPANION_DISCOVERY_TYPE,
932        address="192.168.1.10",
933    )
934    airplay_info = _service_info(
935        AIRPLAY_DISCOVERY_TYPE,
936        address="192.168.1.20",
937    )
938
939    config = player._build_config(
940        airplay_info,
941        Protocol.AirPlay,
942        None,
943        PairingRequirement.NotNeeded,
944    )
945
946    assert config is not None
947    assert str(config.address) == "192.168.1.20"
948
949
950@pytest.mark.parametrize(
951    ("manufacturer", "model", "expected_type"),
952    [
953        ("Apple", "Apple TV 4K", AirPlayControlPlayer),
954        ("Apple", "HomePod Mini", AirPlayControlPlayer),
955        # Macs are multi-purpose devices, not standalone players.
956        ("Apple", "MacBook Pro", GenericAirPlayPlayer),
957        # Advertised control capabilities never promote a third-party receiver.
958        ("Receiver", "Generic", GenericAirPlayPlayer),
959    ],
960)
961async def test_provider_selects_player_model_from_device_identity(
962    manufacturer: str,
963    model: str,
964    expected_type: type[AirPlayPlayer],
965) -> None:
966    """The player model follows the device identity, not discovered capabilities."""
967    provider = AirPlayProvider.__new__(AirPlayProvider)
968    provider.mass = MagicMock()
969    provider.logger = logging.getLogger("test.airplay.provider")
970    provider.config = MagicMock()
971    provider.config.instance_id = "airplay"
972    provider._bridge_manager = MagicMock()
973    provider._bridge_manager.evaluate_bridge = AsyncMock()
974    provider.dashboards = MagicMock()
975    provider._companion_info_by_address = {
976        "192.168.1.10": _service_info(
977            COMPANION_DISCOVERY_TYPE,
978            properties={"rpFl": "0x367A2"},
979        )
980    }
981    provider._mrp_info_by_address = {}
982    provider.mass.discovery.async_find_mdns_service = AsyncMock(return_value=None)
983    provider.mass.players.register = AsyncMock()
984    provider.mass.players.get_player.return_value = None
985    provider.mass.config.get_raw_player_config_value.side_effect = (
986        lambda _player_id, key, default=None: True if key == "enabled" else default
987    )
988    info = _service_info(
989        AIRPLAY_DISCOVERY_TYPE,
990        properties={"features": AP2_FEATURES},
991    )
992
993    with (
994        patch(
995            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
996            return_value="192.168.1.10",
997        ),
998        patch(
999            "music_assistant.providers.airplay.provider.get_model_info",
1000            return_value=(manufacturer, model),
1001        ),
1002    ):
1003        await provider._setup_player("player", "Player", info)
1004
1005    player = provider.mass.players.register.await_args.args[0]
1006    assert isinstance(player, expected_type)
1007
1008
1009async def test_apple_device_setup_attaches_discovered_companion_service() -> None:
1010    """An Apple device picks up its Companion service discovered by name."""
1011    provider = AirPlayProvider.__new__(AirPlayProvider)
1012    provider.mass = MagicMock()
1013    provider.logger = logging.getLogger("test.airplay.provider")
1014    provider.config = MagicMock()
1015    provider.config.instance_id = "airplay"
1016    provider._bridge_manager = MagicMock()
1017    provider._bridge_manager.evaluate_bridge = AsyncMock()
1018    provider.dashboards = MagicMock()
1019    provider._companion_info_by_address = {}
1020    provider._mrp_info_by_address = {}
1021    companion_info = _service_info(
1022        COMPANION_DISCOVERY_TYPE,
1023        properties={"rpFl": "0x367A2"},
1024    )
1025
1026    async def _find_service(
1027        service_type: str,
1028        display_name: str,
1029        timeout: float = 3.0,
1030    ) -> MagicMock | None:
1031        _ = display_name, timeout
1032        return companion_info if service_type == COMPANION_DISCOVERY_TYPE else None
1033
1034    provider.mass.discovery.async_find_mdns_service = AsyncMock(side_effect=_find_service)
1035    provider.mass.players.register = AsyncMock()
1036    provider.mass.players.get_player.return_value = None
1037    provider.mass.config.get_raw_player_config_value.side_effect = (
1038        lambda _player_id, key, default=None: True if key == "enabled" else default
1039    )
1040    info = _service_info(
1041        AIRPLAY_DISCOVERY_TYPE,
1042        properties={"model": "AppleTV11,1"},
1043    )
1044    info.parsed_addresses.return_value = ["fd00::10", "192.168.1.10"]
1045
1046    with (
1047        patch(
1048            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
1049            return_value="fd00::10",
1050        ),
1051        patch(
1052            "music_assistant.providers.airplay.provider.get_model_info",
1053            return_value=("Apple", "Apple TV 4K"),
1054        ),
1055    ):
1056        await provider._setup_player("apple", "Apple TV", info)
1057
1058    player = provider.mass.players.register.await_args.args[0]
1059    assert isinstance(player, AirPlayControlPlayer)
1060    assert player.companion_discovery_info is companion_info
1061
1062
1063async def test_companion_discovery_is_attached_and_retained_during_sleep() -> None:
1064    """The last Companion endpoint remains available when mDNS withdraws it."""
1065    provider = AirPlayProvider.__new__(AirPlayProvider)
1066    provider.mass = MagicMock()
1067    provider._companion_info_by_address = {}
1068    player = _make_control_player(companion_flags=None)
1069    player.address = "fd00::10"
1070    assert player.airplay_discovery_info is not None
1071    cast("MagicMock", player.airplay_discovery_info).parsed_addresses.return_value = [
1072        "fd00::10",
1073        "192.168.1.10",
1074    ]
1075    set_companion_info = AsyncMock()
1076    player.set_companion_discovery_info = set_companion_info  # type: ignore[method-assign]
1077    provider.get_players = MagicMock(return_value=[player])  # type: ignore[method-assign]
1078    provider.mass.players.get_player.return_value = player
1079    info = _service_info(COMPANION_DISCOVERY_TYPE)
1080
1081    await provider._handle_companion_service_state_change(
1082        info.name,
1083        ServiceStateChange.Added,
1084        info,
1085    )
1086    set_companion_info.assert_awaited_once_with(info)
1087
1088    await provider._handle_companion_service_state_change(
1089        info.name,
1090        ServiceStateChange.Removed,
1091        None,
1092    )
1093    set_companion_info.assert_awaited_once_with(info)
1094    assert provider._companion_info_by_address["192.168.1.10"] is info
1095
1096
1097async def test_control_service_cache_prunes_addresses_no_longer_advertised() -> None:
1098    """A control service that moves to a new address leaves no stale cache entry."""
1099    provider = AirPlayProvider.__new__(AirPlayProvider)
1100    provider.mass = MagicMock()
1101    provider._companion_info_by_address = {}
1102    provider.get_players = MagicMock(return_value=[])  # type: ignore[method-assign]
1103    info = _service_info(COMPANION_DISCOVERY_TYPE)
1104    other_info = _service_info(COMPANION_DISCOVERY_TYPE, address="192.168.1.60")
1105    other_info.name = f"other.{COMPANION_DISCOVERY_TYPE}"
1106    moved_info = _service_info(COMPANION_DISCOVERY_TYPE, address="192.168.1.40")
1107
1108    for service_info in (info, other_info, moved_info):
1109        await provider._handle_companion_service_state_change(
1110            service_info.name,
1111            ServiceStateChange.Updated,
1112            service_info,
1113        )
1114
1115    assert "192.168.1.10" not in provider._companion_info_by_address
1116    assert provider._companion_info_by_address["192.168.1.40"] is moved_info
1117    assert provider._companion_info_by_address["192.168.1.60"] is other_info
1118
1119
1120async def test_mrp_discovery_is_attached_by_shared_address() -> None:
1121    """A native MRP service is attached to its controlled AirPlay endpoint."""
1122    provider = AirPlayProvider.__new__(AirPlayProvider)
1123    provider.mass = MagicMock()
1124    provider._mrp_info_by_address = {}
1125    player = _make_control_player(companion_flags=None)
1126    set_mrp_info = AsyncMock()
1127    player.set_mrp_discovery_info = set_mrp_info  # type: ignore[method-assign]
1128    provider.get_players = MagicMock(return_value=[player])  # type: ignore[method-assign]
1129    info = _service_info(MRP_DISCOVERY_TYPE)
1130
1131    await provider._handle_mrp_service_state_change(
1132        info.name,
1133        ServiceStateChange.Added,
1134        info,
1135    )
1136
1137    set_mrp_info.assert_awaited_once_with(info)
1138    assert provider._mrp_info_by_address["192.168.1.10"] is info
1139
1140
1141async def test_late_companion_discovery_never_changes_player_model() -> None:
1142    """A control service appearing for a generic endpoint never changes its model."""
1143    provider = AirPlayProvider.__new__(AirPlayProvider)
1144    provider.mass = MagicMock()
1145    provider.logger = logging.getLogger("test.airplay.provider")
1146    provider.config = MagicMock()
1147    provider.config.instance_id = "airplay"
1148    provider._companion_info_by_address = {}
1149    # a bare provider (built via __new__) lacks the bridge manager that
1150    # supported_features consults; a real one exists before any player does
1151    provider._bridge_manager = MagicMock(get_bridge=MagicMock(return_value=None))
1152    generic_player = GenericAirPlayPlayer(
1153        provider=provider,
1154        player_id=PLAYER_ID,
1155        raop_discovery_info=None,
1156        airplay_discovery_info=_service_info(AIRPLAY_DISCOVERY_TYPE),
1157        address="192.168.1.10",
1158        display_name="Controlled receiver",
1159        manufacturer="Receiver",
1160        model="Model",
1161    )
1162    provider.get_players = MagicMock(return_value=[generic_player])  # type: ignore[method-assign]
1163    provider.mass.players.unregister = AsyncMock()
1164    provider.mass.players.register = AsyncMock()
1165    info = _service_info(
1166        COMPANION_DISCOVERY_TYPE,
1167        properties={"rpFl": "0x367A2"},
1168    )
1169
1170    await provider._handle_companion_service_state_change(
1171        info.name,
1172        ServiceStateChange.Added,
1173        info,
1174    )
1175
1176    provider.mass.players.unregister.assert_not_awaited()
1177    provider.mass.players.register.assert_not_awaited()
1178
1179
1180async def test_generic_device_setup_skips_control_discovery() -> None:
1181    """A non-Apple receiver registers without Companion/MRP service lookups."""
1182    provider = AirPlayProvider.__new__(AirPlayProvider)
1183    provider.mass = MagicMock()
1184    provider.logger = logging.getLogger("test.airplay.provider")
1185    provider.config = MagicMock()
1186    provider.config.instance_id = "airplay"
1187    provider._bridge_manager = MagicMock()
1188    provider._bridge_manager.evaluate_bridge = AsyncMock()
1189    provider.dashboards = MagicMock()
1190    provider._companion_info_by_address = {}
1191    provider._mrp_info_by_address = {}
1192    provider.mass.discovery.async_find_mdns_service = AsyncMock(return_value=None)
1193    provider.mass.players.register = AsyncMock()
1194    provider.mass.players.get_player.return_value = None
1195    provider.mass.config.get_raw_player_config_value.side_effect = (
1196        lambda _player_id, key, default=None: True if key == "enabled" else default
1197    )
1198    info = _service_info(
1199        AIRPLAY_DISCOVERY_TYPE,
1200        properties={"model": "ThirdPartyReceiver1,1", "features": "0x0"},
1201    )
1202
1203    with (
1204        patch(
1205            "music_assistant.providers.airplay.provider.get_primary_ip_address_from_zeroconf",
1206            return_value="192.168.1.10",
1207        ),
1208        patch(
1209            "music_assistant.providers.airplay.provider.get_model_info",
1210            return_value=("Receiver", "Generic"),
1211        ),
1212    ):
1213        await provider._setup_player("player", "Player", info)
1214
1215    player = provider.mass.players.register.await_args.args[0]
1216    assert isinstance(player, GenericAirPlayPlayer)
1217    control_lookups = [
1218        lookup_call
1219        for lookup_call in provider.mass.discovery.async_find_mdns_service.await_args_list
1220        if lookup_call.args
1221        and lookup_call.args[0] in (COMPANION_DISCOVERY_TYPE, MRP_DISCOVERY_TYPE)
1222    ]
1223    assert not control_lookups
1224
1225
1226async def test_related_discovery_finds_differently_named_cached_service() -> None:
1227    """Control discovery can match a cached service by address instead of name."""
1228    provider = AirPlayProvider.__new__(AirPlayProvider)
1229    provider.mass = MagicMock()
1230    provider.mass.discovery.aiozc.zeroconf.cache.cache = {
1231        "_companion-link._tcp.local.": {},
1232        "uuid._companion-link._tcp.local.": {},
1233    }
1234    info = _service_info(COMPANION_DISCOVERY_TYPE)
1235    info.async_request = AsyncMock(return_value=True)
1236
1237    with patch(
1238        "music_assistant.providers.airplay.provider.AsyncServiceInfo",
1239        return_value=info,
1240    ) as service_info:
1241        result = await provider._find_cached_discovery_info(
1242            COMPANION_DISCOVERY_TYPE,
1243            {"192.168.1.10"},
1244        )
1245
1246    assert result is info
1247    service_info.assert_called_once_with(
1248        COMPANION_DISCOVERY_TYPE,
1249        "uuid._companion-link._tcp.local.",
1250    )
1251
1252
1253async def test_list_installed_app_ids_none_when_feature_unavailable() -> None:
1254    """No AppList feature available on the device returns None."""
1255    player = _make_control_player()
1256    with patch.object(player, "_device_for_feature", return_value=None):
1257        assert await player.async_list_installed_app_ids() is None
1258
1259
1260async def test_list_installed_app_ids_none_on_command_error() -> None:
1261    """A control-channel error while listing apps returns None."""
1262    player = _make_control_player()
1263    device = MagicMock()
1264    device.apps.app_list = AsyncMock(side_effect=pyatv_exceptions.CommandError("boom"))
1265    with patch.object(player, "_device_for_feature", return_value=device):
1266        assert await player.async_list_installed_app_ids() is None
1267
1268
1269async def test_list_installed_app_ids_returns_bundle_ids() -> None:
1270    """A successful app list returns the set of installed bundle ids."""
1271    player = _make_control_player()
1272    device = MagicMock()
1273    device.apps.app_list = AsyncMock(return_value=[App("App A", "com.a"), App("App B", "com.b")])
1274    with patch.object(player, "_device_for_feature", return_value=device):
1275        assert await player.async_list_installed_app_ids() == {"com.a", "com.b"}
1276
1277
1278async def test_launch_app_raises_when_feature_unavailable() -> None:
1279    """Launching with no LaunchApp feature raises PlayerCommandFailed."""
1280    player = _make_control_player()
1281    with (
1282        patch.object(player, "_device_for_feature", return_value=None),
1283        pytest.raises(PlayerCommandFailed),
1284    ):
1285        await player.async_launch_app("musicassistant://dashboard/show")
1286
1287
1288async def test_launch_app_wraps_command_error() -> None:
1289    """A control-channel error while launching raises PlayerCommandFailed."""
1290    player = _make_control_player()
1291    device = MagicMock()
1292    device.apps.launch_app = AsyncMock(side_effect=pyatv_exceptions.CommandError("nope"))
1293    with (
1294        patch.object(player, "_device_for_feature", return_value=device),
1295        pytest.raises(PlayerCommandFailed),
1296    ):
1297        await player.async_launch_app("musicassistant://dashboard/show")
1298
1299
1300async def test_launch_app_dispatches_to_device() -> None:
1301    """Launching on an available device dispatches the value over Companion."""
1302    player = _make_control_player()
1303    device = MagicMock()
1304    device.apps.launch_app = AsyncMock()
1305    with patch.object(player, "_device_for_feature", return_value=device):
1306        await player.async_launch_app("musicassistant://dashboard/show")
1307    device.apps.launch_app.assert_awaited_once_with("musicassistant://dashboard/show")
1308