music-assistant-server

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