music-assistant-server

1"""Unit tests for the AirPlay stream CLI argument assembly."""
2
3import asyncio
4import errno
5import logging
6import os
7import select
8import threading
9from collections.abc import AsyncGenerator, AsyncIterator, Callable, Coroutine
10from contextlib import asynccontextmanager, suppress
11from pathlib import Path
12from typing import Any
13from unittest.mock import AsyncMock, MagicMock, call, patch
14
15import pytest
16from music_assistant_models.enums import ContentType, PlaybackState
17from music_assistant_models.errors import PlayerCommandFailed
18from music_assistant_models.media_items import AudioFormat
19
20from music_assistant.helpers.named_pipe import WRITE_POLL_INTERVAL_MS, AsyncNamedPipeWriter
21from music_assistant.providers.airplay.constants import (
22    AIRPLAY_ARTWORK_SIZE,
23    AIRPLAY_JOIN_START_ACK_TIMEOUT_MS,
24    AIRPLAY_START_ACK_TIMEOUT_MS,
25    CONF_AIRPLAY_CREDENTIALS,
26    CONF_BUFFER_DEPTH,
27    CONF_ENCRYPTION,
28    CONF_PASSWORD,
29    CONF_STREAMING_MODE,
30    STREAMING_MODE_AP2_COMPAT,
31    STREAMING_MODE_AP2_NTP,
32    STREAMING_MODE_AP2_PTP,
33    STREAMING_MODE_AUTO,
34    STREAMING_MODE_RAOP,
35    AirPlayRemoteCommand,
36    ClockReadiness,
37    StreamingProtocol,
38)
39from music_assistant.providers.airplay.stream import AirPlayStream, CliError
40
41START_UNIX_MS = 1_750_000_000_000
42AP2_FEATURES = "0x4A7FDFD5,0x3C177FDE"
43
44
45def _make_cli_proc(*, quiesced: bool = True, calls: list[str] | None = None) -> MagicMock:
46    """
47    Build a mock cliairplay process that answers its awaited calls.
48
49    :param quiesced: What holding stdin quiet reports about emptying the buffer.
50    :param calls: Records "quiesce" as stdin is held, for ordering assertions.
51    """
52
53    @asynccontextmanager
54    async def _stdin_quiesced(*_args: object) -> AsyncIterator[bool]:
55        if calls is not None:
56            calls.append("quiesce")
57        yield quiesced
58
59    return MagicMock(closed=False, stdin_quiesced=_stdin_quiesced)
60
61
62def _make_player() -> MagicMock:
63    """Build a mock AirPlay player with both discovery records present."""
64    player = MagicMock()
65    player.player_id = "apaabbccddeeff"
66    player.display_name = "Player A"
67    player.address = "192.168.1.50"
68    player.protocol = StreamingProtocol.AIRPLAY2
69    player.protocol_override = None
70    player.volume_level = 40
71    player.device_info.mac_address = "AA:BB:CC:DD:EE:FF"
72    player.device_info.ip_address = "192.168.1.50"
73    player.device_info.manufacturer = "Acme, Inc."
74    player.device_info.model = "Test1,1"
75    player.logger = logging.getLogger("test.airplay.player")
76    player.config.get_value = MagicMock(side_effect=lambda _key, default=None: default)
77    player.state.active_group = None
78    player.streaming_mode = STREAMING_MODE_AUTO
79    player.streaming_mode_options = [
80        MagicMock(value=STREAMING_MODE_AUTO),
81        MagicMock(value=STREAMING_MODE_AP2_NTP),
82    ]
83    player.synced_to = None
84    player.group_members = []
85
86    airplay_info = MagicMock()
87    airplay_info.port = 7000
88    airplay_info.server = "playera.local."
89    airplay_info.decoded_properties = {
90        "features": "0x5A7FFFF7,0x1E",
91        "flags": "0x4",
92        "model": "Test1,1",
93        "manufacturer": "Acme, Inc.",  # contains a space: must be skipped in --txt
94    }
95    player.airplay_discovery_info = airplay_info
96
97    raop_info = MagicMock()
98    raop_info.port = 5000
99    raop_info.name = "AABBCCDDEEFF@Player A._raop._tcp.local."
100    raop_info.decoded_properties = {"et": "0,4", "md": "0,1,2", "cn": "0,1"}
101    player.raop_discovery_info = raop_info
102
103    prov = MagicMock()
104    prov.dacp_id = "ABCDEF0123456789"
105    prov.ptp_daemon_ready = True
106    prov.logger = logging.getLogger("test.airplay.prov")
107    # auto-detected publish ip: reachable address of this host, but not the interface
108    # the stream to this device leaves from - so it is never handed to the binary
109    prov.mass.streams.publish_ip = "192.168.1.99"
110    prov.mass.streams.get_source_ip = AsyncMock(return_value="192.168.1.5")
111    prov.mass.streams.get_publish_ip = MagicMock(return_value=None)
112    player.provider = prov
113    return player
114
115
116async def _build_args(player: MagicMock) -> list[str]:
117    """Build the CLI args for the given player with the externals patched out."""
118    stream = AirPlayStream(player)
119    with patch(
120        "music_assistant.providers.airplay.stream.get_cli_binary",
121        return_value="/fake/cliairplay",
122    ):
123        return await stream._build_cli_args()
124
125
126def _arg_value(args: list[str], flag: str) -> Any:
127    """Return the value following the given flag in the argument list."""
128    return args[args.index(flag) + 1]
129
130
131def _acking_write_cli_command(
132    stream: AirPlayStream, at_unix_ms: int | None = None
133) -> Callable[[str], Coroutine[Any, Any, bool]]:
134    """
135    Build a ``_write_cli_command`` replacement that acks a START like the binary.
136
137    :param stream: The stream whose START commands are answered.
138    :param at_unix_ms: Scheduled audible instant to report back; defaults to the
139        commanded instant, i.e. the binary found it feasible.
140    """
141
142    async def write_command(command: str) -> bool:
143        # A START nothing acks otherwise pays the real ack timeout in wall clock;
144        # feeding the ack here leaves start()'s wait already released.
145        if "ACTION=START" in command:
146            requested = int(command.split("START_UNIX_MS=")[1].split("\n", 1)[0])
147            stream._handle_status_line(
148                f"[STATUS] started requested_unix_ms={requested} "
149                f"at_unix_ms={requested if at_unix_ms is None else at_unix_ms}"
150            )
151        return True
152
153    return write_command
154
155
156@pytest.mark.asyncio
157async def test_cli_args_default_auto() -> None:
158    """Default (no protocol override) passes --protocol auto with the full mDNS TXT."""
159    player = _make_player()
160    args = await _build_args(player)
161
162    assert _arg_value(args, "--protocol") == "auto"
163    assert "--start-unix-ms" not in args
164    # legacy timing args are gone
165    assert "--ntpstart" not in args
166    assert "--wait" not in args
167    # AirPlay 2 service is the connection target when it may be used
168    assert _arg_value(args, "--port") == "7000"
169    assert _arg_value(args, "--name") == "Player A"
170    assert _arg_value(args, "--hostname") == "playera.local."
171    # RAOP mDNS props still passed for the RAOP-based flows
172    assert _arg_value(args, "--udn") == "AABBCCDDEEFF@Player A._raop._tcp.local."
173    assert _arg_value(args, "--et") == "0,4"
174    assert _arg_value(args, "--cn") == "0,1"
175    # full TXT for route selection; pairs containing whitespace are skipped
176    txt = _arg_value(args, "--txt")
177    assert "features=0x5A7FFFF7,0x1E" in txt
178    assert "flags=0x4" in txt
179    assert "manufacturer" not in txt
180    # default format
181    assert _arg_value(args, "--samplerate") == "44100"
182    assert _arg_value(args, "--bitdepth") == "16"
183    # no explicit latency override configured
184    assert "--latency" not in args
185    # PTP daemon is running: stream attaches to the shared clock
186    assert "--ptp-shared" in args
187    # networking: the interface the timing packets leave from is pinned
188    assert _arg_value(args, "--if") == "192.168.1.5"
189    assert "--publish-ip" not in args
190    # the target is the only positional argument; PREPARE selects stdin
191    assert args[-1] == "192.168.1.50"
192    assert "-" not in args
193    assert "--cmdpipe" in args
194
195
196@pytest.mark.asyncio
197async def test_cli_args_auto_publish_ip_is_never_advertised() -> None:
198    """
199    An auto-detected publish IP must not reach --publish-ip.
200
201    The binary treats it as authoritative for the PTP timing-peer list, while the
202    timing packets leave from the resolved --if interface. A peer list naming any
203    other address makes the receiver discard our clock and play silence.
204    """
205    player = _make_player()
206    player.provider.mass.streams.publish_ip = "10.45.0.20"
207
208    args = await _build_args(player)
209
210    assert _arg_value(args, "--if") == "192.168.1.5"
211    assert "--publish-ip" not in args
212    assert "10.45.0.20" not in args
213
214
215@pytest.mark.asyncio
216async def test_cli_args_configured_publish_ip_is_advertised() -> None:
217    """An explicitly configured publish IP is a reachability statement and is passed on."""
218    player = _make_player()
219    player.provider.mass.streams.get_publish_ip = MagicMock(return_value="10.45.0.20")
220
221    args = await _build_args(player)
222
223    assert _arg_value(args, "--publish-ip") == "10.45.0.20"
224    player.provider.mass.streams.get_publish_ip.assert_called_once_with("192.168.1.50")
225
226
227@pytest.mark.asyncio
228async def test_cli_args_publish_ip_omitted_when_it_matches_the_interface() -> None:
229    """A publish IP identical to the bound interface adds nothing to the peer list."""
230    player = _make_player()
231    player.provider.mass.streams.get_publish_ip = MagicMock(return_value="192.168.1.5")
232
233    args = await _build_args(player)
234
235    assert _arg_value(args, "--if") == "192.168.1.5"
236    assert "--publish-ip" not in args
237
238
239@pytest.mark.asyncio
240async def test_cli_args_no_interface_pin_leaves_routing_to_the_binary() -> None:
241    """With no interface to pin, --if is dropped so the routing table decides."""
242    player = _make_player()
243    player.provider.mass.streams.get_source_ip = AsyncMock(return_value=None)
244
245    args = await _build_args(player)
246
247    assert "--if" not in args
248
249
250@pytest.mark.asyncio
251async def test_cli_args_log_the_pinned_interface(caplog: pytest.LogCaptureFixture) -> None:
252    """The resolved interface is stated outright, so a user's log shows what it bound to."""
253    player = _make_player()
254
255    with caplog.at_level(logging.DEBUG):
256        await _build_args(player)
257
258    assert (
259        "cliairplay network binding for player apaabbccddeeff: "
260        "if=192.168.1.5 publish_ip=<not configured>" in caplog.text
261    )
262
263
264@pytest.mark.asyncio
265async def test_cli_args_log_the_unpinned_interface_and_publish_ip(
266    caplog: pytest.LogCaptureFixture,
267) -> None:
268    """An unpinned interface and a configured publish IP are named for what they are."""
269    player = _make_player()
270    player.provider.mass.streams.get_source_ip = AsyncMock(return_value=None)
271    player.provider.mass.streams.get_publish_ip = MagicMock(return_value="10.45.0.20")
272
273    with caplog.at_level(logging.DEBUG):
274        await _build_args(player)
275
276    assert (
277        "cliairplay network binding for player apaabbccddeeff: "
278        "if=<all interfaces> publish_ip=10.45.0.20" in caplog.text
279    )
280
281
282@pytest.mark.asyncio
283async def test_cli_args_raop_override() -> None:
284    """A forced RAOP protocol targets the RAOP service and skips AP2-only args."""
285    player = _make_player()
286    player.protocol_override = StreamingProtocol.RAOP
287    args = await _build_args(player)
288
289    assert _arg_value(args, "--protocol") == "raop"
290    assert _arg_value(args, "--port") == "5000"
291    assert "--name" not in args
292    assert "--hostname" not in args
293    assert "--ptp-shared" not in args
294    assert "--encrypt" in args
295
296
297@pytest.mark.asyncio
298async def test_cli_args_raop_encryption_can_be_disabled() -> None:
299    """The legacy encryption preference remains available for incompatible receivers."""
300    player = _make_player()
301    player.protocol_override = StreamingProtocol.RAOP
302    player.config.get_value = MagicMock(
303        side_effect=lambda key, default=None: False if key == CONF_ENCRYPTION else default
304    )
305
306    args = await _build_args(player)
307
308    assert "--encrypt" not in args
309
310
311@pytest.mark.asyncio
312async def test_cli_args_no_ptp_shared_when_daemon_alive_but_not_ready() -> None:
313    """
314    A daemon that is merely alive must not get --ptp-shared.
315
316    Liveness does not mean the daemon is serving: until it publishes its clock
317    there is nothing to attach to, and a stream that asks anyway silently takes
318    its own timing instead - drifting away from the members that got the clock.
319    """
320    player = _make_player()
321    player.provider.ptp_daemon_running = True
322    player.provider.ptp_daemon_ready = False
323    args = await _build_args(player)
324    assert "--ptp-shared" not in args
325
326
327@pytest.mark.asyncio
328async def test_cli_args_no_family_buffer_defaults() -> None:
329    """
330    Every device family stays on Automatic depth: no --latency by default.
331
332    The LinkPlay generations that used to get a deepened realtime queue from
333    the family table manage their own buffer on the buffered stream, so the
334    table ships empty and only the per-player setting adds the argument.
335    """
336    player = _make_player()
337    player.device_info.manufacturer = "Linkplay Technology Inc."
338    args = await _build_args(player)
339    assert "--latency" not in args
340    assert "--ptp-shared" in args
341
342    player = _make_player()
343    player.device_info.manufacturer = "Edifier Inc"
344    player.airplay_discovery_info.decoded_properties["fv"] = "p20.Linkplay.4.6.430230"
345    args = await _build_args(player)
346    assert "--latency" not in args
347
348    player = _make_player()
349    args = await _build_args(player)
350    assert "--latency" not in args
351
352
353@pytest.mark.asyncio
354async def test_cli_args_buffer_depth_config_overrides_auto() -> None:
355    """A configured buffer depth wins over the device-family default."""
356    player = _make_player()
357    player.device_info.manufacturer = "Linkplay Technology Inc."
358    player.config.get_value = MagicMock(
359        side_effect=lambda key, default=None: 1500 if key == CONF_BUFFER_DEPTH else default
360    )
361    args = await _build_args(player)
362    assert _arg_value(args, "--latency") == "1500"
363
364
365@pytest.mark.asyncio
366async def test_cli_args_no_latency_override() -> None:
367    """The playback lead/buffer is binary-managed; MA never passes --latency."""
368    player = _make_player()
369    args = await _build_args(player)
370    assert "--latency" not in args
371
372
373@pytest.mark.asyncio
374async def test_cli_args_hires_pcm_format() -> None:
375    """A 24-bit stream passes --bitdepth 24 while the pipe carries s32le samples."""
376    player = _make_player()
377    hires_format = AudioFormat(content_type=ContentType.PCM_S32LE, sample_rate=48000, bit_depth=24)
378    stream = AirPlayStream(player, pcm_format=hires_format)
379    with patch(
380        "music_assistant.providers.airplay.stream.get_cli_binary",
381        return_value="/fake/cliairplay",
382    ):
383        args = await stream._build_cli_args()
384
385    assert _arg_value(args, "--samplerate") == "48000"
386    assert _arg_value(args, "--bitdepth") == "24"
387    # the ffmpeg pipe format must be the 32-bit container (binary truncates to 24)
388    assert stream.pcm_format.content_type == ContentType.PCM_S32LE
389
390
391@pytest.mark.asyncio
392async def test_cli_args_raop_only_device() -> None:
393    """True legacy RAOP uses the command pipe and has no positional audio source."""
394    player = _make_player()
395    player.airplay_discovery_info = None
396    player.protocol = StreamingProtocol.RAOP
397    args = await _build_args(player)
398
399    assert _arg_value(args, "--protocol") == "auto"
400    assert _arg_value(args, "--port") == "5000"
401    assert "--txt" not in args
402    assert "--name" not in args
403    assert "--encrypt" in args
404    assert "--cmdpipe" in args
405    assert args[-1] == "192.168.1.50"
406    assert "-" not in args
407
408
409@pytest.mark.asyncio
410async def test_cli_args_auto_raop_uses_raop_service_port() -> None:
411    """Auto-selected legacy RAOP targets the RAOP service rather than the AP2 service."""
412    player = _make_player()
413    player.protocol = StreamingProtocol.RAOP
414    player.airplay_discovery_info.decoded_properties["features"] = "0x0"
415
416    args = await _build_args(player)
417
418    assert _arg_value(args, "--protocol") == "auto"
419    assert _arg_value(args, "--port") == "5000"
420    assert "--name" not in args
421
422
423@pytest.mark.asyncio
424async def test_cli_args_pass_raop_feature_fallback_to_auto_router() -> None:
425    """AP2 bits advertised only on _raop.ft still reach the binary route resolver."""
426    player = _make_player()
427    player.airplay_discovery_info.decoded_properties.pop("features")
428    player.raop_discovery_info.decoded_properties["ft"] = AP2_FEATURES
429
430    args = await _build_args(player)
431
432    assert _arg_value(args, "--protocol") == "auto"
433    assert _arg_value(args, "--port") == "7000"
434    assert f"ft={AP2_FEATURES}" in _arg_value(args, "--txt")
435
436
437@pytest.mark.asyncio
438async def test_cli_args_featureless_ap2_only_device_forces_airplay2() -> None:
439    """An AP2-only receiver without feature bits cannot fall through to legacy RAOP."""
440    player = _make_player()
441    player.raop_discovery_info = None
442    player.airplay_discovery_info.decoded_properties = {}
443
444    args = await _build_args(player)
445
446    assert _arg_value(args, "--protocol") == "airplay2"
447    assert _arg_value(args, "--port") == "7000"
448
449
450@pytest.mark.parametrize(
451    ("line", "expected_route"),
452    [
453        (
454            "[STATUS] route protocol=airplay2 flow=realtime timing=ptp buffered=0",
455            "AirPlay 2 (realtime, PTP)",
456        ),
457        # a buffered stream is named by its delivery, whatever flow it was requested as
458        (
459            "[STATUS] route protocol=airplay2 flow=realtime timing=ntp buffered=1",
460            "AirPlay 2 (buffered, NTP)",
461        ),
462        (
463            "[STATUS] route protocol=raop flow=realtime timing=ntp buffered=0",
464            "RAOP",
465        ),
466    ],
467    ids=["airplay2-realtime", "airplay2-buffered", "raop"],
468)
469def test_parse_route_status(
470    line: str, expected_route: str, caplog: pytest.LogCaptureFixture
471) -> None:
472    """The [STATUS] route line resolves the route this stream took and reports it."""
473    player = _make_player()
474    stream = AirPlayStream(player)
475
476    with caplog.at_level(logging.INFO):
477        stream._parse_route_status(line)
478
479    assert stream.active_route == expected_route
480    assert f"Streaming to Player A via {expected_route}" in caplog.text
481
482
483@pytest.mark.asyncio
484async def test_stdout_reader_dispatches_the_route_line() -> None:
485    """The CLI stdout reader hands the [STATUS] route line to the route parser."""
486    player = _make_player()
487    stream = AirPlayStream(player)
488    process = MagicMock()
489    process.read = AsyncMock(
490        side_effect=[
491            b"[STATUS] route protocol=airplay2 flow=realtime timing=ptp buffered=0\n",
492            b"",
493        ]
494    )
495    stream._cli_proc = process
496
497    await stream._stdout_reader()
498
499    assert stream.active_route == "AirPlay 2 (realtime, PTP)"
500
501
502def test_parse_latency_status() -> None:
503    """The [STATUS] latency line is parsed into the stream's latency attributes."""
504    player = _make_player()
505    stream = AirPlayStream(player)
506    stream._parse_latency_status(
507        "[STATUS] latency lead_ms=1750 device_min_frames=11025 device_max_frames=88200 "
508        "warm_lead_ms=1200"
509    )
510    assert stream.latency_lead_ms == 1750
511    assert stream.device_min_frames == 11025
512    assert stream.device_max_frames == 88200
513    assert stream.warm_lead_ms == 1200
514
515
516def test_parse_latency_status_reads_every_field_on_its_own() -> None:
517    """One unusable value must not leave the fields after it stale from the last report."""
518    player = _make_player()
519    stream = AirPlayStream(player)
520    stream._parse_latency_status(
521        "[STATUS] latency lead_ms=1750 device_min_frames=11025 device_max_frames=88200 "
522        "warm_lead_ms=1200"
523    )
524
525    stream._parse_latency_status(
526        "[STATUS] latency lead_ms=900 device_min_frames=garbage device_max_frames=44100 "
527        "warm_lead_ms=300"
528    )
529
530    assert stream.latency_lead_ms == 900
531    assert stream.device_min_frames == 0  # unusable, so unreported
532    assert stream.device_max_frames == 44100
533    assert stream.warm_lead_ms == 300
534
535
536def test_parse_latency_status_logs_the_warm_lead(caplog: pytest.LogCaptureFixture) -> None:
537    """The warm lead drives every warm group anchor, so it belongs in the line."""
538    stream = AirPlayStream(_make_player())
539
540    with caplog.at_level(logging.DEBUG):
541        stream._parse_latency_status(
542            "[STATUS] latency lead_ms=1750 device_min_frames=11025 device_max_frames=88200 "
543            "warm_lead_ms=1200"
544        )
545
546    assert "warm lead=1200ms" in caplog.text
547
548
549def test_mrp_push_accepted_is_not_logged_at_info(caplog: pytest.LogCaptureFixture) -> None:
550    """A push the device accepted is bookkeeping, not something to act on."""
551    stream = AirPlayStream(_make_player())
552
553    with caplog.at_level(logging.INFO):
554        stream._parse_mrp_status("[STATUS] mrp path=command status=200")
555
556    assert caplog.text == ""
557
558
559@pytest.mark.parametrize("status", [302, 403, 500], ids=["redirect", "forbidden", "server-error"])
560def test_mrp_push_rejection_is_reported(status: int, caplog: pytest.LogCaptureFixture) -> None:
561    """Anything but a 2xx is the device not taking the push, so it is reported."""
562    stream = AirPlayStream(_make_player())
563
564    with caplog.at_level(logging.WARNING):
565        stream._parse_mrp_status(f"[STATUS] mrp path=command status={status}")
566
567    assert "Player A" in caplog.text
568    assert str(status) in caplog.text
569
570
571def test_mrp_artwork_rejection_is_reported(caplog: pytest.LogCaptureFixture) -> None:
572    """An artwork rejection carries no path= or status=, and must not read as a plain push."""
573    stream = AirPlayStream(_make_player())
574
575    with caplog.at_level(logging.WARNING):
576        stream._parse_mrp_status(
577            "[STATUS] mrp artwork=rejected reason=progressive_jpeg bytes=48123 "
578            "width=512 height=512 precision=8 sof=0xc2 components=3 progressive=1 "
579            "clear_status=200 staging_max_bytes=131072"
580        )
581
582    assert "rejected the now-playing artwork" in caplog.text
583    assert "progressive_jpeg" in caplog.text
584    assert "HTTP ?" not in caplog.text
585
586
587def test_mrp_artwork_rejection_reaches_the_parser_from_the_stderr_reader(
588    caplog: pytest.LogCaptureFixture,
589) -> None:
590    """The binary reports artwork on stderr, so the status dispatcher must route it."""
591    stream = AirPlayStream(_make_player())
592
593    with caplog.at_level(logging.WARNING):
594        ends_the_loop = stream._handle_status_line(
595            "[STATUS] mrp artwork=rejected reason=progressive_jpeg bytes=48123 "
596            "width=512 height=512 precision=8 sof=0xc2 components=3 progressive=1 "
597            "clear_status=200 staging_max_bytes=131072"
598        )
599
600    assert ends_the_loop is False
601    assert "rejected the now-playing artwork" in caplog.text
602
603
604def test_mrp_channel_status_is_not_read_as_an_http_status(
605    caplog: pytest.LogCaptureFixture,
606) -> None:
607    """The data-channel line reports 0/1, which must not be warned about as a failed push."""
608    stream = AirPlayStream(_make_player())
609
610    with caplog.at_level(logging.WARNING):
611        stream._parse_mrp_status("[STATUS] mrp path=channel status=0")
612
613    assert caplog.text == ""
614
615
616@pytest.mark.parametrize(
617    ("line", "expected"),
618    [
619        # both tables published: the union is kept
620        (
621            "[STATUS] capabilities requested=0x80000 realtime_formats=0x40000 "
622            "realtime_known=1 buffered_formats=0x80000 buffered_known=1",
623            0xC0000,
624        ),
625        # the Apple TV publishes 24-bit for its buffered stream only
626        (
627            "[STATUS] capabilities requested=0x80000 realtime_formats=0x1440800 "
628            "realtime_known=1 buffered_formats=0xe80000 buffered_known=1",
629            0x1EC0800,
630        ),
631        # a table the device did not publish is ignored, even when non-zero
632        (
633            "[STATUS] capabilities requested=0x40000 realtime_formats=0x40000 "
634            "realtime_known=1 buffered_formats=0x80000 buffered_known=0",
635            0x40000,
636        ),
637        # RAOP-compat routes report nothing: the probed value survives
638        (
639            "[STATUS] capabilities requested=0x40000 realtime_formats=0x0 "
640            "realtime_known=0 buffered_formats=0x0 buffered_known=0",
641            0x1234,
642        ),
643        # a malformed mask is ignored rather than raising
644        (
645            "[STATUS] capabilities realtime_formats=nonsense realtime_known=1",
646            0x1234,
647        ),
648    ],
649)
650def test_parse_capabilities_status(line: str, expected: int) -> None:
651    """The [STATUS] capabilities line refreshes the formats the player advertises."""
652    player = _make_player()
653    player.advertised_audio_formats = 0x1234
654    stream = AirPlayStream(player)
655
656    stream._parse_capabilities_status(line)
657
658    assert player.advertised_audio_formats == expected
659
660
661@pytest.mark.parametrize(
662    ("value", "command"),
663    [
664        ("play", AirPlayRemoteCommand.PLAY),
665        ("pause", AirPlayRemoteCommand.PAUSE),
666        ("play_pause", AirPlayRemoteCommand.PLAY_PAUSE),
667        ("next", AirPlayRemoteCommand.NEXT),
668        ("previous", AirPlayRemoteCommand.PREVIOUS),
669    ],
670)
671def test_parse_remote_event(value: str, command: AirPlayRemoteCommand) -> None:
672    """A normalized CLI remote event is dispatched against its own player."""
673    player = _make_player()
674    stream = AirPlayStream(player)
675
676    stream._parse_remote_event(f"[EVENT] remote command={value}")
677
678    player.provider.handle_remote_command.assert_called_once_with(player, command)
679
680
681def test_parse_remote_event_rejects_unknown_command(caplog: pytest.LogCaptureFixture) -> None:
682    """An unknown CLI remote event is reported and ignored."""
683    player = _make_player()
684    stream = AirPlayStream(player)
685
686    with caplog.at_level(logging.WARNING):
687        stream._parse_remote_event("[EVENT] remote command=unsupported")
688
689    player.provider.handle_remote_command.assert_not_called()
690    assert "Ignoring unknown cliairplay remote command: unsupported" in caplog.text
691
692
693@pytest.mark.asyncio
694async def test_stdout_reader_dispatches_remote_events_once() -> None:
695    """The CLI stdout reader dispatches each normalized remote event exactly once."""
696    player = _make_player()
697    stream = AirPlayStream(player)
698    process = MagicMock()
699    output = "".join(f"[EVENT] remote command={command}\n" for command in AirPlayRemoteCommand)
700    process.read = AsyncMock(side_effect=[output.encode(), b""])
701    stream._cli_proc = process
702
703    await stream._stdout_reader()
704
705    assert player.provider.handle_remote_command.call_args_list == [
706        call(player, command) for command in AirPlayRemoteCommand
707    ]
708
709
710def test_command_pipe_paths_are_unique_per_stream() -> None:
711    """A stopped stream cannot remove a replacement stream's command pipe."""
712    player = _make_player()
713    first_stream = AirPlayStream(player)
714    second_stream = AirPlayStream(player)
715    assert first_stream.commands_pipe.path != second_stream.commands_pipe.path
716
717
718@pytest.mark.asyncio
719async def test_command_pipe_write_returns_false_without_reader(
720    tmp_path: Path, caplog: pytest.LogCaptureFixture
721) -> None:
722    """A command pipe write reports a missing reader without retrying or warning."""
723    pipe_path = tmp_path / "commands"
724    os.mkfifo(pipe_path)
725    writer = AsyncNamedPipeWriter(str(pipe_path))
726
727    with (
728        caplog.at_level(logging.WARNING),
729        patch("music_assistant.helpers.named_pipe.os.open", wraps=os.open) as open_pipe,
730    ):
731        assert await writer.write(b"ACTION=STANDBY\n") is False
732
733    # the missing reader is reported once, without retrying or crying wolf
734    open_pipe.assert_called_once()
735    assert not [record for record in caplog.records if record.levelno >= logging.WARNING]
736
737
738@pytest.mark.asyncio
739async def test_command_pipe_write_returns_true_for_complete_write() -> None:
740    """A complete command pipe write reports successful delivery."""
741    data = b"ACTION=STANDBY\n"
742    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
743    writer._write_fd = 42
744
745    with patch("music_assistant.helpers.named_pipe.os.write", return_value=len(data)) as write:
746        assert await writer.write(data) is True
747
748    write.assert_called_once_with(42, data)
749
750
751@pytest.mark.asyncio
752async def test_command_pipe_write_completes_short_write() -> None:
753    """A short command pipe write continues with the remaining data."""
754    data = b"ACTION=STANDBY\n"
755    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
756    writer._write_fd = 42
757
758    with patch(
759        "music_assistant.helpers.named_pipe.os.write",
760        side_effect=[5, len(data) - 5],
761    ) as write:
762        assert await writer.write(data) is True
763
764    assert write.call_args_list == [call(42, memoryview(data)), call(42, memoryview(data)[5:])]
765
766
767@pytest.mark.asyncio
768async def test_command_pipe_write_returns_false_and_resets_fd_on_epipe() -> None:
769    """A closed command pipe reader resets the writer for a later retry."""
770    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
771    writer._write_fd = 42
772
773    with (
774        patch(
775            "music_assistant.helpers.named_pipe.os.write",
776            side_effect=OSError(errno.EPIPE, "reader closed"),
777        ),
778        patch("music_assistant.helpers.named_pipe.os.close") as close_fd,
779    ):
780        assert await writer.write(b"ACTION=STANDBY\n") is False
781
782    close_fd.assert_called_once_with(42)
783    assert writer._write_fd is None
784
785
786@pytest.mark.asyncio
787async def test_command_pipe_write_resets_fd_when_epipe_follows_partial_write() -> None:
788    """An EPIPE after a short write reports failure and resets the writer."""
789    data = b"ACTION=STANDBY\n"
790    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
791    writer._write_fd = 42
792
793    with (
794        patch(
795            "music_assistant.helpers.named_pipe.os.write",
796            side_effect=[5, OSError(errno.EPIPE, "reader closed")],
797        ) as write,
798        patch("music_assistant.helpers.named_pipe.os.close") as close_fd,
799    ):
800        assert await writer.write(data) is False
801
802    assert write.call_args_list == [call(42, memoryview(data)), call(42, memoryview(data)[5:])]
803    close_fd.assert_called_once_with(42)
804    assert writer._write_fd is None
805
806
807@pytest.mark.asyncio
808async def test_command_pipe_writes_are_serialized() -> None:
809    """Concurrent command writes cannot interleave after a short write."""
810    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
811    writer._write_fd = 42
812    written_chunks: list[bytes] = []
813    first_chunk_written = threading.Event()
814    release_first_write = threading.Event()
815
816    def write_chunk(_fd: int, data: memoryview) -> int:
817        chunk = bytes(data)
818        if not written_chunks:
819            written_chunks.append(chunk[:2])
820            first_chunk_written.set()
821            assert release_first_write.wait(timeout=1)
822            return 2
823        written_chunks.append(chunk)
824        return len(chunk)
825
826    with patch("music_assistant.helpers.named_pipe.os.write", side_effect=write_chunk):
827        first_write = asyncio.create_task(writer.write(b"FIRST\n"))
828        assert await asyncio.to_thread(first_chunk_written.wait, 1)
829        second_write = asyncio.create_task(writer.write(b"SECOND\n"))
830        await asyncio.sleep(0)
831        release_first_write.set()
832        assert all(await asyncio.gather(first_write, second_write))
833
834    assert b"".join(written_chunks) == b"FIRST\nSECOND\n"
835
836
837@pytest.mark.asyncio
838async def test_command_pipe_write_propagates_non_epipe_errors() -> None:
839    """An unexpected command pipe write error remains visible to the caller."""
840    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
841    writer._write_fd = 42
842
843    with (
844        patch(
845            "music_assistant.helpers.named_pipe.os.write",
846            side_effect=OSError(errno.EBADF, "bad file descriptor"),
847        ),
848        pytest.raises(OSError, match="bad file descriptor"),
849    ):
850        await writer.write(b"ACTION=STANDBY\n")
851
852
853@pytest.mark.asyncio
854async def test_pipe_write_reports_a_stalled_reader_without_raising(
855    tmp_path: Path, caplog: pytest.LogCaptureFixture
856) -> None:
857    """A write bigger than the pipe buffer reports failure instead of raising."""
858    pipe_path = tmp_path / "audio"
859    os.mkfifo(pipe_path)
860    writer = AsyncNamedPipeWriter(str(pipe_path))
861    # a reader that never reads, so the pipe buffer fills up and stays full
862    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
863
864    try:
865        with (
866            caplog.at_level(logging.WARNING),
867            patch("music_assistant.helpers.named_pipe.WRITE_STALL_TIMEOUT", 0.05),
868        ):
869            assert await writer.write(b"\x00" * 176400) is False
870
871        # the reader is still attached, so the descriptor stays usable
872        assert writer._write_fd is not None
873        assert not [record for record in caplog.records if record.levelno >= logging.WARNING]
874    finally:
875        os.close(read_fd)
876        await writer.remove()
877
878
879@pytest.mark.asyncio
880async def test_pipe_write_completes_a_large_write_while_the_reader_drains(
881    tmp_path: Path,
882) -> None:
883    """A write bigger than the pipe buffer completes against a reader that keeps up."""
884    pipe_path = tmp_path / "audio"
885    os.mkfifo(pipe_path)
886    writer = AsyncNamedPipeWriter(str(pipe_path))
887    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
888    payload = b"\x00" * 176400
889
890    async def drain() -> bytes:
891        received = bytearray()
892        while len(received) < len(payload):
893            try:
894                chunk = os.read(read_fd, 65536)
895            except BlockingIOError:
896                chunk = b""
897            if not chunk:
898                # no writer attached yet or nothing buffered: yield instead of spinning
899                await asyncio.sleep(0.001)
900                continue
901            received += chunk
902        return bytes(received)
903
904    reader = asyncio.create_task(drain())
905    try:
906        async with asyncio.timeout(10):
907            assert await writer.write(payload) is True
908            # every byte arrives exactly once, so a resumed write never repeats itself
909            assert await reader == payload
910    finally:
911        reader.cancel()
912        with suppress(asyncio.CancelledError):
913            await reader
914        os.close(read_fd)
915        await writer.remove()
916
917
918@pytest.mark.asyncio
919async def test_pipe_write_outlasts_a_reader_slower_than_the_stall_timeout(
920    tmp_path: Path,
921) -> None:
922    """The stall budget covers a lack of progress, not the total time a write takes."""
923    pipe_path = tmp_path / "audio"
924    os.mkfifo(pipe_path)
925    writer = AsyncNamedPipeWriter(str(pipe_path))
926    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
927    payload = b"\x00" * 176400
928    stall_timeout = 0.15
929
930    async def drain_slowly() -> int:
931        received = 0
932        while received < len(payload):
933            await asyncio.sleep(0.02)
934            with suppress(BlockingIOError):
935                received += len(os.read(read_fd, 8192))
936        return received
937
938    reader = asyncio.create_task(drain_slowly())
939    loop = asyncio.get_running_loop()
940    started = loop.time()
941    try:
942        async with asyncio.timeout(30):
943            with patch("music_assistant.helpers.named_pipe.WRITE_STALL_TIMEOUT", stall_timeout):
944                assert await writer.write(payload) is True
945            assert await reader == len(payload)
946        # a reader this slow only completes because each pause resets the budget
947        assert loop.time() - started > stall_timeout
948    finally:
949        reader.cancel()
950        with suppress(asyncio.CancelledError):
951            await reader
952        os.close(read_fd)
953        await writer.remove()
954
955
956@pytest.mark.asyncio
957async def test_pipe_write_resumes_after_the_buffer_drains() -> None:
958    """A write that fills the pipe buffer continues once the reader catches up."""
959    data = b"\x00" * 10
960    writer = AsyncNamedPipeWriter("/tmp/audio")  # noqa: S108
961    writer._write_fd = 42
962
963    with (
964        patch(
965            "music_assistant.helpers.named_pipe.os.write",
966            side_effect=[4, BlockingIOError(errno.EAGAIN, "buffer full"), 6],
967        ) as write,
968        patch("music_assistant.helpers.named_pipe.select.poll") as poll,
969    ):
970        assert await writer.write(data) is True
971
972    # the retry picks up where the short write stopped, it never resends
973    assert write.call_args_list == [
974        call(42, memoryview(data)),
975        call(42, memoryview(data)[4:]),
976        call(42, memoryview(data)[4:]),
977    ]
978    poll.return_value.register.assert_called_once_with(42, select.POLLOUT)
979    poll.return_value.poll.assert_called_once_with(WRITE_POLL_INTERVAL_MS)
980
981
982@pytest.mark.asyncio
983async def test_pipe_write_gives_up_when_the_pipe_goes_before_the_first_write() -> None:
984    """A pipe removed the instant a write starts fails cleanly rather than blowing up."""
985    writer = AsyncNamedPipeWriter("/tmp/audio")  # noqa: S108
986
987    with (
988        patch.object(AsyncNamedPipeWriter, "_ensure_write_fd", return_value=True),
989        patch("music_assistant.helpers.named_pipe.os.write") as write,
990    ):
991        assert await writer.write(b"\x00" * 10) is False
992
993    write.assert_not_called()
994
995
996@pytest.mark.asyncio
997async def test_pipe_write_stops_when_the_descriptor_is_taken_mid_stall() -> None:
998    """A stalled write gives up once the pipe is removed out from under it."""
999    writer = AsyncNamedPipeWriter("/tmp/audio")  # noqa: S108
1000    writer._write_fd = 42
1001
1002    def take_descriptor(pipe_writer: AsyncNamedPipeWriter, _write_fd: int) -> None:
1003        pipe_writer._write_fd = None
1004
1005    with (
1006        patch(
1007            "music_assistant.helpers.named_pipe.os.write",
1008            side_effect=[4, BlockingIOError(errno.EAGAIN, "buffer full")],
1009        ) as write,
1010        patch.object(AsyncNamedPipeWriter, "_wait_writable", take_descriptor),
1011    ):
1012        assert await writer.write(b"\x00" * 10) is False
1013
1014    # writing on would land in whatever reopened that descriptor number
1015    assert write.call_count == 2
1016
1017
1018@pytest.mark.asyncio
1019async def test_pipe_write_resets_fd_when_the_reader_closes_during_a_stall(
1020    tmp_path: Path,
1021) -> None:
1022    """A reader that goes away while the buffer is full still resets the writer."""
1023    pipe_path = tmp_path / "audio"
1024    os.mkfifo(pipe_path)
1025    writer = AsyncNamedPipeWriter(str(pipe_path))
1026    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
1027
1028    async def close_reader_mid_stall() -> None:
1029        await asyncio.sleep(0.05)
1030        os.close(read_fd)
1031
1032    closer = asyncio.create_task(close_reader_mid_stall())
1033    try:
1034        async with asyncio.timeout(30):
1035            assert await writer.write(b"\x00" * 176400) is False
1036            await closer
1037        # the departed reader is what ends the write, so it must not be left mid-stall
1038        assert writer._write_fd is None
1039    finally:
1040        closer.cancel()
1041        with suppress(asyncio.CancelledError):
1042            await closer
1043        with suppress(OSError):
1044            os.close(read_fd)
1045        await writer.remove()
1046
1047
1048@pytest.mark.asyncio
1049async def test_command_pipe_wait_for_reader_resolves_when_the_reader_attaches(
1050    tmp_path: Path,
1051) -> None:
1052    """The reader wait ends as soon as the binary opens its end of the command pipe."""
1053    pipe_path = tmp_path / "commands"
1054    os.mkfifo(pipe_path)
1055    writer = AsyncNamedPipeWriter(str(pipe_path))
1056    read_fds: list[int] = []
1057
1058    async def attach_reader() -> None:
1059        await asyncio.sleep(0.01)
1060        read_fds.append(os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK))
1061
1062    reader = asyncio.create_task(attach_reader())
1063    try:
1064        assert await writer.wait_for_reader(timeout=1) is True
1065        assert read_fds  # resolved on the attach, not before it
1066    finally:
1067        await reader
1068        for read_fd in read_fds:
1069            os.close(read_fd)
1070        await writer.remove()
1071
1072
1073@pytest.mark.asyncio
1074async def test_command_pipe_wait_for_reader_gives_up_at_its_timeout(tmp_path: Path) -> None:
1075    """A command pipe nothing ever reads fails the wait within the requested window."""
1076    pipe_path = tmp_path / "commands"
1077    os.mkfifo(pipe_path)
1078    writer = AsyncNamedPipeWriter(str(pipe_path))
1079    loop = asyncio.get_running_loop()
1080
1081    started = loop.time()
1082    assert await writer.wait_for_reader(timeout=0.1) is False
1083
1084    assert 0.1 <= loop.time() - started < 1
1085
1086
1087@pytest.mark.asyncio
1088async def test_command_pipe_wait_for_reader_uses_no_worker_thread(tmp_path: Path) -> None:
1089    """Waiting for the binary's reader never strands a blocking worker thread."""
1090    pipe_path = tmp_path / "commands"
1091    os.mkfifo(pipe_path)
1092    writer = AsyncNamedPipeWriter(str(pipe_path))
1093
1094    with (
1095        patch(
1096            "music_assistant.helpers.named_pipe.os.open",
1097            side_effect=OSError(errno.ENXIO, "no reader"),
1098        ),
1099        patch(
1100            "music_assistant.helpers.named_pipe.asyncio.to_thread",
1101            new_callable=AsyncMock,
1102        ) as to_thread,
1103    ):
1104        assert await writer.wait_for_reader(timeout=0.1) is False
1105
1106    to_thread.assert_not_awaited()
1107
1108
1109@pytest.mark.asyncio
1110async def test_command_pipe_wait_for_reader_defers_to_an_in_flight_write(tmp_path: Path) -> None:
1111    """The reader wait leaves the descriptor to a write that is already under way."""
1112    pipe_path = tmp_path / "commands"
1113    os.mkfifo(pipe_path)
1114    writer = AsyncNamedPipeWriter(str(pipe_path))
1115    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
1116
1117    try:
1118        await writer._write_lock.acquire()
1119        wait = asyncio.create_task(writer.wait_for_reader(timeout=1))
1120        await asyncio.sleep(0)
1121
1122        # a reader is attached the whole time, so only the write lock holds this up
1123        assert not wait.done()
1124
1125        writer._write_lock.release()
1126        assert await wait is True
1127    finally:
1128        os.close(read_fd)
1129        await writer.remove()
1130
1131
1132@pytest.mark.asyncio
1133async def test_command_pipe_write_reuses_the_fd_from_the_reader_wait(tmp_path: Path) -> None:
1134    """A command written after the reader wait rides the descriptor that wait opened."""
1135    pipe_path = tmp_path / "commands"
1136    os.mkfifo(pipe_path)
1137    writer = AsyncNamedPipeWriter(str(pipe_path))
1138    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
1139
1140    try:
1141        assert await writer.wait_for_reader(timeout=1) is True
1142
1143        with patch("music_assistant.helpers.named_pipe.os.open") as open_pipe:
1144            assert await writer.write(b"ACTION=STANDBY\n") is True
1145
1146        open_pipe.assert_not_called()
1147        assert os.read(read_fd, 64) == b"ACTION=STANDBY\n"
1148    finally:
1149        os.close(read_fd)
1150        await writer.remove()
1151
1152
1153@pytest.mark.asyncio
1154async def test_cli_command_updates_timestamp_after_successful_delivery() -> None:
1155    """A delivered command updates the player's last command timestamp."""
1156    player = _make_player()
1157    player.last_command_sent = 10.0
1158    stream = AirPlayStream(player)
1159    stream._cli_proc = _make_cli_proc()
1160
1161    with (
1162        patch.object(stream.commands_pipe, "write", new=AsyncMock(return_value=True)),
1163        patch("music_assistant.providers.airplay.stream.time.time", return_value=20.0),
1164    ):
1165        assert await stream.send_cli_command("ACTION=STANDBY") is True
1166
1167    assert player.last_command_sent == 20.0
1168
1169
1170@pytest.mark.asyncio
1171async def test_cli_command_preserves_timestamp_when_delivery_fails() -> None:
1172    """A dropped command leaves the player's last command timestamp unchanged."""
1173    player = _make_player()
1174    player.last_command_sent = 10.0
1175    stream = AirPlayStream(player)
1176    stream._cli_proc = _make_cli_proc()
1177
1178    with patch.object(stream.commands_pipe, "write", new=AsyncMock(return_value=False)):
1179        assert await stream.send_cli_command("ACTION=STANDBY") is False
1180
1181    assert player.last_command_sent == 10.0
1182
1183
1184@pytest.mark.asyncio
1185async def test_cli_command_preserves_timestamp_when_delivery_raises() -> None:
1186    """A command write error leaves the player's last command timestamp unchanged."""
1187    player = _make_player()
1188    player.last_command_sent = 10.0
1189    stream = AirPlayStream(player)
1190    stream._cli_proc = _make_cli_proc()
1191
1192    with (
1193        patch.object(
1194            stream.commands_pipe,
1195            "write",
1196            new=AsyncMock(side_effect=OSError("command pipe failed")),
1197        ),
1198        pytest.raises(OSError, match="command pipe failed"),
1199    ):
1200        await stream.send_cli_command("ACTION=STANDBY")
1201
1202    assert player.last_command_sent == 10.0
1203
1204
1205@pytest.mark.asyncio
1206async def test_delivered_volume_command_arms_the_echo_grace() -> None:
1207    """
1208    A delivered volume command makes the player ignore the echo of it.
1209
1210    The receiver reports every level it is handed back over DACP; read at face value
1211    that echo is the user reaching for the volume and is written straight back out.
1212    """
1213    player = _make_player()
1214    stream = AirPlayStream(player)
1215    stream._cli_proc = _make_cli_proc()
1216
1217    with patch.object(stream.commands_pipe, "write", new=AsyncMock(return_value=True)):
1218        assert await stream.send_cli_command("VOLUME=40") is True
1219
1220    player.suppress_volume_reports.assert_called_once_with()
1221
1222
1223@pytest.mark.asyncio
1224async def test_delivered_other_command_leaves_the_echo_grace_alone() -> None:
1225    """A command that is not a level cannot come back as one, so it blinds nothing."""
1226    player = _make_player()
1227    stream = AirPlayStream(player)
1228    stream._cli_proc = _make_cli_proc()
1229
1230    with patch.object(stream.commands_pipe, "write", new=AsyncMock(return_value=True)):
1231        assert await stream.send_cli_command("ACTION=STANDBY") is True
1232
1233    player.suppress_volume_reports.assert_not_called()
1234
1235
1236@pytest.mark.asyncio
1237async def test_dropped_volume_command_leaves_the_echo_grace_alone() -> None:
1238    """A volume command the binary never got is never echoed, so the reports stay live."""
1239    player = _make_player()
1240    stream = AirPlayStream(player)
1241    stream._cli_proc = _make_cli_proc()
1242
1243    with patch.object(stream.commands_pipe, "write", new=AsyncMock(return_value=False)):
1244        assert await stream.send_cli_command("VOLUME=40") is False
1245
1246    player.suppress_volume_reports.assert_not_called()
1247
1248
1249@pytest.mark.asyncio
1250async def test_connect_does_not_write_before_the_binary_can_read() -> None:
1251    """Connecting lays out the command pipe and starts cliairplay, pushing nothing into it."""
1252    player = _make_player()
1253    stream = AirPlayStream(player)
1254    # media is available, so a push that never happens is by design and not a missing source
1255    player.current_media = MagicMock(corrected_elapsed_time=12.5)
1256    process = MagicMock(closed=False)
1257    process.start = AsyncMock(return_value=None)
1258    operation_order: list[str] = []
1259
1260    async def create_pipe() -> None:
1261        operation_order.append("pipe")
1262
1263    async def start_process() -> None:
1264        operation_order.append("process")
1265
1266    def consume_task(awaitable: Any) -> MagicMock:
1267        awaitable.close()
1268        task = MagicMock()
1269        task.done.return_value = True
1270        return task
1271
1272    process.start.side_effect = start_process
1273    player.provider.mass.create_task.side_effect = consume_task
1274    with (
1275        patch.object(stream, "_build_cli_args", new_callable=AsyncMock, return_value=["binary"]),
1276        patch(
1277            "music_assistant.providers.airplay.stream.AsyncProcess",
1278            return_value=process,
1279        ),
1280        patch.object(stream.commands_pipe, "create", side_effect=create_pipe),
1281        patch.object(stream, "send_metadata", new_callable=AsyncMock) as send_metadata_mock,
1282    ):
1283        await stream.connect()
1284
1285    assert operation_order == ["pipe", "process"]
1286    send_metadata_mock.assert_not_awaited()
1287
1288
1289@pytest.mark.asyncio
1290async def test_connect_failure_cleans_up_process_and_pipe() -> None:
1291    """A cliairplay process that fails to start cannot leave a live process or FIFO."""
1292    player = _make_player()
1293    stream = AirPlayStream(player)
1294    process = MagicMock(closed=False)
1295    process.start = AsyncMock(side_effect=OSError("process start failed"))
1296    process.kill = AsyncMock()
1297
1298    with (
1299        patch.object(stream, "_build_cli_args", new_callable=AsyncMock, return_value=["binary"]),
1300        patch(
1301            "music_assistant.providers.airplay.stream.AsyncProcess",
1302            return_value=process,
1303        ),
1304        patch.object(stream.commands_pipe, "create", new_callable=AsyncMock),
1305        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
1306        pytest.raises(OSError, match="process start failed"),
1307    ):
1308        await stream.connect()
1309
1310    process.kill.assert_awaited_once()
1311    remove_pipe.assert_awaited_once()
1312    assert stream._cli_proc is None
1313    assert stream._cleanup_complete is True
1314
1315
1316@pytest.mark.asyncio
1317async def test_start_sends_command_and_stamps_position() -> None:
1318    """START is delivered over the command pipe and stamps the media position."""
1319    player = _make_player()
1320    stream = AirPlayStream(player)
1321    stream._cli_proc = _make_cli_proc()
1322    stream._connected.set()
1323
1324    with patch.object(
1325        stream,
1326        "_write_cli_command",
1327        new_callable=AsyncMock,
1328        side_effect=_acking_write_cli_command(stream),
1329    ) as write_command:
1330        assert await stream.start(START_UNIX_MS, 12_000) == START_UNIX_MS
1331
1332    write_command.assert_awaited_once_with(f"START_UNIX_MS={START_UNIX_MS}\nACTION=START")
1333    assert stream._start_position == 12.0
1334    player.set_state_from_stream.assert_called_once_with(elapsed_time=12.0, stream=stream)
1335
1336
1337@pytest.mark.asyncio
1338async def test_start_join_marks_the_command() -> None:
1339    """A late-join START carries START_JOIN=1 so the binary enforces clock readiness."""
1340    player = _make_player()
1341    stream = AirPlayStream(player)
1342    stream._cli_proc = _make_cli_proc()
1343    stream._connected.set()
1344
1345    with patch.object(
1346        stream,
1347        "_write_cli_command",
1348        new_callable=AsyncMock,
1349        side_effect=_acking_write_cli_command(stream),
1350    ) as write_command:
1351        assert await stream.start(START_UNIX_MS, 0, join=True) == START_UNIX_MS
1352
1353    write_command.assert_awaited_once_with(
1354        f"START_UNIX_MS={START_UNIX_MS}\nSTART_JOIN=1\nACTION=START"
1355    )
1356
1357
1358@pytest.mark.asyncio
1359@pytest.mark.parametrize(
1360    "complete_before_start",
1361    [True, False],
1362    ids=["delivered", "rendering"],
1363)
1364async def test_start_transition_artwork_settled_or_retried(complete_before_start: bool) -> None:
1365    """START keeps already-delivered transition artwork settled and retries a superseded render."""
1366    player = _make_player()
1367    stream = AirPlayStream(player)
1368    stream._cli_proc = _make_cli_proc()
1369    stream._connected.set()
1370    metadata = MagicMock(
1371        corrected_elapsed_time=0,
1372        queue_item_id="item-new",
1373        title="New track",
1374        artist="Artist",
1375        album="Album",
1376        duration=180,
1377        image_url="new-image",
1378    )
1379    stream.session = MagicMock(media=metadata)
1380    render_started = asyncio.Event()
1381    release_render = asyncio.Event()
1382    metadata_tasks: list[asyncio.Task[Any]] = []
1383    render_count = 0
1384
1385    def create_task(target: Any, **_kwargs: Any) -> asyncio.Task[Any]:
1386        task = asyncio.create_task(target())
1387        metadata_tasks.append(task)
1388        return task
1389
1390    async def prepare_artwork(_image_url: str, _generation: int) -> str:
1391        nonlocal render_count
1392        render_count += 1
1393        if render_count == 1:
1394            render_started.set()
1395            if not complete_before_start:
1396                await release_render.wait()
1397            return "/cache/pretransition.jpg"
1398        return "/cache/posttransition.jpg"
1399
1400    player.provider.mass.create_task.side_effect = create_task
1401    with (
1402        patch.object(
1403            stream,
1404            "_write_cli_command",
1405            new_callable=AsyncMock,
1406            side_effect=_acking_write_cli_command(stream),
1407        ) as write_command,
1408        patch.object(
1409            stream,
1410            "_prepare_artwork",
1411            new_callable=AsyncMock,
1412            side_effect=prepare_artwork,
1413        ),
1414        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
1415    ):
1416        pretransition_task = asyncio.create_task(stream.send_metadata(None, metadata))
1417        await render_started.wait()
1418        if complete_before_start:
1419            await pretransition_task
1420        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
1421        await asyncio.gather(*metadata_tasks)
1422        release_render.set()
1423        await pretransition_task
1424
1425    commands = [args.args[0] for args in write_command.await_args_list]
1426    start_command = f"START_UNIX_MS={START_UNIX_MS}\nACTION=START"
1427    if complete_before_start:
1428        # artwork delivered inside the transition bundle stays settled;
1429        # re-pushing it around the START would make an Apple TV re-render
1430        # its screen
1431        assert commands[-1] == start_command
1432        assert any("ARTWORKFILE=/cache/pretransition.jpg" in command for command in commands)
1433        assert not any(command.startswith("ARTWORK=") for command in commands)
1434    else:
1435        # the anchor superseded the in-flight render; the post-anchor push
1436        # renders again and delivers the artwork once
1437        assert commands[-2:] == [start_command, "ARTWORK=/cache/posttransition.jpg"]
1438        assert "ARTWORK=/cache/pretransition.jpg" not in commands
1439    assert stream._metadata_generation == 2
1440    assert stream._metadata_artwork_checksum == "new-image"
1441
1442
1443@pytest.mark.asyncio
1444async def test_start_requires_connected_process() -> None:
1445    """START is rejected without a connected cliairplay process."""
1446    stream = AirPlayStream(_make_player())
1447    stream._cli_proc = _make_cli_proc()
1448    # not connected: _connected event never set
1449
1450    with (
1451        patch.object(stream, "_write_cli_command", new_callable=AsyncMock) as write_command,
1452        pytest.raises(RuntimeError, match="without a connected cliairplay process"),
1453    ):
1454        await stream.start(START_UNIX_MS, 0)
1455
1456    write_command.assert_not_awaited()
1457
1458
1459@pytest.mark.asyncio
1460async def test_flush_sends_command_and_awaits_ack() -> None:
1461    """FLUSH is delivered and resolves once the binary reports it flushed."""
1462    stream = AirPlayStream(_make_player())
1463    stream._cli_proc = _make_cli_proc()
1464    stream._connected.set()
1465
1466    with patch.object(
1467        stream, "_write_cli_command", new_callable=AsyncMock, return_value=True
1468    ) as write_command:
1469        flush_task = asyncio.create_task(stream.flush())
1470        await asyncio.sleep(0)
1471        assert stream._handle_status_line("[STATUS] flushed") is False
1472        assert await flush_task is True
1473
1474    write_command.assert_awaited_once_with("ACTION=FLUSH")
1475
1476
1477@pytest.mark.asyncio
1478async def test_flush_times_out_without_ack() -> None:
1479    """FLUSH returns False when the binary never acknowledges it."""
1480    stream = AirPlayStream(_make_player())
1481    stream._cli_proc = _make_cli_proc()
1482    stream._connected.set()
1483
1484    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1485        assert await stream.flush(timeout=0) is False
1486
1487
1488@pytest.mark.asyncio
1489async def test_flush_returns_false_when_command_not_delivered() -> None:
1490    """FLUSH reports failure when the command cannot be delivered."""
1491    stream = AirPlayStream(_make_player())
1492    stream._cli_proc = _make_cli_proc()
1493    stream._connected.set()
1494
1495    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=False):
1496        assert await stream.flush() is False
1497
1498
1499@pytest.mark.asyncio
1500async def test_start_raises_when_command_not_delivered() -> None:
1501    """A dropped START surfaces as an error so the caller can fall back cold."""
1502    stream = AirPlayStream(_make_player())
1503    stream._cli_proc = _make_cli_proc()
1504    stream._connected.set()
1505
1506    with (
1507        patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=False),
1508        pytest.raises(PlayerCommandFailed, match="Could not deliver START"),
1509    ):
1510        await stream.start(1_750_000_000_000, 0)
1511
1512
1513@pytest.mark.asyncio
1514async def test_start_fails_fast_on_reported_start_failure() -> None:
1515    """A reported start failure ends the ack wait at once instead of timing out."""
1516    stream = AirPlayStream(_make_player())
1517    stream._cli_proc = _make_cli_proc()
1518    stream._connected.set()
1519
1520    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1521        start_task = asyncio.create_task(stream.start(START_UNIX_MS, 0, join=True))
1522        await asyncio.sleep(0)
1523        stream._handle_status_line(
1524            '[STATUS] error code=start_failed http=0 detail="no live session to start"'
1525        )
1526        with pytest.raises(PlayerCommandFailed, match="no live session to start"):
1527            await start_task
1528
1529    # a command failure must not poison how a NEW connection is reported
1530    assert stream._connect_error is None
1531
1532
1533@pytest.mark.asyncio
1534async def test_flush_fails_fast_on_reported_flush_failure() -> None:
1535    """A reported flush failure resolves the ack wait as a failure, not a timeout."""
1536    stream = AirPlayStream(_make_player())
1537    stream._cli_proc = _make_cli_proc()
1538    stream._connected.set()
1539
1540    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1541        flush_task = asyncio.create_task(stream.flush())
1542        await asyncio.sleep(0)
1543        stream._handle_status_line(
1544            '[STATUS] error code=flush_failed http=0 detail="session rejected the flush"'
1545        )
1546        assert await flush_task is False
1547
1548    assert stream._connect_error is None
1549
1550
1551@pytest.mark.asyncio
1552async def test_start_failure_does_not_outlive_its_command() -> None:
1553    """A failed START leaves no error behind that would fail the next one."""
1554    stream = AirPlayStream(_make_player())
1555    stream._cli_proc = _make_cli_proc()
1556    stream._connected.set()
1557    stream._handle_status_line("[STATUS] error code=start_failed")
1558
1559    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1560        start_task = asyncio.create_task(stream.start(START_UNIX_MS, 0))
1561        await asyncio.sleep(0)
1562        stream._handle_status_line(
1563            f"[STATUS] started requested_unix_ms={START_UNIX_MS} at_unix_ms={START_UNIX_MS}"
1564        )
1565        assert await start_task == START_UNIX_MS
1566
1567
1568@pytest.mark.asyncio
1569async def test_start_returns_the_instant_the_binary_scheduled() -> None:
1570    """A corrected ack, not the commanded instant, is what the caller maps content onto."""
1571    stream = AirPlayStream(_make_player())
1572    stream._cli_proc = _make_cli_proc()
1573    stream._connected.set()
1574    corrected = START_UNIX_MS + 700
1575
1576    with patch.object(
1577        stream,
1578        "_write_cli_command",
1579        new_callable=AsyncMock,
1580        side_effect=_acking_write_cli_command(stream, corrected),
1581    ):
1582        assert await stream.start(START_UNIX_MS, 0) == corrected
1583
1584
1585@pytest.mark.asyncio
1586async def test_start_fails_when_the_ack_never_arrives() -> None:
1587    """An unacknowledged START fails: nothing may be mapped onto an unconfirmed instant."""
1588    stream = AirPlayStream(_make_player())
1589    stream._cli_proc = _make_cli_proc()
1590    stream._connected.set()
1591
1592    with (
1593        patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True),
1594        patch("music_assistant.providers.airplay.stream.AIRPLAY_START_ACK_TIMEOUT_MS", 10),
1595        pytest.raises(PlayerCommandFailed, match="did not acknowledge its start") as err,
1596    ):
1597        await stream.start(START_UNIX_MS, 0)
1598
1599    # the player and the instant nothing confirmed are the whole diagnostic
1600    assert "Player A" in str(err.value)
1601    assert str(START_UNIX_MS) in str(err.value)
1602
1603
1604@pytest.mark.asyncio
1605async def test_start_accepts_a_malformed_ack_as_the_commanded_instant() -> None:
1606    """An ack that cannot be parsed still answered the START, so the commanded instant stands."""
1607    stream = AirPlayStream(_make_player())
1608    stream._cli_proc = _make_cli_proc()
1609    stream._connected.set()
1610
1611    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1612        start_task = asyncio.create_task(stream.start(START_UNIX_MS, 0))
1613        await asyncio.sleep(0)
1614        stream._handle_status_line("[STATUS] started requested_unix_ms=nonsense at_unix_ms=")
1615        assert await start_task == START_UNIX_MS
1616
1617
1618@pytest.mark.asyncio
1619async def test_start_treats_a_missing_scheduled_instant_as_malformed() -> None:
1620    """An ack without at_unix_ms parses cleanly to 0, which must never be returned."""
1621    stream = AirPlayStream(_make_player())
1622    stream._cli_proc = _make_cli_proc()
1623    stream._connected.set()
1624
1625    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1626        start_task = asyncio.create_task(stream.start(START_UNIX_MS, 0))
1627        await asyncio.sleep(0)
1628        stream._handle_status_line(f"[STATUS] started requested_unix_ms={START_UNIX_MS}")
1629        assert await start_task == START_UNIX_MS
1630
1631
1632@pytest.mark.asyncio
1633@pytest.mark.parametrize(
1634    ("join", "expected_timeout"),
1635    [
1636        (True, AIRPLAY_JOIN_START_ACK_TIMEOUT_MS / 1000),
1637        (False, AIRPLAY_START_ACK_TIMEOUT_MS / 1000),
1638    ],
1639    ids=["join", "plain"],
1640)
1641async def test_start_ack_window_matches_the_arm(join: bool, expected_timeout: float) -> None:
1642    """Each START uses the acknowledgement window assigned to its contract."""
1643    stream = AirPlayStream(_make_player())
1644    stream._cli_proc = _make_cli_proc()
1645    stream._connected.set()
1646    timeouts: list[float] = []
1647
1648    async def record_timeout(awaitable: Any, timeout: float) -> None:
1649        timeouts.append(timeout)
1650        awaitable.close()
1651        raise TimeoutError
1652
1653    with (
1654        patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True),
1655        patch(
1656            "music_assistant.providers.airplay.stream.asyncio.wait_for",
1657            side_effect=record_timeout,
1658        ),
1659        pytest.raises(PlayerCommandFailed),
1660    ):
1661        await stream.start(START_UNIX_MS, 0, join=join)
1662
1663    assert timeouts == [expected_timeout]
1664
1665
1666@pytest.mark.parametrize(
1667    "timeout_ms",
1668    [AIRPLAY_START_ACK_TIMEOUT_MS, AIRPLAY_JOIN_START_ACK_TIMEOUT_MS],
1669    ids=["plain", "join"],
1670)
1671def test_start_ack_window_covers_buffered_anchor_retries(timeout_ms: int) -> None:
1672    """The acknowledgement window outlives cliairplay's buffered anchor retry span."""
1673    assert timeout_ms > 5_500
1674
1675
1676@pytest.mark.asyncio
1677async def test_flush_returns_false_when_not_connected() -> None:
1678    """FLUSH is a no-op returning False before the device connects."""
1679    stream = AirPlayStream(_make_player())
1680    stream._cli_proc = _make_cli_proc()
1681    # not connected
1682
1683    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock) as write_command:
1684        assert await stream.flush() is False
1685
1686    write_command.assert_not_awaited()
1687
1688
1689def test_flushed_status_sets_flush_event() -> None:
1690    """The [STATUS] flushed line releases the flush acknowledgement event."""
1691    stream = AirPlayStream(_make_player())
1692    assert not stream._flushed.is_set()
1693
1694    assert stream._handle_status_line("[STATUS] flushed") is False
1695
1696    assert stream._flushed.is_set()
1697
1698
1699@pytest.mark.asyncio
1700async def test_flush_holds_stdin_quiet_before_commanding_the_flush() -> None:
1701    """
1702    Queued stdin audio is cleared before FLUSH, so the binary's drain removes it.
1703
1704    The command travels on a pipe of its own, so audio still in flight when the
1705    binary drains would survive to be anchored as the next start's first sample.
1706    """
1707    calls: list[str] = []
1708    stream = AirPlayStream(_make_player())
1709    stream._cli_proc = _make_cli_proc(calls=calls)
1710    stream._connected.set()
1711
1712    async def _record_command(command: str) -> bool:
1713        calls.append(command)
1714        return True
1715
1716    with patch.object(stream, "_write_cli_command", side_effect=_record_command):
1717        flush_task = asyncio.create_task(stream.flush())
1718        await asyncio.sleep(0)
1719        assert stream._handle_status_line("[STATUS] flushed") is False
1720        assert await flush_task is True
1721
1722    assert calls == ["quiesce", "ACTION=FLUSH"]
1723
1724
1725@pytest.mark.asyncio
1726async def test_flush_fails_when_queued_audio_cannot_be_cleared() -> None:
1727    """A drain that never completes fails the flush instead of anchoring stale audio."""
1728    stream = AirPlayStream(_make_player())
1729    stream._cli_proc = _make_cli_proc(quiesced=False)
1730    stream._connected.set()
1731
1732    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock) as write_command:
1733        assert await stream.flush() is False
1734
1735    write_command.assert_not_awaited()
1736
1737
1738def test_audio_status_records_the_pending_stdin_depth() -> None:
1739    """The [STATUS] audio line reports how much audio is pending on the binary's stdin."""
1740    stream = AirPlayStream(_make_player())
1741    assert stream.audio_pending_ms == 0
1742
1743    assert stream._handle_status_line("[STATUS] audio buffered_ms=92") is False
1744
1745    assert stream.audio_pending_ms == 92
1746
1747
1748def test_unparsable_audio_status_reports_no_pending_audio() -> None:
1749    """A malformed depth reports none rather than carrying the previous value."""
1750    stream = AirPlayStream(_make_player())
1751    stream._handle_status_line("[STATUS] audio buffered_ms=92")
1752
1753    assert stream._handle_status_line("[STATUS] audio buffered_ms=nonsense") is False
1754
1755    assert stream.audio_pending_ms == 0
1756
1757
1758def test_audio_status_without_a_depth_reports_no_pending_audio() -> None:
1759    """A line omitting the depth reports none rather than carrying the previous value."""
1760    stream = AirPlayStream(_make_player())
1761    stream._handle_status_line("[STATUS] audio buffered_ms=92")
1762
1763    assert stream._handle_status_line("[STATUS] audio ") is False
1764
1765    assert stream.audio_pending_ms == 0
1766
1767
1768@pytest.mark.asyncio
1769async def test_flush_clears_the_pending_stdin_depth() -> None:
1770    """A flush drops the previous cycle's depth so the next report describes the new one."""
1771    stream = AirPlayStream(_make_player())
1772    stream._cli_proc = _make_cli_proc()
1773    stream._connected.set()
1774    stream._handle_status_line("[STATUS] audio buffered_ms=92")
1775
1776    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1777        flush_task = asyncio.create_task(stream.flush())
1778        await asyncio.sleep(0)
1779        stream._handle_status_line("[STATUS] flushed")
1780        assert await flush_task is True
1781
1782    assert stream.audio_pending_ms == 0
1783
1784
1785def test_announce_started_status_records_instant_and_duration() -> None:
1786    """The started report carries the ACTUAL audible instant and the clip duration."""
1787    stream = AirPlayStream(_make_player())
1788    assert not stream._announce_started.is_set()
1789
1790    assert (
1791        stream._handle_status_line(
1792            f"[STATUS] announce_started at_unix_ms={START_UNIX_MS} duration_ms=1800"
1793        )
1794        is False
1795    )
1796
1797    assert stream._announce_started.is_set()
1798    assert stream._announce_ack == (START_UNIX_MS, 1800)
1799
1800
1801def test_announce_started_status_with_unusable_values_reports_zeroes() -> None:
1802    """Unusable fields land on 0 (unreported) instead of failing the whole answer."""
1803    stream = AirPlayStream(_make_player())
1804
1805    stream._handle_status_line("[STATUS] announce_started at_unix_ms=nonsense")
1806
1807    assert stream._announce_started.is_set()
1808    assert stream._announce_ack == (0, 0)
1809
1810
1811@pytest.mark.parametrize(
1812    ("line", "cancelled"),
1813    [
1814        ("[STATUS] announce_done", False),
1815        ("[STATUS] announce_done cancelled=1", True),
1816    ],
1817    ids=["completed", "cancelled"],
1818)
1819def test_announce_done_status_sets_done_and_cancelled(line: str, cancelled: bool) -> None:
1820    """The done report releases the done wait, carrying whether the clip was cut short."""
1821    stream = AirPlayStream(_make_player())
1822    assert not stream._announce_done.is_set()
1823
1824    assert stream._handle_status_line(line) is False
1825
1826    assert stream._announce_done.is_set()
1827    assert stream._announce_done_cancelled is cancelled
1828
1829
1830def test_announce_failed_is_routed_to_the_announce_waiter() -> None:
1831    """A rejected arm answers both announce waits at once and stays off the connect error."""
1832    stream = AirPlayStream(_make_player())
1833
1834    stream._handle_status_line('[STATUS] error code=announce_failed http=0 detail="not playing"')
1835
1836    assert stream._announce_started.is_set()
1837    assert stream._announce_done.is_set()
1838    assert stream._announce_error is not None
1839    assert stream._announce_error.detail == "not playing"
1840    # a command failure must not poison how a NEW connection is reported
1841    assert stream._connect_error is None
1842
1843
1844@pytest.mark.asyncio
1845async def test_announce_sends_the_arm_command() -> None:
1846    """ANNOUNCE is delivered as the four-line arm the binary expects."""
1847    stream = AirPlayStream(_make_player())
1848    stream._cli_proc = _make_cli_proc()
1849    stream._connected.set()
1850
1851    with patch.object(
1852        stream, "_write_cli_command", new_callable=AsyncMock, return_value=True
1853    ) as write_command:
1854        assert await stream.announce("/fake/clip.pcm", START_UNIX_MS, -12) is True
1855
1856    write_command.assert_awaited_once_with(
1857        f"ANNOUNCE_FILE=/fake/clip.pcm\nANNOUNCE_AT_UNIX_MS={START_UNIX_MS}\n"
1858        "ANNOUNCE_DUCK_DB=-12\nACTION=ANNOUNCE"
1859    )
1860
1861
1862@pytest.mark.asyncio
1863async def test_announce_resets_the_previous_answer() -> None:
1864    """Arming clears every slot so only this arm's answer is read."""
1865    stream = AirPlayStream(_make_player())
1866    stream._cli_proc = _make_cli_proc()
1867    stream._connected.set()
1868    stream._handle_status_line("[STATUS] announce_started at_unix_ms=5 duration_ms=6")
1869    stream._handle_status_line("[STATUS] announce_done cancelled=1")
1870    stream._handle_status_line("[STATUS] error code=announce_failed")
1871
1872    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1873        assert await stream.announce("/fake/clip.pcm", 0, -12) is True
1874
1875    assert not stream._announce_started.is_set()
1876    assert not stream._announce_done.is_set()
1877    assert stream._announce_ack is None
1878    assert stream._announce_error is None
1879    assert stream._announce_done_cancelled is False
1880
1881
1882@pytest.mark.asyncio
1883async def test_announce_requires_a_running_connected_stream() -> None:
1884    """An arm on a stream that is not up is refused without touching the pipe."""
1885    stream = AirPlayStream(_make_player())
1886    stream._cli_proc = _make_cli_proc()
1887    # not connected
1888
1889    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock) as write_command:
1890        assert await stream.announce("/fake/clip.pcm", 0, -12) is False
1891
1892    write_command.assert_not_awaited()
1893
1894
1895@pytest.mark.asyncio
1896async def test_wait_announce_started_returns_the_ack() -> None:
1897    """The started wait hands back the acked instant and duration."""
1898    stream = AirPlayStream(_make_player())
1899
1900    wait_task = asyncio.create_task(stream.wait_announce_started(1.0))
1901    await asyncio.sleep(0)
1902    stream._handle_status_line(
1903        f"[STATUS] announce_started at_unix_ms={START_UNIX_MS} duration_ms=900"
1904    )
1905
1906    assert await wait_task == (START_UNIX_MS, 900)
1907
1908
1909@pytest.mark.asyncio
1910async def test_wait_announce_started_resolves_on_done_without_started() -> None:
1911    """A done (cancelled) without a start means the clip never played: None, right away."""
1912    stream = AirPlayStream(_make_player())
1913
1914    wait_task = asyncio.create_task(stream.wait_announce_started(30.0))
1915    await asyncio.sleep(0)
1916    stream._handle_status_line("[STATUS] announce_done cancelled=1")
1917
1918    assert await wait_task is None
1919
1920
1921@pytest.mark.asyncio
1922async def test_wait_announce_started_times_out_on_a_silent_binary() -> None:
1923    """An outdated binary ignores the arm entirely; the bounded wait returns None."""
1924    stream = AirPlayStream(_make_player())
1925
1926    assert await stream.wait_announce_started(0) is None
1927
1928
1929@pytest.mark.asyncio
1930async def test_wait_announce_started_returns_none_on_reported_failure() -> None:
1931    """A reported announce failure answers the started wait as a failure, not a timeout."""
1932    stream = AirPlayStream(_make_player())
1933
1934    wait_task = asyncio.create_task(stream.wait_announce_started(30.0))
1935    await asyncio.sleep(0)
1936    stream._handle_status_line("[STATUS] error code=announce_failed")
1937
1938    assert await wait_task is None
1939
1940
1941@pytest.mark.asyncio
1942@pytest.mark.parametrize(
1943    ("line", "expected"),
1944    [
1945        ("[STATUS] announce_done", True),
1946        ("[STATUS] announce_done cancelled=1", False),
1947        ("[STATUS] error code=announce_failed", False),
1948    ],
1949    ids=["completed", "cancelled", "failed"],
1950)
1951async def test_wait_announce_done_outcomes(line: str, expected: bool) -> None:
1952    """Only a completed clip resolves the done wait as True."""
1953    stream = AirPlayStream(_make_player())
1954
1955    wait_task = asyncio.create_task(stream.wait_announce_done(1.0))
1956    await asyncio.sleep(0)
1957    stream._handle_status_line(line)
1958
1959    assert await wait_task is expected
1960
1961
1962@pytest.mark.asyncio
1963async def test_wait_announce_done_times_out() -> None:
1964    """The done wait stays bounded (eof can end the status stream mid-clip)."""
1965    stream = AirPlayStream(_make_player())
1966
1967    assert await stream.wait_announce_done(0) is False
1968
1969
1970@pytest.mark.asyncio
1971async def test_cli_args_streaming_mode_lanes() -> None:
1972    """
1973    The streaming-mode pin maps onto the binary's protocol/timing arguments.
1974
1975    The timing lanes ride --timing on a forced airplay2 protocol, the compat
1976    mode forces the auth-setup + RAOP flow, RAOP arrives through the protocol
1977    override exactly as before, and Automatic leaves the route to the binary.
1978    """
1979    player = _make_player()
1980    player.streaming_mode = STREAMING_MODE_AP2_NTP
1981    args = await _build_args(player)
1982    assert _arg_value(args, "--protocol") == "airplay2"
1983    assert _arg_value(args, "--timing") == "ntp"
1984
1985    player = _make_player()
1986    player.streaming_mode = STREAMING_MODE_AP2_PTP
1987    args = await _build_args(player)
1988    assert _arg_value(args, "--protocol") == "airplay2"
1989    assert _arg_value(args, "--timing") == "ptp"
1990
1991    player = _make_player()
1992    player.streaming_mode = STREAMING_MODE_AP2_COMPAT
1993    args = await _build_args(player)
1994    assert _arg_value(args, "--protocol") == "airplay2-compat"
1995    assert "--timing" not in args
1996
1997    player = _make_player()
1998    args = await _build_args(player)
1999    assert _arg_value(args, "--protocol") == "auto"
2000    assert "--timing" not in args
2001
2002    player = _make_player()
2003    player.streaming_mode = STREAMING_MODE_RAOP
2004    player.protocol_override = StreamingProtocol.RAOP
2005    player.protocol = StreamingProtocol.RAOP
2006    args = await _build_args(player)
2007    assert _arg_value(args, "--protocol") == "raop"
2008    assert "--timing" not in args
2009
2010
2011@pytest.mark.asyncio
2012async def test_clock_stall_switches_solo_auto_player_to_ntp() -> None:
2013    """
2014    A measured PTP stall on a solo Automatic player self-heals onto NTP.
2015
2016    The visible streaming-mode setting is written (so the user can see and
2017    revert the decision) and a playback restart is scheduled; a synced member
2018    or a pinned mode only gets the warning.
2019    """
2020    player = _make_player()
2021    stream = AirPlayStream(player)
2022    stream._handle_status_line(
2023        "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
2024        "ready_in_ms=0 ready_at_unix_ms=0"
2025    )
2026    mass = player.provider.mass
2027    mass.config.set_raw_player_config_value.assert_called_once_with(
2028        player.player_id, CONF_STREAMING_MODE, STREAMING_MODE_AP2_NTP
2029    )
2030    assert mass.create_task.called
2031
2032    # A grouped member is reported, never moved: restarting one member of a
2033    # live sync group would desync it.
2034    grouped_player = _make_player()
2035    grouped_player.synced_to = "apleader"
2036    grouped = AirPlayStream(grouped_player)
2037    grouped._handle_status_line(
2038        "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
2039        "ready_in_ms=0 ready_at_unix_ms=0"
2040    )
2041    grouped_player.provider.mass.config.set_raw_player_config_value.assert_not_called()
2042
2043    # An explicitly pinned mode is the user's choice: warn only.
2044    pinned_player = _make_player()
2045    pinned_player.streaming_mode = STREAMING_MODE_AP2_PTP
2046    pinned = AirPlayStream(pinned_player)
2047    pinned._handle_status_line(
2048        "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
2049        "ready_in_ms=0 ready_at_unix_ms=0"
2050    )
2051    pinned_player.provider.mass.config.set_raw_player_config_value.assert_not_called()
2052
2053
2054def test_native_control_failure_switches_automatic_player_to_compatibility() -> None:
2055    """A terminal native control failure persists the compatibility route once."""
2056    player = _make_player()
2057    stream = AirPlayStream(player)
2058
2059    stream._handle_status_line("[ERROR] AirPlay 2 control channel failed")
2060    stream._handle_status_line("[ERROR] AirPlay 2 control channel failed")
2061
2062    player.provider.mass.config.set_raw_player_config_value.assert_called_once_with(
2063        player.player_id, CONF_STREAMING_MODE, STREAMING_MODE_AP2_COMPAT
2064    )
2065    player.provider.mass.create_task.assert_not_called()
2066
2067
2068def test_native_control_failure_does_not_override_pinned_mode() -> None:
2069    """A terminal native control failure leaves an explicit streaming mode unchanged."""
2070    player = _make_player()
2071    player.streaming_mode = STREAMING_MODE_AP2_PTP
2072    stream = AirPlayStream(player)
2073
2074    stream._handle_status_line("[ERROR] AirPlay 2 control channel failed")
2075
2076    player.provider.mass.config.set_raw_player_config_value.assert_not_called()
2077    player.provider.mass.create_task.assert_not_called()
2078
2079
2080def test_unrelated_cli_error_does_not_switch_to_compatibility() -> None:
2081    """A different runtime failure does not diagnose the native control route."""
2082    player = _make_player()
2083    stream = AirPlayStream(player)
2084
2085    stream._handle_status_line("[ERROR] AirPlay 2 audio send failed")
2086
2087    player.provider.mass.config.set_raw_player_config_value.assert_not_called()
2088
2089
2090@pytest.mark.asyncio
2091async def test_clock_ready_projection_resolves_the_wait() -> None:
2092    """A probing receiver reports when its clock becomes usable, from its first probe."""
2093    stream = AirPlayStream(_make_player())
2094
2095    assert (
2096        stream._handle_status_line(
2097            "[STATUS] clock_ready mode=ptp state=probing streak_ms=0 exchanges=1 "
2098            f"ready_in_ms=2300 ready_at_unix_ms={START_UNIX_MS}"
2099        )
2100        is False
2101    )
2102
2103    assert await stream.wait_clock_ready(timeout=0.01) == (
2104        ClockReadiness.PROJECTED,
2105        START_UNIX_MS,
2106    )
2107
2108
2109@pytest.mark.asyncio
2110async def test_clock_ready_cold_line_keeps_waiting_for_a_projection() -> None:
2111    """A receiver that has not probed yet carries no projection, so the wait goes on."""
2112    stream = AirPlayStream(_make_player())
2113
2114    stream._handle_status_line(
2115        "[STATUS] clock_ready mode=ptp state=cold streak_ms=0 exchanges=0 "
2116        "ready_in_ms=0 ready_at_unix_ms=0"
2117    )
2118
2119    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.UNREPORTED, 0)
2120    assert not stream._clock_ready.is_set()
2121
2122    stream._handle_status_line(
2123        "[STATUS] clock_ready mode=ptp state=ready streak_ms=2400 exchanges=9 "
2124        f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
2125    )
2126
2127    assert await stream.wait_clock_ready(timeout=0.01) == (
2128        ClockReadiness.PROJECTED,
2129        START_UNIX_MS,
2130    )
2131
2132
2133@pytest.mark.asyncio
2134async def test_clock_ready_ntp_resolves_without_a_projection() -> None:
2135    """NTP timing has no receiver clock to wait for, so the wait ends with nothing."""
2136    stream = AirPlayStream(_make_player())
2137
2138    stream._handle_status_line(
2139        "[STATUS] clock_ready mode=ntp state=ready streak_ms=0 exchanges=0 "
2140        f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
2141    )
2142
2143    assert stream._clock_ready.is_set()
2144    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.NOT_APPLICABLE, 0)
2145
2146
2147@pytest.mark.asyncio
2148async def test_wait_clock_ready_times_out_for_a_binary_that_never_reports() -> None:
2149    """A binary that does not report readiness is told apart from one that answered."""
2150    stream = AirPlayStream(_make_player())
2151
2152    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.UNREPORTED, 0)
2153
2154
2155@pytest.mark.asyncio
2156async def test_clock_ready_stalled_state_warns_once(caplog: pytest.LogCaptureFixture) -> None:
2157    """
2158    A stalled receiver that cannot be self-healed is reported loudly, once.
2159
2160    A grouped member is never auto-switched (moving one member of a live sync
2161    group would desync it), so it takes the warn-only path; the solo Automatic
2162    self-heal has its own test.
2163    """
2164    grouped_player = _make_player()
2165    grouped_player.synced_to = "apleader"
2166    stream = AirPlayStream(grouped_player)
2167
2168    with caplog.at_level(logging.DEBUG):
2169        ended = stream._handle_status_line(
2170            "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
2171            "ready_in_ms=0 ready_at_unix_ms=0"
2172        )
2173        stream._handle_status_line(
2174            "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
2175            "ready_in_ms=0 ready_at_unix_ms=0"
2176        )
2177
2178    warnings = [record for record in caplog.records if record.levelno == logging.WARNING]
2179    assert ended is False
2180    assert len(warnings) == 1
2181    assert "Player A" in warnings[0].getMessage()
2182    assert "319/320" in warnings[0].getMessage()
2183    assert stream._clock_ready.is_set()
2184    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.STALLED, 0)
2185
2186
2187def test_clock_ready_stall_warning_is_ptp_only(caplog: pytest.LogCaptureFixture) -> None:
2188    """An NTP-timed session has no clock of ours to answer, so it never reads as a stall."""
2189    stream = AirPlayStream(_make_player())
2190
2191    with caplog.at_level(logging.DEBUG):
2192        stream._handle_status_line(
2193            "[STATUS] clock_ready mode=ntp state=stalled streak_ms=0 exchanges=0 "
2194            f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
2195        )
2196
2197    assert [record for record in caplog.records if record.levelno >= logging.WARNING] == []
2198    assert stream._clock_ready_at_unix_ms == 0
2199
2200
2201@pytest.mark.parametrize("state", ["cold", "probing", "ready"])
2202def test_clock_ready_handshake_states_do_not_warn(
2203    state: str, caplog: pytest.LogCaptureFixture
2204) -> None:
2205    """Any state but a stall is a normal step of the clock handshake and stays quiet."""
2206    stream = AirPlayStream(_make_player())
2207
2208    with caplog.at_level(logging.DEBUG):
2209        ended = stream._handle_status_line(
2210            f"[STATUS] clock_ready mode=ptp state={state} streak_ms=900 exchanges=4 "
2211            f"ready_in_ms=940 ready_at_unix_ms={START_UNIX_MS}"
2212        )
2213
2214    assert ended is False
2215    assert [record for record in caplog.records if record.levelno >= logging.WARNING] == []
2216
2217
2218def test_elapsed_includes_start_position() -> None:
2219    """Reported progress is the current anchor's media base plus the binary delta."""
2220    player = _make_player()
2221    stream = AirPlayStream(player)
2222    stream._start_position = 12.0
2223
2224    stream._update_elapsed(1.5)
2225
2226    player.set_state_from_stream.assert_called_once_with(
2227        state=PlaybackState.PLAYING,
2228        elapsed_time=13.5,
2229        stream=stream,
2230    )
2231
2232
2233def test_reanchor_status_sets_the_cumulative_total() -> None:
2234    """Each [STATUS] REANCHOR line carries the authoritative total, so it SETS the shift."""
2235    stream = AirPlayStream(_make_player())
2236    assert stream.cumulative_shift_seconds == 0.0
2237
2238    assert (
2239        stream._handle_status_line(
2240            "[STATUS] REANCHOR shifted_frames=67870 total_shifted_frames=67870 sample_rate=44100"
2241        )
2242        is False
2243    )
2244    assert stream.cumulative_shift_seconds == pytest.approx(67870 / 44100)
2245
2246    # the second event reports the running total, never a delta to add
2247    stream._handle_status_line(
2248        "[STATUS] REANCHOR shifted_frames=67870 total_shifted_frames=135740 sample_rate=44100"
2249    )
2250    assert stream.cumulative_shift_seconds == pytest.approx(135740 / 44100)
2251
2252
2253def test_human_readable_reanchor_warn_is_not_counted() -> None:
2254    """The binary's warn line accompanies the status line; counting it would double it."""
2255    stream = AirPlayStream(_make_player())
2256
2257    ended = stream._handle_status_line(
2258        "[AP2] Re-anchored after PCM starvation: shifted_frames=67870 lead_frames=77175 count=1"
2259    )
2260
2261    assert ended is False
2262    assert stream.cumulative_shift_seconds == 0.0
2263
2264
2265def test_reanchor_status_prefers_line_sample_rate() -> None:
2266    """The sample rate carried on the [STATUS] REANCHOR line wins over the stream format."""
2267    player = _make_player()
2268    stream = AirPlayStream(player)
2269    stream.pcm_format = AudioFormat(
2270        content_type=ContentType.PCM_S16LE, sample_rate=48000, bit_depth=16
2271    )
2272
2273    stream._handle_status_line(
2274        "[STATUS] REANCHOR shifted_frames=44100 total_shifted_frames=44100 sample_rate=44100"
2275    )
2276
2277    # converts at 44100 (from the line), not 48000 (the stream format) -> exactly 1.0s
2278    assert stream.cumulative_shift_seconds == pytest.approx(1.0)
2279
2280
2281def test_reanchor_status_ignores_line_without_total() -> None:
2282    """A [STATUS] REANCHOR line missing the total leaves the shift unchanged."""
2283    stream = AirPlayStream(_make_player())
2284    stream.cumulative_shift_seconds = 1.5
2285
2286    stream._handle_status_line("[STATUS] REANCHOR shifted_frames=67870 sample_rate=44100")
2287
2288    assert stream.cumulative_shift_seconds == 1.5
2289
2290
2291@pytest.mark.asyncio
2292async def test_start_resets_reanchor_shift() -> None:
2293    """A START re-anchors from scratch, clearing the accumulated shift."""
2294    stream = AirPlayStream(_make_player())
2295    stream._cli_proc = _make_cli_proc()
2296    stream._connected.set()
2297    stream.cumulative_shift_seconds = 3.078
2298
2299    with patch.object(
2300        stream,
2301        "_write_cli_command",
2302        new_callable=AsyncMock,
2303        side_effect=_acking_write_cli_command(stream),
2304    ):
2305        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
2306
2307    assert stream.cumulative_shift_seconds == 0.0
2308
2309
2310@pytest.mark.asyncio
2311async def test_connect_resets_accumulated_shift() -> None:
2312    """A fresh cliairplay process starts from a zero playout shift."""
2313    player = _make_player()
2314    stream = AirPlayStream(player)
2315    stream.cumulative_shift_seconds = 5.0
2316    process = MagicMock(closed=False)
2317    process.start = AsyncMock(return_value=None)
2318
2319    def consume_task(awaitable: Any) -> MagicMock:
2320        awaitable.close()
2321        task = MagicMock()
2322        task.done.return_value = True
2323        return task
2324
2325    player.provider.mass.create_task.side_effect = consume_task
2326    with (
2327        patch.object(stream, "_build_cli_args", new_callable=AsyncMock, return_value=["binary"]),
2328        patch("music_assistant.providers.airplay.stream.AsyncProcess", return_value=process),
2329        patch.object(stream.commands_pipe, "create", new_callable=AsyncMock),
2330    ):
2331        await stream.connect()
2332
2333    assert stream.cumulative_shift_seconds == 0.0
2334
2335
2336@pytest.mark.asyncio
2337async def test_initial_metadata_skips_artwork() -> None:
2338    """The pre-connect metadata push cannot delay setup on artwork rendering."""
2339    player = _make_player()
2340    stream = AirPlayStream(player)
2341    stream._cli_proc = _make_cli_proc()
2342    metadata = MagicMock(
2343        title="Track",
2344        artist="Artist",
2345        album="Album",
2346        duration=180,
2347        image_url="image",
2348    )
2349
2350    with (
2351        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2352        patch.object(
2353            stream,
2354            "_render_and_send_artwork",
2355            new_callable=AsyncMock,
2356        ) as send_artwork,
2357    ):
2358        await stream.send_metadata(0, metadata, send_artwork=False)
2359
2360    # the metadata push resets the device position to zero, so a push at the
2361    # start of a track needs no separate progress correction
2362    assert send_command.await_count == 1
2363    assert "TITLE=Track" in send_command.await_args_list[0].args[0]
2364    assert stream._last_progress_sent == 0
2365    send_artwork.assert_not_awaited()
2366
2367
2368@pytest.mark.asyncio
2369async def test_wait_for_connection_pushes_metadata_immediately() -> None:
2370    """
2371    Track metadata is pushed the instant the binary can receive it.
2372
2373    Receivers that gate audio rendering on receiving timeline-anchored metadata
2374    (e.g. Sonos over native AirPlay 2) must not be left silent while a deferred
2375    push is pending, so the metadata callback runs synchronously on connect
2376    while only the volume resend stays on the delayed path.
2377    """
2378    player = _make_player()
2379    player.volume_muted = False
2380    stream = AirPlayStream(player)
2381    stream._connected.set()  # connection already established
2382    player.provider.mass.call_later = MagicMock()
2383    operation_order: list[str] = []
2384
2385    async def wait_for_reader(_timeout: float) -> bool:
2386        operation_order.append("reader")
2387        return True
2388
2389    async def send_current_metadata(**_kwargs: Any) -> None:
2390        operation_order.append("metadata")
2391
2392    with (
2393        patch.object(stream, "_cli_proc", MagicMock()),  # non-None so the method proceeds
2394        patch.object(stream.commands_pipe, "wait_for_reader", side_effect=wait_for_reader),
2395        patch.object(stream, "_send_current_metadata", side_effect=send_current_metadata),
2396        patch.object(stream, "send_cli_command", return_value=None),  # avoid a real coroutine
2397    ):
2398        await stream.wait_for_connection()
2399
2400    # Nothing is written before the binary has a reader on the command pipe.
2401    assert operation_order == ["reader", "metadata"]
2402    # Metadata pushed synchronously on connect...
2403    player.on_player_media_updated.assert_called_once_with()
2404    # ...and never routed through the delayed call_later path.
2405    deferred_callables = [call.args[1] for call in player.provider.mass.call_later.call_args_list]
2406    assert player.on_player_media_updated not in deferred_callables
2407    # The volume resend is still deferred (existing behavior preserved).
2408    assert player.provider.mass.call_later.call_count == 1
2409    assert player.provider.mass.call_later.call_args_list[0].args[0] == 2
2410
2411
2412@pytest.mark.asyncio
2413async def test_deferred_volume_resend_reads_the_state_when_it_fires() -> None:
2414    """The repeated volume push must carry the volume at fire time, not at connect time."""
2415    player = _make_player()
2416    player.volume_muted = False
2417    player.volume_level = 40
2418    stream = AirPlayStream(player)
2419    stream._connected.set()  # connection already established
2420    player.provider.mass.call_later = MagicMock()
2421
2422    with (
2423        patch.object(stream, "_cli_proc", MagicMock()),  # non-None so the method proceeds
2424        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=True)),
2425        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2426        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2427    ):
2428        await stream.wait_for_connection()
2429        send_command.assert_awaited_with("VOLUME=40")
2430        deferred = player.provider.mass.call_later.call_args_list[0].args[1]
2431
2432        # a volume change between the connect and the resend firing must survive
2433        player.volume_level = 75
2434        await deferred()
2435        send_command.assert_awaited_with("VOLUME=75")
2436
2437        # ...and so must a mute
2438        player.volume_muted = True
2439        await deferred()
2440
2441    send_command.assert_awaited_with("VOLUME=0")
2442
2443
2444@pytest.mark.asyncio
2445async def test_wait_for_connection_sends_volume_when_player_owns_it() -> None:
2446    """The initial volume push is sent when this output owns its own volume."""
2447    player = _make_player()
2448    player.owns_volume = True
2449    player.volume_muted = False
2450    stream = AirPlayStream(player)
2451    stream._connected.set()
2452
2453    with (
2454        patch.object(stream, "_cli_proc", MagicMock()),
2455        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=True)),
2456        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2457        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2458    ):
2459        await stream.wait_for_connection()
2460
2461    send_command.assert_awaited_with(f"VOLUME={player.volume_level}")
2462
2463
2464@pytest.mark.asyncio
2465async def test_wait_for_connection_skips_volume_when_another_control_owns_it() -> None:
2466    """No unsolicited volume push when another control owns this output's volume."""
2467    player = _make_player()
2468    player.owns_volume = False
2469    player.volume_muted = False
2470    stream = AirPlayStream(player)
2471    stream._connected.set()
2472
2473    with (
2474        patch.object(stream, "_cli_proc", MagicMock()),
2475        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=True)),
2476        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2477        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2478    ):
2479        await stream.wait_for_connection()
2480
2481    send_command.assert_not_awaited()
2482
2483
2484@pytest.mark.asyncio
2485async def test_wait_for_connection_sends_volume_when_muted_without_ownership() -> None:
2486    """A latched mute is still pushed even when another control owns the volume."""
2487    player = _make_player()
2488    player.owns_volume = False
2489    player.volume_muted = True
2490    stream = AirPlayStream(player)
2491    stream._connected.set()
2492
2493    with (
2494        patch.object(stream, "_cli_proc", MagicMock()),
2495        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=True)),
2496        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2497        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2498    ):
2499        await stream.wait_for_connection()
2500
2501    send_command.assert_awaited_with("VOLUME=0")
2502
2503
2504@pytest.mark.asyncio
2505async def test_wait_for_connection_fails_on_an_unread_command_pipe() -> None:
2506    """A binary that never attaches to the command pipe can never be anchored: fail the connect."""
2507    player = _make_player()
2508    player.logger = MagicMock()
2509    player.volume_muted = False
2510    stream = AirPlayStream(player)
2511    stream._connected.set()
2512    stream._cli_proc = _make_cli_proc()
2513
2514    with (
2515        patch.object(
2516            stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=False)
2517        ) as wait_for_reader,
2518        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2519        patch.object(stream, "send_cli_command", return_value=None),
2520        pytest.raises(PlayerCommandFailed, match="command pipe") as err,
2521    ):
2522        await stream.wait_for_connection()
2523
2524    wait_for_reader.assert_awaited_once()
2525    assert player.display_name in str(err.value)
2526
2527
2528@pytest.mark.asyncio
2529async def test_wait_for_connection_stays_quiet_about_a_stopped_stream() -> None:
2530    """A stream torn down while connecting took its command pipe along, which is no fault."""
2531    player = _make_player()
2532    player.logger = MagicMock()
2533    player.volume_muted = False
2534    stream = AirPlayStream(player)
2535    stream._connected.set()
2536    stream._cli_proc = _make_cli_proc()
2537    stream._stopping = True
2538
2539    with (
2540        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=False)),
2541        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2542        patch.object(stream, "send_cli_command", return_value=None),
2543    ):
2544        await stream.wait_for_connection()
2545
2546    player.logger.warning.assert_not_called()
2547
2548
2549@pytest.mark.asyncio
2550async def test_prepare_artwork_returns_cache_path() -> None:
2551    """Artwork preparation returns the shared cache path without a per-player copy."""
2552    player = _make_player()
2553    stream = AirPlayStream(player)
2554    image_url = "https://example.com/artwork.png"
2555    cached_path = "/cache/thumbnails/artwork_flat.jpg"
2556
2557    with patch(
2558        "music_assistant.providers.airplay.stream.get_image_thumb_path",
2559        new=AsyncMock(return_value=cached_path),
2560    ) as get_thumb_path:
2561        result = await stream._prepare_artwork(image_url, 1)
2562
2563    assert result == cached_path
2564    assert not hasattr(stream, "_artwork_paths")
2565    get_thumb_path.assert_awaited_once_with(
2566        stream.mass,
2567        image_url,
2568        AIRPLAY_ARTWORK_SIZE,
2569        "",
2570        image_format="JPEG",
2571        flatten_transparency=True,
2572    )
2573
2574
2575@pytest.mark.asyncio
2576async def test_stop_cleans_up_when_stop_command_fails() -> None:
2577    """A command-pipe failure cannot skip process and stream cleanup."""
2578    player = _make_player()
2579    stream = AirPlayStream(player)
2580    process = MagicMock()
2581    process.closed = False
2582    process.kill = AsyncMock()
2583    stream._cli_proc = process
2584
2585    with (
2586        patch.object(
2587            stream.commands_pipe,
2588            "write",
2589            new_callable=AsyncMock,
2590            side_effect=OSError("command pipe failed"),
2591        ),
2592        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
2593        pytest.raises(OSError, match="command pipe failed"),
2594    ):
2595        await stream.stop(force=True)
2596
2597    assert stream._stopped is True
2598    assert stream._cleanup_complete is True
2599    remove_pipe.assert_awaited_once()
2600    process.kill.assert_awaited_once()
2601    player.set_state_from_stream.assert_called_once_with(
2602        state=PlaybackState.IDLE,
2603        elapsed_time=0,
2604        stream=stream,
2605    )
2606
2607
2608@pytest.mark.asyncio
2609async def test_stop_awaits_cancelled_stdout_reader() -> None:
2610    """Stream teardown waits for the stdout reader to release process resources."""
2611    player = _make_player()
2612    stream = AirPlayStream(player)
2613    process = MagicMock()
2614    process.closed = False
2615    process.kill = AsyncMock()
2616    stream._cli_proc = process
2617    reader_started = asyncio.Event()
2618
2619    async def _stdout_reader() -> None:
2620        reader_started.set()
2621        await asyncio.Event().wait()
2622
2623    reader_task = asyncio.create_task(_stdout_reader())
2624    stream._stdout_reader_task = reader_task
2625    await reader_started.wait()
2626
2627    with (
2628        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock),
2629        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2630    ):
2631        await stream.stop(force=True)
2632
2633    assert reader_task.cancelled()
2634    process.kill.assert_awaited_once()
2635
2636
2637@pytest.mark.asyncio
2638async def test_force_stop_does_not_wait_for_artwork_render() -> None:
2639    """Force-stop tears down immediately while remote artwork rendering finishes."""
2640    player = _make_player()
2641    stream = AirPlayStream(player)
2642    process = MagicMock()
2643    process.closed = False
2644    process.kill = AsyncMock()
2645    stream._cli_proc = process
2646    metadata = MagicMock(
2647        title="Track",
2648        artist="Artist",
2649        album="Album",
2650        duration=180,
2651        image_url="slow-image",
2652    )
2653    artwork_started = asyncio.Event()
2654    release_artwork = asyncio.Event()
2655
2656    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2657        artwork_started.set()
2658        await release_artwork.wait()
2659        return "late.jpg"
2660
2661    with (
2662        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock),
2663        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2664        patch.object(
2665            stream,
2666            "_prepare_artwork",
2667            new_callable=AsyncMock,
2668            side_effect=_prepare_artwork,
2669        ),
2670    ):
2671        metadata_task = asyncio.create_task(stream.send_metadata(0, metadata))
2672        await artwork_started.wait()
2673        await asyncio.wait_for(stream.stop(force=True), timeout=0.5)
2674        release_artwork.set()
2675        await metadata_task
2676
2677    assert stream._cleanup_complete is True
2678    process.kill.assert_awaited_once()
2679
2680
2681@pytest.mark.asyncio
2682async def test_process_eof_cleans_up_command_pipe() -> None:
2683    """A naturally ended CLI stream removes its command pipe."""
2684    player = _make_player()
2685    stream = AirPlayStream(player)
2686    process = MagicMock()
2687    stream._cli_proc = process
2688
2689    async def _stderr_lines() -> AsyncGenerator[str]:
2690        yield "[STATUS] eof"
2691
2692    with (
2693        patch.object(process, "iter_stderr", return_value=_stderr_lines()),
2694        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
2695    ):
2696        await stream._stderr_reader()
2697
2698    assert stream._stopped is True
2699    remove_pipe.assert_awaited_once()
2700    player.schedule_group_rejoin.assert_not_called()
2701
2702
2703async def _run_unexpected_process_death(player: MagicMock) -> AirPlayStream:
2704    """Drive the stderr reader through an unexpected process exit."""
2705    stream = AirPlayStream(player)
2706    process = MagicMock()
2707    stream._cli_proc = process
2708
2709    async def _stderr_lines() -> AsyncGenerator[str]:
2710        yield "some final log line"
2711
2712    with (
2713        patch.object(process, "iter_stderr", return_value=_stderr_lines()),
2714        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2715    ):
2716        await stream._stderr_reader()
2717    return stream
2718
2719
2720@pytest.mark.asyncio
2721async def test_unexpected_death_of_synced_child_schedules_rejoin() -> None:
2722    """A grouped member whose process dies unexpectedly gets a re-join scheduled."""
2723    player = _make_player()
2724    player.synced_to = "leader"
2725    player.group_members = []
2726    # the leader's other members are captured as fallback candidates in case
2727    # leadership transfers while the re-join backoff runs
2728    leader = MagicMock()
2729    leader.group_members = ["leader", player.player_id, "sibling"]
2730    player.provider.mass.players.get_player.return_value = leader
2731
2732    stream = await _run_unexpected_process_death(player)
2733
2734    player.schedule_group_rejoin.assert_called_once_with(["leader", "sibling"])
2735    player.set_state_from_stream.assert_called_once_with(
2736        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2737    )
2738
2739
2740@pytest.mark.asyncio
2741async def test_unexpected_death_of_leader_schedules_rejoin_to_members() -> None:
2742    """A dying leader re-joins towards its surviving members (leadership transfers)."""
2743    player = _make_player()
2744    player.synced_to = None
2745    player.group_members = [player.player_id, "child1", "child2"]
2746
2747    await _run_unexpected_process_death(player)
2748
2749    player.schedule_group_rejoin.assert_called_once_with(["child1", "child2"])
2750    # the controller sets the leader's final state (transfer or dissolve)
2751    player.set_state_from_stream.assert_not_called()
2752
2753
2754@pytest.mark.asyncio
2755async def test_unexpected_death_of_solo_player_schedules_no_rejoin() -> None:
2756    """An ungrouped player's process death only marks the player idle."""
2757    player = _make_player()
2758    player.synced_to = None
2759    player.group_members = []
2760
2761    stream = await _run_unexpected_process_death(player)
2762
2763    player.schedule_group_rejoin.assert_not_called()
2764    player.set_state_from_stream.assert_called_once_with(
2765        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2766    )
2767
2768
2769@pytest.mark.asyncio
2770async def test_unexpected_death_of_static_group_member_drops_member_only() -> None:
2771    """A static group member's death drops just that member, never the whole group."""
2772    player = _make_player()
2773    player.synced_to = "leader"
2774    player.group_members = []
2775    # the player is a static member of an actively playing group player, for
2776    # which cmd_ungroup would release (stop) the WHOLE group
2777    player.state.active_group = "syncgroup1"
2778    group_player = MagicMock()
2779    group_player.static_group_members = ["leader", player.player_id]
2780    leader = MagicMock()
2781    leader.group_members = ["leader", player.player_id]
2782    player.provider.mass.players.get_player.side_effect = lambda player_id: {
2783        "syncgroup1": group_player,
2784        "leader": leader,
2785    }.get(player_id)
2786
2787    await _run_unexpected_process_death(player)
2788
2789    players_controller = player.provider.mass.players
2790    players_controller.cmd_set_members.assert_called_once_with(
2791        "leader", player_ids_to_remove=[player.player_id]
2792    )
2793    players_controller.cmd_ungroup.assert_not_called()
2794    player.schedule_group_rejoin.assert_called_once_with(["leader"])
2795
2796
2797@pytest.mark.asyncio
2798async def test_process_eof_during_render_does_not_send_artwork() -> None:
2799    """A cache lookup finishing after EOF cannot send stale artwork."""
2800    player = _make_player()
2801    stream = AirPlayStream(player)
2802    render_started = asyncio.Event()
2803    release_render = asyncio.Event()
2804
2805    async def _get_image_thumb_path(*_args: Any, **_kwargs: Any) -> str:
2806        render_started.set()
2807        await release_render.wait()
2808        return "/cache/thumbnails/artwork.jpg"
2809
2810    with (
2811        patch(
2812            "music_assistant.providers.airplay.stream.get_image_thumb_path",
2813            new_callable=AsyncMock,
2814            side_effect=_get_image_thumb_path,
2815        ),
2816        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2817    ):
2818        render_task = asyncio.create_task(stream._render_and_send_artwork("image", 1))
2819        await render_started.wait()
2820        stream._stopped = True
2821        release_render.set()
2822        await render_task
2823
2824    send_command.assert_not_awaited()
2825
2826
2827@pytest.mark.asyncio
2828async def test_concurrent_metadata_updates_only_send_latest_artwork() -> None:
2829    """An older slow artwork render cannot overwrite a newer track update."""
2830    player = _make_player()
2831    stream = AirPlayStream(player)
2832    process = MagicMock()
2833    process.closed = False
2834    stream._cli_proc = process
2835    first_render_started = asyncio.Event()
2836    release_first_render = asyncio.Event()
2837
2838    old_metadata = MagicMock(
2839        title="Old track",
2840        artist="Artist",
2841        album="Album",
2842        duration=180,
2843        image_url="old-image",
2844    )
2845    new_metadata = MagicMock(
2846        title="New track",
2847        artist="Artist",
2848        album="Album",
2849        duration=180,
2850        image_url="new-image",
2851    )
2852
2853    async def _prepare_artwork(image_url: str, _generation: int) -> str:
2854        if image_url == "old-image":
2855            first_render_started.set()
2856            await release_first_render.wait()
2857            return "old.jpg"
2858        return "new.jpg"
2859
2860    with (
2861        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2862        patch.object(
2863            stream,
2864            "_prepare_artwork",
2865            new_callable=AsyncMock,
2866            side_effect=_prepare_artwork,
2867        ),
2868        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
2869    ):
2870        old_task = asyncio.create_task(stream.send_metadata(0, old_metadata))
2871        await first_render_started.wait()
2872        new_task = asyncio.create_task(stream.send_metadata(0, new_metadata))
2873        # the new update supersedes the old render once the old push's render
2874        # budget lapses and the metadata lock is released
2875        await new_task
2876        assert stream._metadata_generation == 2
2877        release_first_render.set()
2878        await old_task
2879
2880    commands = [call.args[0].decode() for call in write_command.await_args_list]
2881    assert any("TITLE=New track" in command for command in commands)
2882    assert not any("ARTWORK=old.jpg" in command for command in commands)
2883    assert "ARTWORKFILE=new.jpg\n" in commands[-1]
2884    assert commands[-1].endswith("ACTION=SENDMETA\n")
2885
2886
2887@pytest.mark.asyncio
2888async def test_artwork_url_form_change_does_not_resend_artwork() -> None:
2889    """Alternating URL forms of the same imageproxy image send artwork only once."""
2890    player = _make_player()
2891    stream = AirPlayStream(player)
2892    stream._cli_proc = _make_cli_proc()
2893
2894    def make_metadata(image_url: str) -> MagicMock:
2895        return MagicMock(
2896            corrected_elapsed_time=0,
2897            queue_item_id="item-1",
2898            title="Track",
2899            artist="Artist",
2900            album="Album",
2901            duration=180,
2902            image_url=image_url,
2903        )
2904
2905    # the queue session builds the image URL on the stream server base, the
2906    # player state on the webserver base - same image id behind both forms
2907    image_id = "ab" * 32
2908    other_image_id = "cd" * 32
2909    session_media = make_metadata(
2910        f"http://192.168.1.5:8097/imageproxy/{image_id}?size=512&fmt=jpeg"
2911    )
2912    state_media = make_metadata(f"http://192.168.1.5:8095/imageproxy/{image_id}?size=512&fmt=png")
2913    other_image_media = make_metadata(
2914        f"http://192.168.1.5:8095/imageproxy/{other_image_id}?size=512&fmt=png"
2915    )
2916
2917    with (
2918        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2919        patch.object(
2920            stream,
2921            "_prepare_artwork",
2922            new_callable=AsyncMock,
2923            return_value="/cache/thumb.jpg",
2924        ) as prepare_artwork,
2925    ):
2926        await stream.send_metadata(None, session_media)
2927        generation_after_first_send = stream._metadata_generation
2928        # the post-START push (session media) and the media-updated push
2929        # (player state) alternate on every seek
2930        await stream.send_metadata(None, state_media)
2931        await stream.send_metadata(None, session_media)
2932        assert stream._metadata_generation == generation_after_first_send
2933        await stream.send_metadata(None, other_image_media)
2934
2935    commands = [call.args[0].decode() for call in write_command.await_args_list]
2936    bundled = [command for command in commands if "ARTWORKFILE=" in command]
2937    resends = [command for command in commands if command.startswith("ARTWORK=")]
2938    assert len(bundled) == 1
2939    assert resends == ["ARTWORK=/cache/thumb.jpg\n"]
2940    assert prepare_artwork.await_count == 2
2941    assert stream._metadata_artwork_checksum == other_image_id
2942
2943
2944@pytest.mark.asyncio
2945async def test_metadata_revert_resends_text_after_superseded_artwork() -> None:
2946    """Reverting while artwork renders restores the previously displayed track text."""
2947    player = _make_player()
2948    stream = AirPlayStream(player)
2949    process = MagicMock()
2950    process.closed = False
2951    stream._cli_proc = process
2952    first_metadata = MagicMock(
2953        title="First track",
2954        artist="Artist",
2955        album="Album",
2956        duration=180,
2957        image_url=None,
2958    )
2959    second_metadata = MagicMock(
2960        title="Second track",
2961        artist="Artist",
2962        album="Album",
2963        duration=180,
2964        image_url="second-image",
2965    )
2966    first_checksum = "First track|Artist|Album|180|None"
2967    stream._metadata_text_checksum = first_checksum
2968    stream._pending_metadata_checksum = first_checksum
2969    artwork_started = asyncio.Event()
2970    release_artwork = asyncio.Event()
2971
2972    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2973        artwork_started.set()
2974        await release_artwork.wait()
2975        return "second.jpg"
2976
2977    with (
2978        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2979        patch.object(
2980            stream,
2981            "_prepare_artwork",
2982            new_callable=AsyncMock,
2983            side_effect=_prepare_artwork,
2984        ),
2985    ):
2986        second_task = asyncio.create_task(stream.send_metadata(0, second_metadata))
2987        await artwork_started.wait()
2988        revert_task = asyncio.create_task(stream.send_metadata(0, first_metadata))
2989        await asyncio.sleep(0)
2990        release_artwork.set()
2991        await asyncio.gather(second_task, revert_task)
2992
2993    metadata_commands = [
2994        call.args[0].decode()
2995        for call in write_command.await_args_list
2996        if "ACTION=SENDMETA" in call.args[0].decode()
2997    ]
2998    assert "TITLE=Second track" in metadata_commands[0]
2999    assert "TITLE=First track" in metadata_commands[-1]
3000
3001
3002@pytest.mark.asyncio
3003async def test_repeated_metadata_retries_superseded_artwork() -> None:
3004    """A B-to-C-to-B update sequence still applies B artwork after supersession."""
3005    player = _make_player()
3006    stream = AirPlayStream(player)
3007    process = MagicMock()
3008    process.closed = False
3009    stream._cli_proc = process
3010    initial_text_checksum = "item-initial|Initial|Artist|Album"
3011    initial_checksum = f"{initial_text_checksum}|initial-image"
3012    stream._metadata_artwork_checksum = "initial-image"
3013    stream._metadata_text_checksum = initial_text_checksum
3014    stream._pending_metadata_checksum = initial_checksum
3015    metadata_b = MagicMock(
3016        queue_item_id="item-b",
3017        title="Track B",
3018        artist="Artist",
3019        album="Album",
3020        duration=180,
3021        image_url="b-image",
3022    )
3023    metadata_c = MagicMock(
3024        queue_item_id="item-c",
3025        title="Track C",
3026        artist="Artist",
3027        album="Album",
3028        duration=180,
3029        image_url="c-image",
3030    )
3031    first_artwork_started = asyncio.Event()
3032    release_first_artwork = asyncio.Event()
3033    c_artwork_started = asyncio.Event()
3034    release_c_artwork = asyncio.Event()
3035    b_render_count = 0
3036
3037    async def _prepare_artwork(image_url: str, _generation: int) -> str:
3038        nonlocal b_render_count
3039        if image_url == "b-image":
3040            b_render_count += 1
3041            if b_render_count == 1:
3042                first_artwork_started.set()
3043                await release_first_artwork.wait()
3044                return "b-stale.jpg"
3045            return "b-final.jpg"
3046        c_artwork_started.set()
3047        await release_c_artwork.wait()
3048        return "c.jpg"
3049
3050    with (
3051        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3052        patch.object(
3053            stream,
3054            "_prepare_artwork",
3055            new_callable=AsyncMock,
3056            side_effect=_prepare_artwork,
3057        ) as prepare_artwork,
3058        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
3059    ):
3060        first_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
3061        await first_artwork_started.wait()
3062        c_task = asyncio.create_task(stream.send_metadata(0, metadata_c))
3063        await c_artwork_started.wait()
3064        final_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
3065        await final_b_task
3066        release_first_artwork.set()
3067        release_c_artwork.set()
3068        await asyncio.gather(first_b_task, c_task)
3069
3070    rendered_images = [args.args[0] for args in prepare_artwork.await_args_list]
3071    commands = [args.args[0].decode() for args in write_command.await_args_list]
3072    assert rendered_images == ["b-image", "c-image", "b-image"]
3073    assert "ARTWORK=b-stale.jpg\n" not in commands
3074    assert "ARTWORK=c.jpg\n" not in commands
3075    # the final B render completed within the budget, so it rides the bundle
3076    assert "ARTWORKFILE=b-final.jpg\n" in commands[-1]
3077    assert commands[-1].endswith("ACTION=SENDMETA\n")
3078    assert stream._metadata_artwork_checksum == "b-image"
3079
3080
3081@pytest.mark.asyncio
3082async def test_send_metadata_passes_cached_artwork_path_to_binary() -> None:
3083    """The staged artwork carries the absolute cache path returned by preparation."""
3084    player = _make_player()
3085    stream = AirPlayStream(player)
3086    metadata = MagicMock(
3087        duration=180,
3088        title="Track",
3089        artist="Artist",
3090        album="Album",
3091        image_url="https://example.com/artwork.png",
3092    )
3093    cached_path = "/cache/thumbnails/artwork_flat.jpg"
3094    send_command = AsyncMock()
3095
3096    with (
3097        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value=cached_path)),
3098        patch.object(stream, "send_cli_command", new=send_command),
3099    ):
3100        await stream.send_metadata(None, metadata)
3101
3102    assert f"ARTWORKFILE={cached_path}\n" in send_command.await_args_list[-1].args[0]
3103
3104
3105@pytest.mark.asyncio
3106async def test_track_change_bundles_ready_artwork_into_a_single_push() -> None:
3107    """A track change whose artwork renders within budget lands as ONE bundled write."""
3108    stream = AirPlayStream(_make_player())
3109    stream._cli_proc = _make_cli_proc()
3110    metadata = MagicMock(
3111        queue_item_id="item-1",
3112        duration=180,
3113        title="Track",
3114        artist="Artist",
3115        album="Album",
3116        image_url="image",
3117    )
3118
3119    with (
3120        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3121        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
3122    ):
3123        await stream.send_metadata(0, metadata)
3124
3125    assert write_command.await_count == 1
3126    lines = write_command.await_args_list[0].args[0].decode().splitlines()
3127    assert "TITLE=Track" in lines
3128    assert "ITEMID=item-1" in lines
3129    # the artwork is staged before the SENDMETA applies the whole bundle
3130    assert lines[-2:] == ["ARTWORKFILE=/cache/art.jpg", "ACTION=SENDMETA"]
3131    # the push resets the device position to zero: no PROGRESS correction
3132    assert stream._last_progress_sent == 0
3133    assert stream._metadata_artwork_checksum == "image"
3134
3135
3136@pytest.mark.asyncio
3137async def test_track_change_artwork_missing_the_budget_follows_as_artwork_command() -> None:
3138    """A render missing the bundling budget still delivers via ARTWORK once it completes."""
3139    stream = AirPlayStream(_make_player())
3140    stream._cli_proc = _make_cli_proc()
3141    metadata = MagicMock(
3142        queue_item_id="item-1",
3143        duration=180,
3144        title="Track",
3145        artist="Artist",
3146        album="Album",
3147        image_url="image",
3148    )
3149    release_render = asyncio.Event()
3150
3151    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3152        await release_render.wait()
3153        return "/cache/late.jpg"
3154
3155    with (
3156        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3157        patch.object(
3158            stream, "_prepare_artwork", new_callable=AsyncMock, side_effect=_prepare_artwork
3159        ),
3160        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.01),
3161    ):
3162        push = asyncio.create_task(stream.send_metadata(0, metadata))
3163        async with asyncio.timeout(2):
3164            while write_command.await_count == 0:
3165                await asyncio.sleep(0)
3166        # the identity bundle went out without artwork once the budget lapsed
3167        assert stream._metadata_artwork_checksum == ""
3168        release_render.set()
3169        await push
3170
3171    commands = [args.args[0].decode() for args in write_command.await_args_list]
3172    assert "ARTWORKFILE" not in commands[0]
3173    assert commands[0].endswith("ACTION=SENDMETA\n")
3174    assert [command for command in commands if command.startswith("ARTWORK=")] == [
3175        "ARTWORK=/cache/late.jpg\n"
3176    ]
3177    assert stream._metadata_artwork_checksum == "image"
3178
3179
3180@pytest.mark.asyncio
3181async def test_pending_start_interrupts_the_artwork_wait() -> None:
3182    """A pending START releases the bounded artwork wait instead of queueing behind it."""
3183    player = _make_player()
3184    stream = AirPlayStream(player)
3185    stream._cli_proc = _make_cli_proc()
3186    stream._connected.set()
3187    metadata = MagicMock(
3188        queue_item_id="item-1",
3189        duration=180,
3190        title="Track",
3191        artist="Artist",
3192        album="Album",
3193        image_url="image",
3194    )
3195    release_render = asyncio.Event()
3196    render_started = asyncio.Event()
3197
3198    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3199        render_started.set()
3200        await release_render.wait()
3201        return "/cache/late.jpg"
3202
3203    with (
3204        patch.object(
3205            stream,
3206            "_write_cli_command",
3207            new_callable=AsyncMock,
3208            side_effect=_acking_write_cli_command(stream),
3209        ) as write_command,
3210        patch.object(
3211            stream, "_prepare_artwork", new_callable=AsyncMock, side_effect=_prepare_artwork
3212        ),
3213    ):
3214        push = asyncio.create_task(stream.send_metadata(0, metadata))
3215        await render_started.wait()
3216        # the metadata push sits in its render budget holding the lock; the
3217        # START must release that wait instead of losing its anchor lead to it
3218        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
3219        release_render.set()
3220        await push
3221
3222    commands = [args.args[0] for args in write_command.await_args_list]
3223    assert commands[0].endswith("ACTION=SENDMETA\n")
3224    assert "ARTWORKFILE" not in commands[0]
3225    assert commands[1].startswith(f"START_UNIX_MS={START_UNIX_MS}")
3226
3227
3228@pytest.mark.asyncio
3229async def test_track_change_starting_mid_track_sends_a_progress_correction() -> None:
3230    """A track change landing mid-position corrects the timeline after the bundle."""
3231    stream = AirPlayStream(_make_player())
3232    stream._cli_proc = _make_cli_proc()
3233    metadata = MagicMock(
3234        queue_item_id="item-1",
3235        duration=180,
3236        title="Track",
3237        artist="Artist",
3238        album="Album",
3239        image_url="image",
3240    )
3241
3242    with (
3243        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3244        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
3245    ):
3246        await stream.send_metadata(120, metadata)
3247
3248    commands = [args.args[0].decode() for args in write_command.await_args_list]
3249    assert len(commands) == 2
3250    assert commands[0].endswith("ACTION=SENDMETA\n")
3251    # the push reset the device position to zero, so the mid-track start is
3252    # corrected right after
3253    assert commands[1].endswith("PROGRESS=120\n")
3254    assert stream._last_progress_sent == 120
3255
3256
3257@pytest.mark.asyncio
3258async def test_failed_artwork_delivery_is_retried() -> None:
3259    """A dropped ARTWORK command remains pending for the next metadata update."""
3260    stream = AirPlayStream(_make_player())
3261    metadata = MagicMock(
3262        queue_item_id="item-1",
3263        duration=180,
3264        title="Track",
3265        artist="Artist",
3266        album="Album",
3267        image_url="image",
3268    )
3269    artwork_path = "/cache/thumbnails/artwork.jpg"
3270    release_render = asyncio.Event()
3271
3272    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3273        # the first render misses the bundling budget, so the artwork goes
3274        # out through the stand-alone ARTWORK command
3275        if not release_render.is_set():
3276            await release_render.wait()
3277        return artwork_path
3278
3279    with (
3280        patch.object(
3281            stream,
3282            "_prepare_artwork",
3283            new_callable=AsyncMock,
3284            side_effect=_prepare_artwork,
3285        ) as prepare_artwork,
3286        patch.object(
3287            stream,
3288            "send_cli_command",
3289            new_callable=AsyncMock,
3290            side_effect=[True, False, True],
3291        ) as send_command,
3292        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.01),
3293    ):
3294        first_push = asyncio.create_task(stream.send_metadata(None, metadata))
3295        async with asyncio.timeout(2):
3296            while send_command.await_count == 0:
3297                await asyncio.sleep(0)
3298        release_render.set()
3299        await first_push
3300        assert stream._metadata_artwork_checksum == ""
3301        await stream.send_metadata(None, metadata)
3302
3303    assert prepare_artwork.await_count == 2
3304    assert [args.args[0] for args in send_command.await_args_list].count(
3305        f"ARTWORK={artwork_path}"
3306    ) == 2
3307    assert stream._metadata_artwork_checksum == "image"
3308
3309
3310@pytest.mark.asyncio
3311async def test_text_refinement_keeps_delivered_artwork_settled() -> None:
3312    """A text-only metadata update after delivery does not re-render unchanged art."""
3313    stream = AirPlayStream(_make_player())
3314    metadata = MagicMock(
3315        duration=180,
3316        title="Track",
3317        artist="Artist",
3318        album="Album",
3319        image_url="image",
3320    )
3321    refined = MagicMock(
3322        queue_item_id=metadata.queue_item_id,
3323        duration=180,
3324        title="Track (Remastered)",
3325        artist="Artist",
3326        album="Album",
3327        image_url="image",
3328    )
3329    artwork_path = "/cache/thumbnails/artwork.jpg"
3330
3331    with (
3332        patch.object(
3333            stream,
3334            "_prepare_artwork",
3335            new_callable=AsyncMock,
3336            return_value=artwork_path,
3337        ) as prepare_artwork,
3338        patch.object(
3339            stream,
3340            "send_cli_command",
3341            new_callable=AsyncMock,
3342            return_value=True,
3343        ) as send_command,
3344    ):
3345        await stream.send_metadata(None, metadata)
3346        assert stream._metadata_artwork_checksum == "image"
3347        # the refinement bumps the metadata generation (pending identity
3348        # changed), which before the identity settle re-armed the artwork
3349        await stream.send_metadata(None, refined)
3350
3351    prepare_artwork.assert_awaited_once()
3352    commands = [args.args[0] for args in send_command.await_args_list]
3353    assert sum(f"ARTWORKFILE={artwork_path}\n" in command for command in commands) == 1
3354    assert "ARTWORKFILE" not in commands[-1]
3355    assert "TITLE=Track (Remastered)" in commands[-1]
3356
3357
3358# --- Structured connect failures reported by the binary ---
3359
3360
3361@pytest.mark.asyncio
3362async def test_connect_error_status_line_is_parsed() -> None:
3363    """The machine-readable failure line is captured with all of its fields."""
3364    stream = AirPlayStream(_make_player())
3365
3366    stream._handle_status_line(
3367        '[STATUS] error code=auth_required http=401 detail="RTSP setup rejected"'
3368    )
3369
3370    assert stream._connect_error == CliError("auth_required", 401, "RTSP setup rejected")
3371
3372
3373@pytest.mark.asyncio
3374async def test_started_ack_status_line_parsing() -> None:
3375    """A started ack releases the START wait; a malformed one carries no details."""
3376    stream = AirPlayStream(_make_player())
3377
3378    stream._handle_status_line(
3379        "[STATUS] started requested_unix_ms=1750000000000 at_unix_ms=1750000000004"
3380    )
3381    assert stream._started.is_set()
3382    assert stream._start_ack == (1750000000000, 1750000000004)
3383
3384    stream._started.clear()
3385    stream._start_ack = None
3386    stream._handle_status_line("[STATUS] started requested_unix_ms=garbage at_unix_ms=1")
3387    assert stream._started.is_set()
3388    assert stream._start_ack is None
3389
3390
3391# --- Post-commit anchor verification ---
3392
3393
3394def test_anchor_corrected_status_line_rebases_the_position() -> None:
3395    """A correction carrying a content cut moves the reported-position base by it."""
3396    stream = AirPlayStream(_make_player())
3397    stream._start_position = 12.0
3398
3399    ended = stream._handle_status_line(
3400        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3401        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3402    )
3403
3404    assert ended is False
3405    assert stream._start_position == 12.5
3406
3407
3408def test_anchor_corrected_status_line_logs_a_warning(caplog: pytest.LogCaptureFixture) -> None:
3409    """The correction is logged loudly, including the display name and the delta."""
3410    stream = AirPlayStream(_make_player())
3411
3412    with caplog.at_level(logging.WARNING):
3413        stream._handle_status_line(
3414            "[STATUS] anchor_corrected requested_unix_ms=0 "
3415            "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3416        )
3417
3418    assert "Player A" in caplog.text
3419    assert "+500 ms" in caplog.text
3420
3421
3422def test_anchor_corrected_status_line_tolerates_malformed_line() -> None:
3423    """A malformed anchor_corrected line is dropped instead of raising or rebasing."""
3424    stream = AirPlayStream(_make_player())
3425    stream._start_position = 12.0
3426
3427    ended = stream._handle_status_line("[STATUS] anchor_corrected requested_unix_ms=garbage")
3428
3429    assert ended is False
3430    assert stream._start_position == 12.0
3431
3432
3433def test_content_cut_short_rebases_the_position_and_warns(
3434    caplog: pytest.LogCaptureFixture,
3435) -> None:
3436    """A cut that ended early gives back the ms the correction over-advanced the base by."""
3437    stream = AirPlayStream(_make_player())
3438    stream._start_position = 12.0
3439    stream._handle_status_line(
3440        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3441        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3442    )
3443    assert stream._start_position == 12.5
3444
3445    with caplog.at_level(logging.WARNING):
3446        ended = stream._handle_status_line(
3447            "[STATUS] content_cut requested_ms=500 cut_ms=180 cut_bytes=31752 drain_ms=210"
3448        )
3449
3450    assert ended is False
3451    assert stream._start_position == pytest.approx(12.18)
3452    assert "AirPlay content cut" in caplog.text
3453    assert "Player A" in caplog.text
3454    assert "320 ms short" in caplog.text
3455
3456
3457def test_content_cut_in_full_leaves_the_position_alone(caplog: pytest.LogCaptureFixture) -> None:
3458    """A cut that took what it asked for needs no correction and stays quiet."""
3459    stream = AirPlayStream(_make_player())
3460    stream._start_position = 12.0
3461    stream._handle_status_line(
3462        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3463        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3464    )
3465
3466    caplog.clear()
3467    with caplog.at_level(logging.WARNING):
3468        # a few ms below the request is byte quantization, not a short cut
3469        stream._handle_status_line(
3470            "[STATUS] content_cut requested_ms=500 cut_ms=498 cut_bytes=87887 drain_ms=505"
3471        )
3472
3473    assert stream._start_position == 12.5
3474    assert caplog.text == ""
3475
3476
3477def test_content_cut_after_a_new_anchor_is_not_reconciled() -> None:
3478    """A cut settling after a START must not be taken off that START's absolute base."""
3479    stream = AirPlayStream(_make_player())
3480    stream._start_position = 12.0
3481    stream._handle_status_line(
3482        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3483        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3484    )
3485    stream.rebase_position(30_000)
3486
3487    stream._handle_status_line(
3488        "[STATUS] content_cut requested_ms=500 cut_ms=0 cut_bytes=0 drain_ms=12"
3489    )
3490
3491    assert stream._start_position == 30.0
3492
3493
3494def test_content_cut_status_line_tolerates_malformed_line() -> None:
3495    """A malformed content_cut line is dropped instead of raising or rebasing."""
3496    stream = AirPlayStream(_make_player())
3497    stream._start_position = 12.0
3498    stream._handle_status_line(
3499        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3500        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3501    )
3502
3503    ended = stream._handle_status_line("[STATUS] content_cut requested_ms=500 cut_ms=garbage")
3504
3505    assert ended is False
3506    assert stream._start_position == 12.5
3507
3508
3509def test_clock_verified_status_line_is_debug_logged(caplog: pytest.LogCaptureFixture) -> None:
3510    """A clock_verified line needs no server action beyond a debug note of the margin."""
3511    stream = AirPlayStream(_make_player())
3512
3513    with caplog.at_level(logging.DEBUG):
3514        ended = stream._handle_status_line("[STATUS] clock_verified margin_ms=42")
3515
3516    assert ended is False
3517    assert "42" in caplog.text
3518
3519
3520@pytest.mark.asyncio
3521async def test_connect_error_status_line_tolerates_missing_fields() -> None:
3522    """A failure line without http/detail still yields the reported code."""
3523    stream = AirPlayStream(_make_player())
3524
3525    stream._handle_status_line("[STATUS] error code=connect_failed")
3526
3527    assert stream._connect_error == CliError("connect_failed", 0, "")
3528
3529
3530@pytest.mark.asyncio
3531async def test_auth_required_surfaces_password_required_error() -> None:
3532    """A device asking for a password produces an actionable, translated error."""
3533    stream = AirPlayStream(_make_player())
3534    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
3535    stream._process_ended.set()
3536
3537    with (
3538        patch.object(stream, "_cli_proc", MagicMock()),
3539        pytest.raises(PlayerCommandFailed) as err,
3540    ):
3541        await stream.wait_for_connection()
3542
3543    assert err.value.translation_key == "password_required"
3544
3545
3546@pytest.mark.asyncio
3547async def test_auth_failed_surfaces_authentication_failed_error() -> None:
3548    """A rejected password is reported as an authentication failure, not a timeout."""
3549    stream = AirPlayStream(_make_player())
3550    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
3551    stream._process_ended.set()
3552
3553    with (
3554        patch.object(stream, "_cli_proc", MagicMock()),
3555        pytest.raises(PlayerCommandFailed) as err,
3556    ):
3557        await stream.wait_for_connection()
3558
3559    assert err.value.translation_key == "authentication_failed"
3560
3561
3562@pytest.mark.asyncio
3563@pytest.mark.parametrize("code", ["auth_required", "auth_failed"])
3564async def test_refused_connection_is_not_reported_as_a_password_problem(code: str) -> None:
3565    """A device that turns the handshake away points at pairing, not at a password."""
3566    stream = AirPlayStream(_make_player())
3567    stream._handle_status_line(f'[STATUS] error code={code} http=403 detail="refused"')
3568    stream._process_ended.set()
3569
3570    with (
3571        patch.object(stream, "_cli_proc", MagicMock()),
3572        pytest.raises(PlayerCommandFailed) as err,
3573    ):
3574        await stream.wait_for_connection()
3575
3576    assert err.value.translation_key == "connection_refused"
3577
3578
3579@pytest.mark.asyncio
3580@pytest.mark.parametrize("code", ["auth_required", "auth_failed"])
3581async def test_refused_connection_never_marks_the_password_invalid(code: str) -> None:
3582    """
3583    A refusal must not leave a player demanding a password it may not even have.
3584
3585    tvOS 26 answers the pairing handshake with 403 for reasons unrelated to any
3586    secret, and the marker persists across restarts - so latching it there would
3587    strand the player in a setup flow no password can complete.
3588    """
3589    player = _make_player()
3590    stream = AirPlayStream(player)
3591
3592    stream._handle_status_line(f'[STATUS] error code={code} http=403 detail="refused"')
3593
3594    player.set_password_invalid.assert_not_called()
3595
3596
3597@pytest.mark.asyncio
3598async def test_generic_connect_failure_keeps_the_timeout_semantics() -> None:
3599    """A non-auth failure keeps raising the plain timeout its callers already handle."""
3600    stream = AirPlayStream(_make_player())
3601    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
3602    stream._process_ended.set()
3603
3604    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
3605        await stream.wait_for_connection()
3606
3607
3608@pytest.mark.asyncio
3609async def test_dead_process_fails_the_connect_wait_immediately() -> None:
3610    """
3611    A binary that reports no reason at all leaves the wait its plain timeout error.
3612
3613    It must still end the moment the process is gone instead of running out the
3614    full connect timeout.
3615    """
3616    stream = AirPlayStream(_make_player())
3617    stream._process_ended.set()  # process died without emitting a [STATUS] error line
3618
3619    started = asyncio.get_running_loop().time()
3620    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
3621        await stream.wait_for_connection()
3622
3623    assert asyncio.get_running_loop().time() - started < 1
3624
3625
3626@pytest.mark.asyncio
3627async def test_auth_failed_marks_the_stored_password_invalid() -> None:
3628    """A rejected password is persisted so the player keeps offering its setup action."""
3629    player = _make_player()
3630    stream = AirPlayStream(player)
3631
3632    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
3633
3634    player.set_password_invalid.assert_called_once_with(True)
3635
3636
3637@pytest.mark.asyncio
3638async def test_auth_required_also_marks_the_password_as_needed() -> None:
3639    """
3640    A device that demanded a password we could not supply flips into setup.
3641
3642    Devices can enforce a password without announcing it (stale TXT records), so
3643    the runtime signal must set the marker too - it is the only reliable one.
3644    """
3645    player = _make_player()
3646    stream = AirPlayStream(player)
3647
3648    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
3649
3650    player.set_password_invalid.assert_called_once_with(True)
3651
3652
3653@pytest.mark.asyncio
3654async def test_plain_connect_failures_leave_the_password_marker_alone() -> None:
3655    """A non-authentication failure says nothing about the stored password."""
3656    player = _make_player()
3657    stream = AirPlayStream(player)
3658
3659    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
3660
3661    player.set_password_invalid.assert_not_called()
3662
3663
3664@pytest.mark.asyncio
3665async def test_successful_connect_clears_the_password_marker() -> None:
3666    """Whatever the device accepted is a working password."""
3667    player = _make_player()
3668    stream = AirPlayStream(player)
3669
3670    stream._handle_status_line("[STATUS] connected")
3671
3672    assert stream.connected is True
3673    player.set_password_invalid.assert_called_once_with(False)
3674
3675
3676# --- Password preflight ---
3677
3678
3679@pytest.mark.asyncio
3680async def test_connect_refuses_password_device_without_password_or_credentials() -> None:
3681    """A password-protected AirPlay 2 device with nothing to authenticate never spawns a process."""
3682    player = _make_player()
3683    player.password_required = True
3684    player.get_setup_value = MagicMock(return_value=None)
3685    stream = AirPlayStream(player)
3686
3687    with pytest.raises(PlayerCommandFailed) as err:
3688        await stream.connect()
3689
3690    assert err.value.translation_key == "password_required"
3691    assert stream._cli_proc is None
3692
3693
3694@pytest.mark.asyncio
3695async def test_password_preflight_passes_with_password_or_credentials() -> None:
3696    """Either a configured password or stored credentials let the connect proceed."""
3697    player = _make_player()
3698    player.password_required = True
3699    player.get_setup_value = MagicMock(return_value=None)
3700    player.config.get_value = MagicMock(
3701        side_effect=lambda key, default=None: "s3cret" if key == CONF_PASSWORD else default
3702    )
3703    AirPlayStream(player)._check_password_preflight()
3704
3705    # credentials alone are enough: the binary's pair-verify leg may still succeed
3706    player.config.get_value = MagicMock(side_effect=lambda _key, default=None: default)
3707    player.get_setup_value = MagicMock(
3708        side_effect=lambda key, default=None: (
3709            "ab" * 96 if key == CONF_AIRPLAY_CREDENTIALS else default
3710        )
3711    )
3712    AirPlayStream(player)._check_password_preflight()
3713
3714
3715@pytest.mark.asyncio
3716async def test_password_preflight_skipped_for_raop() -> None:
3717    """The preflight only guards the native AirPlay 2 flow; RAOP carries its own password."""
3718    player = _make_player()
3719    player.protocol = StreamingProtocol.RAOP
3720    player.password_required = True
3721    player.get_setup_value = MagicMock(return_value=None)
3722
3723    AirPlayStream(player)._check_password_preflight()
3724