music-assistant-server

1"""Unit tests for the AirPlay provider."""
2
3import asyncio
4import logging
5import time
6from contextlib import AbstractContextManager
7from typing import TYPE_CHECKING, cast
8from unittest.mock import AsyncMock, MagicMock, patch
9
10from music_assistant_models.enums import PlaybackState
11
12from music_assistant.constants import VERBOSE_LOG_LEVEL
13from music_assistant.providers.airplay.constants import (
14    AIRPLAY_COLD_GROUP_START_LEAD_MS,
15    PTP_DAEMON_WARN_BURST,
16    PTP_DAEMON_WARN_WINDOW,
17    AirPlayRemoteCommand,
18    ClockReadiness,
19    StreamingProtocol,
20)
21from music_assistant.providers.airplay.player import AirPlayPlayer
22from music_assistant.providers.airplay.provider import ENV_PYATV_DEBUG, AirPlayProvider
23from music_assistant.providers.airplay.sendspin_bridge import (
24    SendspinAirPlayBridge,
25    SendspinBridgeManager,
26)
27from music_assistant.providers.airplay.stream import AirPlayStream
28from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
29from music_assistant.providers.airplay_receiver import airplay_receiver_port
30
31if TYPE_CHECKING:
32    import pytest
33
34INSTANCE_ID = "airplay"
35START_UNIX_MS = 1_750_000_000_000
36
37# A representative readiness line emitted by `cliairplay --ptp-daemon` on the
38# real device (from a live DEBUG log) once it has bound 319/320 and opened its
39# control channel.
40DAEMON_UP_LINE = (
41    "[15:44:56.101] ap2_ptp_run_daemon:1467 [PTP] daemon up: engine on 319/320, "
42    "clock in /cliairplay-ptp, control on 127.0.0.1:9010"
43)
44
45
46def _make_provider() -> AirPlayProvider:
47    """Build a bare provider wired to a mocked Music Assistant instance."""
48    prov = AirPlayProvider.__new__(AirPlayProvider)
49    prov.mass = MagicMock()
50    prov.logger = logging.getLogger("test.airplay.provider")
51    prov.config = MagicMock()
52    prov.config.instance_id = INSTANCE_ID
53    return prov
54
55
56def test_remote_pause_routes_to_sendspin_session() -> None:
57    """A bridged device pause targets the owning Sendspin session."""
58    prov = _make_provider()
59    players = cast("MagicMock", prov.mass.players)
60    manager = SendspinBridgeManager(prov)
61    prov._bridge_manager = manager
62
63    airplay_player = MagicMock()
64    airplay_player.player_id = "apc43875e9e53a"
65    airplay_player.playback_state = PlaybackState.PLAYING
66
67    bridge = SendspinAirPlayBridge(prov, airplay_player, MagicMock())
68    bridge._bridge_client_id = "sendspin-bridge"
69    bridge._airplay_stream = airplay_player.stream = MagicMock()
70    manager._bridges[airplay_player.player_id] = bridge
71
72    bridge_player = MagicMock()
73    bridge_player.synced_to = "sendspin-leader"
74    bridge_player.protocol_parent_id = airplay_player.player_id
75    bridge_player.player_id = "sendspin-bridge"
76
77    sync_leader = MagicMock()
78    sync_leader.protocol_parent_id = "sendspin-session"
79    sync_leader.player_id = "sendspin-leader"
80    players.get_player.side_effect = {
81        "sendspin-bridge": bridge_player,
82        "sendspin-leader": sync_leader,
83    }.get
84
85    prov.handle_remote_command(airplay_player, AirPlayRemoteCommand.PAUSE)
86
87    players.cmd_pause.assert_called_once_with("sendspin-session")
88
89
90def test_remote_pause_keeps_normal_airplay_routing() -> None:
91    """An inactive bridge does not change normal AirPlay pause routing."""
92    prov = _make_provider()
93    players = cast("MagicMock", prov.mass.players)
94    manager = SendspinBridgeManager(prov)
95    prov._bridge_manager = manager
96    airplay_player = MagicMock()
97    airplay_player.player_id = "apc43875e9e53a"
98    airplay_player.playback_state = PlaybackState.PLAYING
99    airplay_player.stream = MagicMock()
100
101    bridge = SendspinAirPlayBridge(prov, airplay_player, MagicMock())
102    bridge._bridge_client_id = "sendspin-bridge"
103    manager._bridges[airplay_player.player_id] = bridge
104
105    prov.handle_remote_command(airplay_player, AirPlayRemoteCommand.PAUSE)
106
107    players.cmd_pause.assert_called_once_with(airplay_player.player_id)
108
109
110async def test_unload_awaits_cancelled_ptp_stdout_reader() -> None:
111    """Provider unload waits for its cancelled PTP stdout reader before process cleanup."""
112    prov = AirPlayProvider.__new__(AirPlayProvider)
113    prov.mass = MagicMock()
114    prov._ptp_daemon_ready = None
115    prov._ptp_daemon = None
116    prov._dacp_server = MagicMock()
117    prov._dacp_server.__bool__.return_value = False
118    prov._dacp_info = MagicMock()
119    prov._dacp_info.__bool__.return_value = False
120    reader_started = asyncio.Event()
121
122    async def _stdout_reader() -> None:
123        reader_started.set()
124        await asyncio.Event().wait()
125
126    reader_task = asyncio.create_task(_stdout_reader())
127    prov._ptp_daemon_stdout_task = reader_task
128    await reader_started.wait()
129
130    await prov.unload()
131
132    assert reader_task.cancelled()
133
134
135# --- Shared PTP clock daemon: readiness signal ---------------------------------
136
137
138def _ptp_provider() -> AirPlayProvider:
139    """Build a bare provider with just a logger for PTP readiness tests."""
140    prov = AirPlayProvider.__new__(AirPlayProvider)
141    prov.logger = logging.getLogger("test.airplay.provider")
142    prov._ptp_daemon = None
143    prov._ptp_daemon_ready = None
144    return prov
145
146
147def _live_ptp_daemon() -> MagicMock:
148    """Build a process mock that remains alive until its wait task is cancelled."""
149    daemon = MagicMock()
150    daemon.closed = False
151    daemon.returncode = None
152
153    async def _wait_forever() -> int:
154        await asyncio.Event().wait()
155        return 0
156
157    daemon.wait = AsyncMock(side_effect=_wait_forever)
158    return daemon
159
160
161async def test_ptp_daemon_spawn_advertises_dacp_identity() -> None:
162    """The spawned PTP daemon is handed the provider's DACP id as its clock identity."""
163    prov = _ptp_provider()
164    prov.dacp_id = "AABBCCDD11223344"
165    prov.mass = MagicMock()
166    prov.mass.streams.get_source_ip = AsyncMock(return_value=None)
167    # The spawn mirrors the live logger level into --debug flags; pin the
168    # level so suite ordering (other tests raising verbosity) cannot leak
169    # extra args into this assertion.
170    prov.logger.setLevel(logging.INFO)
171
172    def _consume_task(coro: object) -> MagicMock:
173        if asyncio.iscoroutine(coro):
174            coro.close()
175        return MagicMock()
176
177    prov.mass.create_task.side_effect = _consume_task
178
179    with (
180        patch(
181            "music_assistant.providers.airplay.provider.get_cli_binary",
182            AsyncMock(return_value="/bin/cliairplay"),
183        ),
184        patch("music_assistant.providers.airplay.provider.AsyncProcess") as process_cls,
185    ):
186        process_cls.return_value.start = AsyncMock(return_value=None)
187        await prov._start_ptp_daemon()
188
189    assert process_cls.call_args.args[0] == [
190        "/bin/cliairplay",
191        "--ptp-daemon",
192        "--dacp",
193        "AABBCCDD11223344",
194    ]
195
196
197async def test_ptp_daemon_spawn_pins_operator_configured_interface() -> None:
198    """One daemon serves every player, so only an explicit operator pin narrows it."""
199    prov = _ptp_provider()
200    prov.dacp_id = "AABBCCDD11223344"
201    prov.mass = MagicMock()
202    prov.mass.streams.get_source_ip = AsyncMock(return_value="192.168.1.5")
203    prov.logger.setLevel(logging.INFO)
204
205    def _consume_task(coro: object) -> MagicMock:
206        if asyncio.iscoroutine(coro):
207            coro.close()
208        return MagicMock()
209
210    prov.mass.create_task.side_effect = _consume_task
211
212    with (
213        patch(
214            "music_assistant.providers.airplay.provider.get_cli_binary",
215            AsyncMock(return_value="/bin/cliairplay"),
216        ),
217        patch("music_assistant.providers.airplay.provider.AsyncProcess") as process_cls,
218    ):
219        process_cls.return_value.start = AsyncMock(return_value=None)
220        await prov._start_ptp_daemon()
221
222    args = process_cls.call_args.args[0]
223    assert args[args.index("--if") + 1] == "192.168.1.5"
224    prov.mass.streams.get_source_ip.assert_awaited_once_with()
225
226
227async def test_ptp_daemon_spawn_quiet_at_debug_level() -> None:
228    """A normal debug session keeps the daemon quiet (no per-packet tracing)."""
229    prov = _ptp_provider()
230    prov.dacp_id = "AABBCCDD11223344"
231    prov.mass = MagicMock()
232    prov.mass.streams.get_source_ip = AsyncMock(return_value=None)
233    prov.logger.setLevel(logging.DEBUG)
234
235    def _consume_task(coro: object) -> MagicMock:
236        if asyncio.iscoroutine(coro):
237            coro.close()
238        return MagicMock()
239
240    prov.mass.create_task.side_effect = _consume_task
241
242    with (
243        patch(
244            "music_assistant.providers.airplay.provider.get_cli_binary",
245            AsyncMock(return_value="/bin/cliairplay"),
246        ),
247        patch("music_assistant.providers.airplay.provider.AsyncProcess") as process_cls,
248    ):
249        process_cls.return_value.start = AsyncMock(return_value=None)
250        await prov._start_ptp_daemon()
251
252    assert "--debug" not in process_cls.call_args.args[0]
253
254
255async def test_ptp_daemon_spawn_traces_at_verbose_level() -> None:
256    """The daemon's per-packet trace needs verbose logging AND the opt-in toggle."""
257    prov = _ptp_provider()
258    prov.dacp_id = "AABBCCDD11223344"
259    prov.mass = MagicMock()
260    prov.mass.streams.get_source_ip = AsyncMock(return_value=None)
261    prov.logger.setLevel(VERBOSE_LOG_LEVEL)
262    prov.config = MagicMock()
263    prov.config.get_value = MagicMock(return_value=True)
264
265    def _consume_task(coro: object) -> MagicMock:
266        if asyncio.iscoroutine(coro):
267            coro.close()
268        return MagicMock()
269
270    prov.mass.create_task.side_effect = _consume_task
271
272    with (
273        patch(
274            "music_assistant.providers.airplay.provider.get_cli_binary",
275            AsyncMock(return_value="/bin/cliairplay"),
276        ),
277        patch("music_assistant.providers.airplay.provider.AsyncProcess") as process_cls,
278    ):
279        process_cls.return_value.start = AsyncMock(return_value=None)
280        await prov._start_ptp_daemon()
281
282    assert process_cls.call_args.args[0][-2:] == ["--debug", "10"]
283
284    # Without the opt-in the daemon stays quiet even on a verbose session
285    # (its ~10 lines/s timing trace would otherwise flood every verbose log).
286    prov.config.get_value = MagicMock(return_value=False)
287    with (
288        patch(
289            "music_assistant.providers.airplay.provider.get_cli_binary",
290            AsyncMock(return_value="/bin/cliairplay"),
291        ),
292        patch("music_assistant.providers.airplay.provider.AsyncProcess") as process_cls,
293    ):
294        process_cls.return_value.start = AsyncMock(return_value=None)
295        await prov._start_ptp_daemon()
296    assert "--debug" not in process_cls.call_args.args[0]
297
298
299def test_pyatv_logging_quiet_at_debug_level() -> None:
300    """A normal debug session keeps pyatv's own (very chatty) logging quiet."""
301    prov = _ptp_provider()
302    prov.logger.setLevel(logging.DEBUG)
303
304    prov._set_pyatv_log_level()
305
306    # pyatv debug output is dropped; only INFO and above pass through
307    assert logging.getLogger("pyatv").level == logging.INFO
308
309
310def test_pyatv_logging_quiet_at_verbose_level() -> None:
311    """A verbose session keeps pyatv's protocol chatter out of the log too."""
312    prov = _ptp_provider()
313    prov.logger.setLevel(VERBOSE_LOG_LEVEL)
314
315    prov._set_pyatv_log_level()
316
317    # verbose is there to surface our own diagnostics, not pyatv's protocol dumps
318    assert logging.getLogger("pyatv").level == logging.INFO
319
320
321def test_pyatv_logging_traces_with_opt_in(monkeypatch: pytest.MonkeyPatch) -> None:
322    """The dedicated opt-in releases pyatv's own debug logging."""
323    prov = _ptp_provider()
324    prov.logger.setLevel(VERBOSE_LOG_LEVEL)
325    monkeypatch.setenv(ENV_PYATV_DEBUG, "1")
326
327    prov._set_pyatv_log_level()
328
329    assert logging.getLogger("pyatv").level == logging.DEBUG
330
331
332def test_ptp_daemon_ready_event_set_on_daemon_up_line() -> None:
333    """The readiness event is set when the daemon prints its 'daemon up' line."""
334    prov = _ptp_provider()
335    prov._ptp_daemon_ready = asyncio.Event()
336
337    prov._handle_ptp_daemon_line(DAEMON_UP_LINE)
338
339    assert prov._ptp_daemon_ready.is_set()
340
341
342def test_ptp_daemon_line_without_marker_leaves_event_clear() -> None:
343    """A normal daemon log line does not trip the readiness event."""
344    prov = _ptp_provider()
345    prov._ptp_daemon_ready = asyncio.Event()
346
347    prov._handle_ptp_daemon_line(
348        "[15:44:56.101] ap2_ptp_engine_start:1173 [PTP] Engine started on UDP 319/320"
349    )
350
351    assert not prov._ptp_daemon_ready.is_set()
352
353
354def test_ptp_daemon_line_handler_tolerates_no_event() -> None:
355    """The line handler is a no-op for readiness before any daemon has started."""
356    prov = _ptp_provider()  # _ptp_daemon_ready is None
357    # Must not raise even when the readiness gate does not exist yet.
358    prov._handle_ptp_daemon_line(DAEMON_UP_LINE)
359
360
361def test_ptp_daemon_problem_lines_are_rate_limited(caplog: pytest.LogCaptureFixture) -> None:
362    """A repeating trace line matching a marker must not fill the log at WARNING."""
363    prov = _ptp_provider()
364    trace = "[15:44:56.101] [PTP] slave offset seq=7 error=0.000012"
365
366    with caplog.at_level(logging.WARNING):
367        for _ in range(PTP_DAEMON_WARN_BURST + 20):
368            prov._handle_ptp_daemon_line(trace)
369
370    warnings = [record for record in caplog.records if record.levelno == logging.WARNING]
371    assert len(warnings) == PTP_DAEMON_WARN_BURST
372
373
374def test_ptp_daemon_reports_what_it_suppressed(
375    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
376) -> None:
377    """The count of suppressed lines is reported once the window rolls over."""
378    prov = _ptp_provider()
379    trace = "[15:44:56.101] [PTP] slave offset seq=7 error=0.000012"
380    clock = 1000.0
381    monkeypatch.setattr("music_assistant.providers.airplay.provider.time.monotonic", lambda: clock)
382
383    with caplog.at_level(logging.WARNING):
384        for _ in range(PTP_DAEMON_WARN_BURST + 3):
385            prov._handle_ptp_daemon_line(trace)
386        caplog.clear()
387        clock += PTP_DAEMON_WARN_WINDOW + 1
388        prov._handle_ptp_daemon_line(trace)
389
390    messages = [record.getMessage() for record in caplog.records]
391    assert any("3 further problem line(s) were suppressed" in message for message in messages)
392    # and the window reopens for real problems
393    assert any(message.endswith(trace) for message in messages)
394
395
396def test_ptp_daemon_warn_window_runs_from_its_first_line(
397    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
398) -> None:
399    """time.monotonic() counts from boot, so a server started early must still get a full window."""
400    prov = _ptp_provider()
401    trace = "[15:44:56.101] [PTP] slave offset seq=7 error=0.000012"
402    clock = 5.0  # five seconds of uptime
403    monkeypatch.setattr("music_assistant.providers.airplay.provider.time.monotonic", lambda: clock)
404    for _ in range(PTP_DAEMON_WARN_BURST + 2):
405        prov._handle_ptp_daemon_line(trace)
406
407    # 56s into the window, so it must not have rolled over yet
408    clock += PTP_DAEMON_WARN_WINDOW - 4
409    caplog.clear()
410    with caplog.at_level(logging.WARNING):
411        prov._handle_ptp_daemon_line(trace)
412
413    assert caplog.text == ""
414
415
416def test_ptp_daemon_restart_gets_a_fresh_warning_budget(
417    caplog: pytest.LogCaptureFixture,
418) -> None:
419    """A replacement daemon's startup failure must not be eaten by the old one's budget."""
420    prov = _ptp_provider()
421    for _ in range(PTP_DAEMON_WARN_BURST + 2):
422        prov._handle_ptp_daemon_line("[15:44:56.101] [PTP] slave offset seq=7 error=0.000012")
423
424    caplog.clear()
425    with caplog.at_level(logging.WARNING):
426        prov._reset_ptp_daemon_warn_budget()
427        prov._handle_ptp_daemon_line("[15:44:57.002] [PTP] Cannot bind UDP 319: Permission denied")
428
429    messages = [record.getMessage() for record in caplog.records]
430    assert any("2 further problem line(s) from the previous daemon" in m for m in messages)
431    assert any("Cannot bind UDP 319" in m for m in messages)
432
433
434async def test_diagnostics_report_daemon_readiness_and_per_stream_route() -> None:
435    """A live daemon that never bound its ports must not read as healthy timing."""
436    prov = _ptp_provider()
437    prov._dacp_server = MagicMock(is_serving=MagicMock(return_value=True))
438    prov._ptp_daemon = _live_ptp_daemon()
439    prov._ptp_daemon_ready = asyncio.Event()  # spawned, never reported ready
440    ntp_player = MagicMock(protocol=StreamingProtocol.AIRPLAY2)
441    ntp_player.stream = MagicMock(running=True, active_route="AirPlay 2 (buffered, NTP)")
442    raop_player = MagicMock(protocol=StreamingProtocol.RAOP)
443    raop_player.stream = MagicMock(running=True, active_route="RAOP")
444
445    with patch.object(prov, "get_players", return_value=[ntp_player, raop_player]):
446        diagnostics = await prov.get_diagnostics()
447
448    assert diagnostics["ptp_daemon_running"] is True
449    assert diagnostics["ptp_daemon_ready"] is False
450    assert diagnostics["streams_by_route"] == {"AirPlay 2 (buffered, NTP)": 1, "RAOP": 1}
451
452
453async def test_ptp_daemon_bind_failure_degrades_without_restart() -> None:
454    """
455    Verify a privileged-port bind failure leaves streams on NTP timing.
456
457    The provider must clear readiness without restarting an immediate bind failure.
458    """
459    prov = _ptp_provider()
460    daemon = MagicMock()
461    daemon.wait = AsyncMock(return_value=2)
462    ready = asyncio.Event()
463    ready.set()
464    prov.logger = MagicMock()
465    prov._ptp_daemon = daemon
466    prov._ptp_daemon_ready = ready
467    prov._ptp_daemon_stop_requested = False
468    prov._ptp_daemon_started = time.monotonic()
469    prov._ptp_daemon_restarted = False
470
471    with patch.object(prov, "_start_ptp_daemon", new_callable=AsyncMock) as restart:
472        await prov._ptp_daemon_monitor(daemon)
473
474    restart.assert_not_awaited()
475    assert not await prov.wait_ptp_daemon_ready(timeout=0)
476    warning = prov.logger.warning
477    warning.assert_called_once()
478    assert warning.call_args.args[1] == 2
479    assert "CAP_NET_BIND_SERVICE" in warning.call_args.args[0]
480
481
482async def test_wait_ptp_daemon_ready_true_when_signalled() -> None:
483    """wait_ptp_daemon_ready returns True once readiness has been signalled."""
484    prov = _ptp_provider()
485    prov._ptp_daemon = _live_ptp_daemon()
486    prov._ptp_daemon_ready = asyncio.Event()
487    prov._ptp_daemon_ready.set()
488
489    assert await prov.wait_ptp_daemon_ready(timeout=0.1) is True
490
491
492async def test_wait_ptp_daemon_ready_false_without_daemon() -> None:
493    """With no daemon ever started (event is None), readiness is False."""
494    prov = _ptp_provider()
495
496    assert await prov.wait_ptp_daemon_ready(timeout=0.1) is False
497
498
499async def test_wait_ptp_daemon_ready_times_out_when_never_ready() -> None:
500    """A spawned-but-not-ready daemon yields False after the bounded wait."""
501    prov = _ptp_provider()
502    prov._ptp_daemon = _live_ptp_daemon()
503    prov._ptp_daemon_ready = asyncio.Event()  # but readiness never signalled
504
505    assert await prov.wait_ptp_daemon_ready(timeout=0.05) is False
506
507
508async def test_wait_ptp_daemon_ready_false_fast_when_daemon_gone() -> None:
509    """A daemon that has exited short-circuits to False without burning the timeout."""
510    prov = _ptp_provider()
511    prov._ptp_daemon = None  # process has exited
512    prov._ptp_daemon_ready = asyncio.Event()  # stale gate left un-set
513
514    started = time.monotonic()
515    assert await prov.wait_ptp_daemon_ready(timeout=5.0) is False
516    assert time.monotonic() - started < 1.0
517
518
519async def test_wait_ptp_daemon_ready_false_fast_when_process_exited() -> None:
520    """An exited process object cannot keep readiness waiting on a stale event."""
521    prov = _ptp_provider()
522    prov._ptp_daemon = MagicMock(closed=True, returncode=1)
523    prov._ptp_daemon_ready = asyncio.Event()
524
525    started = time.monotonic()
526    assert await prov.wait_ptp_daemon_ready(timeout=5.0) is False
527    assert time.monotonic() - started < 1.0
528
529
530async def test_wait_ptp_daemon_ready_rejects_stale_ready_event() -> None:
531    """A stale readiness signal cannot make an exited daemon appear usable."""
532    prov = _ptp_provider()
533    prov._ptp_daemon = MagicMock(closed=True, returncode=1)
534    prov._ptp_daemon_ready = asyncio.Event()
535    prov._ptp_daemon_ready.set()
536
537    assert await prov.wait_ptp_daemon_ready(timeout=0.1) is False
538
539
540async def test_wait_ptp_daemon_ready_signalled_during_wait() -> None:
541    """A readiness signal that arrives while waiting resolves the wait to True."""
542    prov = _ptp_provider()
543    prov._ptp_daemon = _live_ptp_daemon()
544    prov._ptp_daemon_ready = asyncio.Event()
545
546    async def _signal_soon() -> None:
547        await asyncio.sleep(0.02)
548        prov._handle_ptp_daemon_line(DAEMON_UP_LINE)
549
550    signaller = asyncio.create_task(_signal_soon())
551    try:
552        assert await prov.wait_ptp_daemon_ready(timeout=1.0) is True
553    finally:
554        await signaller
555
556
557# --- Session-wide PTP timing decision ------------------------------------------
558
559
560def _ap2_player() -> MagicMock:
561    """Return a mock player that resolves to native AirPlay 2."""
562    player = MagicMock()
563    player.protocol = StreamingProtocol.AIRPLAY2
564    return player
565
566
567def _raop_player() -> MagicMock:
568    """Return a mock player that resolves to legacy RAOP."""
569    player = MagicMock()
570    player.protocol = StreamingProtocol.RAOP
571    return player
572
573
574async def _ack_commanded_instant(start_unix_ms: int = 0, *_args: object, **_kwargs: object) -> int:
575    """Ack a START at exactly the instant it was commanded, as a feasible one is."""
576    return start_unix_ms
577
578
579def _make_ptp_session(prov: MagicMock, sync_clients: list[MagicMock]) -> AirPlayStreamSession:
580    """Build a stream session wired to a mock provider for PTP-decision tests."""
581    pcm_format = MagicMock()
582    return AirPlayStreamSession(
583        prov, cast("list[AirPlayPlayer]", sync_clients), pcm_format, MagicMock()
584    )
585
586
587async def test_session_resolves_shared_ptp_when_daemon_ready() -> None:
588    """A ready daemon makes the whole session opt into shared PTP, no warning."""
589    prov = MagicMock()
590    prov.wait_ptp_daemon_ready = AsyncMock(return_value=True)
591    session = _make_ptp_session(prov, [_ap2_player(), _ap2_player()])
592
593    assert await session._resolve_shared_ptp() is True
594    prov.logger.warning.assert_not_called()
595
596
597async def test_session_degrades_and_warns_for_group_when_not_ready() -> None:
598    """A not-ready daemon degrades a group to no shared PTP and warns once."""
599    prov = MagicMock()
600    prov.wait_ptp_daemon_ready = AsyncMock(return_value=False)
601    session = _make_ptp_session(prov, [_ap2_player(), _ap2_player()])
602
603    assert await session._resolve_shared_ptp() is False
604    prov.logger.warning.assert_called_once()
605
606
607async def test_session_single_ap2_player_not_ready_does_not_warn() -> None:
608    """A lone native AP2 player degrades silently: self-bind is fine with no partner."""
609    prov = MagicMock()
610    prov.wait_ptp_daemon_ready = AsyncMock(return_value=False)
611    session = _make_ptp_session(prov, [_ap2_player()])
612
613    assert await session._resolve_shared_ptp() is False
614    prov.logger.warning.assert_not_called()
615
616
617async def test_session_skips_ptp_wait_for_raop_members() -> None:
618    """A RAOP-only session does not spend playback lead waiting for PTP."""
619    prov = MagicMock()
620    prov.wait_ptp_daemon_ready = AsyncMock(return_value=True)
621    session = _make_ptp_session(prov, [_raop_player(), _raop_player()])
622
623    assert await session._resolve_shared_ptp() is False
624    prov.wait_ptp_daemon_ready.assert_not_awaited()
625
626
627async def test_raop_session_resolves_ptp_for_first_ap2_late_joiner() -> None:
628    """A deferred AP2 join still commits the session to the shared PTP clock."""
629    prov = MagicMock()
630    raop_player = _raop_player()
631    raop_player.player_id = "raop"
632    raop_player.playback_state = PlaybackState.PLAYING
633    raop_player.stream = MagicMock()
634    raop_player.stream.running = True
635    raop_player.stream.cumulative_shift_seconds = 0.0
636    ap2_player = _ap2_player()
637    ap2_player.player_id = "airplay2"
638    ap2_player.stream = None
639    session = _make_ptp_session(prov, [raop_player])
640    pcm_format = cast("MagicMock", session.pcm_format)
641    pcm_format.pcm_sample_size = 176_400
642    pcm_format.bit_depth = 16
643    pcm_format.channels = 2
644    session.start_time = time.time() - 5
645    session.seconds_streamed = 5
646
647    async def _wait_ptp_daemon_ready() -> bool:
648        assert not session._lock.locked()
649        return True
650
651    prov.wait_ptp_daemon_ready = AsyncMock(side_effect=_wait_ptp_daemon_ready)
652
653    async def _start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
654        player.stream = MagicMock(running=True, connected=True)
655        player.stream.wait_for_connection = AsyncMock()
656        player.stream.flush = AsyncMock(return_value=True)
657        # Verified-start API defaults: every START is acked at the commanded
658        # instant, with no warm-lead constraint and no receiver clock
659        # projection, so the test asserts the commanded values directly.
660        player.stream.start = AsyncMock(side_effect=_ack_commanded_instant)
661        player.stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.UNREPORTED, 0))
662        player.stream.warm_lead_ms = 0
663        player.stream.flushed_head_unix_ms = 0
664
665    with patch.object(
666        session, "_start_client", new_callable=AsyncMock, side_effect=_start_client
667    ) as mock_start:
668        await session.add_client(cast("AirPlayPlayer", ap2_player))
669
670    prov.wait_ptp_daemon_ready.assert_awaited_once()
671    assert session.use_shared_ptp is True
672    assert session._shared_ptp_resolved is True
673    await_args = mock_start.await_args
674    assert await_args is not None
675    assert await_args.args[1] is True
676
677
678async def test_session_start_applies_uniform_ptp_decision_to_all_members() -> None:
679    """start() resolves the timing source once and passes it identically to every member."""
680    prov = MagicMock()
681    prov.wait_ptp_daemon_ready = AsyncMock(return_value=True)
682    players = [_ap2_player(), _ap2_player(), _ap2_player()]
683    for player in players:
684        player.stream = MagicMock()
685        player.stream.wait_for_connection = AsyncMock()
686        player.stream.wait_audio_present = AsyncMock(return_value=True)
687        player.stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.UNREPORTED, 0))
688        player.stream.start = AsyncMock(side_effect=_ack_commanded_instant)
689    session = _make_ptp_session(prov, players)
690
691    with (
692        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
693        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
694    ):
695        await session.start(MagicMock())
696
697    # One start per member, and every member received the same resolved decision.
698    assert mock_start.call_count == len(players)
699    ptp_decisions = {call.args[1] for call in mock_start.call_args_list}
700    assert ptp_decisions == {True}
701    assert session.use_shared_ptp is True
702
703
704async def test_session_start_calculates_anchor_after_ptp_resolution() -> None:
705    """PTP startup time cannot consume the audible setup lead."""
706    prov = MagicMock()
707    players = [_ap2_player(), _ap2_player()]
708    for player in players:
709        player.stream = MagicMock()
710        player.stream.wait_for_connection = AsyncMock()
711        player.stream.wait_audio_present = AsyncMock(return_value=True)
712        player.stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.UNREPORTED, 0))
713        player.stream.start = AsyncMock(side_effect=_ack_commanded_instant)
714        player.config.get_value = MagicMock(return_value=0)
715    session = _make_ptp_session(prov, players)
716    now = 100.0
717
718    async def _resolve_shared_ptp(_ap2_members: int | None = None) -> bool:
719        nonlocal now
720        now = 103.0
721        return True
722
723    with (
724        patch.object(session, "_resolve_shared_ptp", new=_resolve_shared_ptp),
725        patch(
726            "music_assistant.providers.airplay.stream_session.time.time",
727            side_effect=lambda: now,
728        ),
729        patch.object(session, "_start_client", new_callable=AsyncMock),
730        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
731    ):
732        await session.start(MagicMock())
733
734    # anchor = now (103_000 ms) + the COLD group start lead: a cold group
735    # start covers the members' receiver-side clock acquisition, unlike the
736    # short event-confirmed warm leads.
737    expected = 103_000 + AIRPLAY_COLD_GROUP_START_LEAD_MS
738    assert session.start_unix_ms == expected
739    for player in players:
740        player.stream.start.assert_awaited_once()
741        assert player.stream.start.await_args.args[0] == expected
742
743
744# --- Session decision reaches the CLI args (overrides bare readiness) ----------
745
746
747def _stream_player(*, ptp_daemon_ready: bool) -> MagicMock:
748    """Build a minimal AirPlay player mock sufficient for _build_cli_args."""
749    player = MagicMock()
750    player.player_id = "apaabbccddeeff"
751    player.display_name = "Player A"
752    player.address = "192.168.1.50"
753    player.protocol = StreamingProtocol.AIRPLAY2
754    player.protocol_override = None
755    player.volume_level = 40
756    player.device_info.mac_address = "AA:BB:CC:DD:EE:FF"
757    player.device_info.ip_address = "192.168.1.50"
758    player.device_info.manufacturer = "Acme, Inc."
759    player.device_info.model = "Test1,1"
760    player.logger = logging.getLogger("test.airplay.player")
761    player.config.get_value = MagicMock(return_value=None)
762    # Keep the arg build on its shortest path: no discovery records to expand.
763    player.airplay_discovery_info = None
764    player.raop_discovery_info = None
765
766    prov = MagicMock()
767    prov.dacp_id = "ABCDEF0123456789"
768    prov.ptp_daemon_ready = ptp_daemon_ready
769    prov.logger = logging.getLogger("test.airplay.prov")
770    prov.mass.streams.publish_ip = "192.168.1.99"
771    prov.mass.streams.get_source_ip = AsyncMock(return_value="192.168.1.5")
772    prov.mass.streams.get_publish_ip = MagicMock(return_value=None)
773    player.provider = prov
774    return player
775
776
777async def _build_args(player: MagicMock, use_shared_ptp: bool | None) -> list[str]:
778    """Assemble CLI args for the player with the externals patched out."""
779    stream = AirPlayStream(player)
780    with patch(
781        "music_assistant.providers.airplay.stream.get_cli_binary",
782        return_value="/fake/cliairplay",
783    ):
784        return await stream._build_cli_args(use_shared_ptp)
785
786
787async def test_build_cli_args_explicit_shared_ptp_overrides_unready_daemon() -> None:
788    """An explicit True adds --ptp-shared even when the daemon reads as not-ready."""
789    player = _stream_player(ptp_daemon_ready=False)
790
791    args = await _build_args(player, use_shared_ptp=True)
792
793    assert "--ptp-shared" in args
794
795
796async def test_build_cli_args_explicit_no_shared_ptp_overrides_ready_daemon() -> None:
797    """An explicit False omits --ptp-shared even while the daemon is ready."""
798    player = _stream_player(ptp_daemon_ready=True)
799
800    args = await _build_args(player, use_shared_ptp=False)
801
802    assert "--ptp-shared" not in args
803
804
805async def test_build_cli_args_none_falls_back_to_daemon_readiness() -> None:
806    """Callers without a group-wide decision (None) gate --ptp-shared on daemon readiness."""
807    assert "--ptp-shared" in await _build_args(
808        _stream_player(ptp_daemon_ready=True), use_shared_ptp=None
809    )
810    assert "--ptp-shared" not in await _build_args(
811        _stream_player(ptp_daemon_ready=False), use_shared_ptp=None
812    )
813
814
815# --- Own AirPlay Receiver filtering in discovery -------------------------------
816
817RECEIVER_INSTANCE_ID = "airplay_receiver--test1234"
818HOST_IPS = ("192.168.1.10", "172.30.32.1")
819
820
821def _receiver_filter_provider(
822    provider_configs: dict[str, dict[str, object]],
823    receiver_instances: tuple[MagicMock, ...] = (),
824) -> AirPlayProvider:
825    """Build a bare provider wired to raw provider configs and running receiver instances."""
826
827    def get_setup_value(instance_id: str, key: str) -> object | None:
828        setup_data = provider_configs.get(instance_id, {}).get("setup_data")
829        return setup_data.get(key) if isinstance(setup_data, dict) else None
830
831    prov = AirPlayProvider.__new__(AirPlayProvider)
832    prov.mass = MagicMock()
833    prov.logger = logging.getLogger("test.airplay.provider")
834    prov.mass.config.get.return_value = provider_configs
835    prov.mass.config.get_provider_setup_value.side_effect = get_setup_value
836    prov.mass.get_provider_instances.return_value = list(receiver_instances)
837    return prov
838
839
840def _receiver_config(
841    airplay_name: str | None = "Garage [AirPlay]",
842    enabled: bool = True,
843    use_setup_data: bool = False,
844) -> dict[str, dict[str, object]]:
845    """Build the raw provider config store with a single AirPlay Receiver instance."""
846    values: dict[str, object] = (
847        {"airplay_name": airplay_name} if airplay_name and not use_setup_data else {}
848    )
849    setup_data: dict[str, object] = (
850        {"airplay_name": airplay_name} if airplay_name and use_setup_data else {}
851    )
852    return {
853        RECEIVER_INSTANCE_ID: {
854            "domain": "airplay_receiver",
855            "instance_id": RECEIVER_INSTANCE_ID,
856            "enabled": enabled,
857            "values": values,
858            "setup_data": setup_data,
859        },
860        "spotify": {"domain": "spotify", "instance_id": "spotify", "values": {}},
861    }
862
863
864def _discovery_info(
865    addresses: list[str],
866    port: int | None = 7123,
867    properties: dict[str, str] | None = None,
868) -> MagicMock:
869    """Build a minimal mdns service info stand-in for the receiver filter."""
870    info = MagicMock()
871    info.parsed_addresses.return_value = addresses
872    info.port = port
873    info.decoded_properties = properties or {}
874    return info
875
876
877def _patch_host_ips() -> AbstractContextManager[AsyncMock]:
878    """Patch the host IP enumeration used by the receiver filter."""
879    return patch(
880        "music_assistant.providers.airplay.provider.get_ip_addresses",
881        AsyncMock(return_value=HOST_IPS),
882    )
883
884
885async def test_own_receiver_filtered_by_name_without_txt_record() -> None:
886    """A host-local advertisement matching a receiver name is ours, even without TXT data."""
887    prov = _receiver_filter_provider(_receiver_config())
888    # advertised via a secondary host interface (e.g. a docker bridge) and
889    # with an unresolved TXT record - both must not defeat the filter
890    info = _discovery_info(["172.30.32.1"], port=9999)
891
892    with _patch_host_ips():
893        assert await prov._is_own_airplay_receiver("Garage [AirPlay]", info) is True
894
895
896async def test_own_receiver_filtered_with_default_name() -> None:
897    """A receiver instance without an explicit name advertises the default name."""
898    prov = _receiver_filter_provider(_receiver_config(airplay_name=None))
899    info = _discovery_info(["192.168.1.10"])
900
901    with _patch_host_ips():
902        assert await prov._is_own_airplay_receiver("Music Assistant", info) is True
903
904
905async def test_own_receiver_filtered_with_setup_flow_name() -> None:
906    """A receiver name stored by its setup flow is used before the instance loads."""
907    prov = _receiver_filter_provider(_receiver_config(use_setup_data=True))
908    info = _discovery_info(["192.168.1.10"], port=9999)
909
910    with _patch_host_ips():
911        assert await prov._is_own_airplay_receiver("Garage [AirPlay]", info) is True
912
913
914async def test_own_receiver_filtered_on_loopback() -> None:
915    """A loopback advertisement matching a receiver name is filtered without host IP probing."""
916    prov = _receiver_filter_provider(_receiver_config())
917    info = _discovery_info(["127.0.0.1"])
918
919    assert await prov._is_own_airplay_receiver("Garage [AirPlay]", info) is True
920
921
922async def test_own_receiver_filtered_by_port_and_model() -> None:
923    """A host-local ShairportSync advertisement on a receiver port is ours, any name."""
924    prov = _receiver_filter_provider(_receiver_config())
925    info = _discovery_info(
926        ["192.168.1.10"],
927        port=airplay_receiver_port(RECEIVER_INSTANCE_ID),
928        properties={"am": "ShairportSync"},
929    )
930
931    with _patch_host_ips():
932        assert await prov._is_own_airplay_receiver("Renamed Receiver", info) is True
933
934
935async def test_own_receiver_filtered_by_running_instance_port() -> None:
936    """The port of a running receiver instance is honoured next to the derived ports."""
937    receiver = MagicMock()
938    receiver.airplay_port = 7555
939    prov = _receiver_filter_provider(_receiver_config(), receiver_instances=(receiver,))
940    info = _discovery_info(["192.168.1.10"], port=7555, properties={"am": "ShairportSync"})
941
942    with _patch_host_ips():
943        assert await prov._is_own_airplay_receiver("Renamed Receiver", info) is True
944
945
946async def test_same_name_receiver_on_other_host_not_filtered() -> None:
947    """A device with a matching name on another host must not be filtered."""
948    prov = _receiver_filter_provider(_receiver_config())
949    info = _discovery_info(["192.168.1.55"], properties={"am": "ShairportSync"})
950
951    with _patch_host_ips():
952        assert await prov._is_own_airplay_receiver("Garage [AirPlay]", info) is False
953
954
955async def test_user_shairport_on_same_host_not_filtered() -> None:
956    """A user-run shairport-sync on this host with its own name and port stays usable."""
957    prov = _receiver_filter_provider(_receiver_config())
958    info = _discovery_info(["192.168.1.10"], port=5000, properties={"am": "ShairportSync"})
959
960    with _patch_host_ips():
961        assert await prov._is_own_airplay_receiver("My Shairport", info) is False
962
963
964async def test_local_receiver_with_other_model_not_filtered_on_port_collision() -> None:
965    """A non-shairport local receiver is kept even when its port collides (e.g. macOS on 7000)."""
966    receiver = MagicMock()
967    receiver.airplay_port = 7000
968    prov = _receiver_filter_provider(_receiver_config(), receiver_instances=(receiver,))
969    info = _discovery_info(["192.168.1.10"], port=7000, properties={"am": "MacBookAir10,1"})
970
971    with _patch_host_ips():
972        assert await prov._is_own_airplay_receiver("Marcel's MacBook Air", info) is False
973
974
975async def test_disabled_receiver_config_not_filtered() -> None:
976    """A disabled receiver instance cannot be advertising, so its identity is ignored."""
977    prov = _receiver_filter_provider(_receiver_config(enabled=False))
978    info = _discovery_info(["192.168.1.10"])
979
980    with _patch_host_ips():
981        assert await prov._is_own_airplay_receiver("Garage [AirPlay]", info) is False
982
983
984async def test_no_receiver_configs_never_filters() -> None:
985    """Without any configured receiver instance the filter is a no-op."""
986    prov = _receiver_filter_provider(
987        {"spotify": {"domain": "spotify", "instance_id": "spotify", "values": {}}}
988    )
989    info = _discovery_info(["192.168.1.10"], properties={"am": "ShairportSync"})
990
991    assert await prov._is_own_airplay_receiver("Garage [AirPlay]", info) is False
992
993
994async def test_setup_player_skips_own_receiver_before_service_lookup() -> None:
995    """Discovery setup bails out on own receivers before any counterpart service lookup."""
996    prov = AirPlayProvider.__new__(AirPlayProvider)
997    prov.mass = MagicMock()
998    prov.logger = logging.getLogger("test.airplay.provider")
999    prov.mass.config.get_raw_player_config_value.return_value = True
1000    prov._is_own_airplay_receiver = AsyncMock(return_value=True)  # type: ignore[method-assign]
1001    info = _discovery_info(["192.168.1.10"])
1002    info.type = "_raop._tcp.local."
1003
1004    await prov._setup_player("apdb1ff0aae80e", "Garage [AirPlay]", info)
1005
1006    prov.mass.discovery.async_find_mdns_service.assert_not_called()
1007    prov.mass.players.register.assert_not_called()
1008