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