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, Callable, Coroutine
10from contextlib import 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    AirPlayRemoteCommand,
30    ClockReadiness,
31    StreamingProtocol,
32)
33from music_assistant.providers.airplay.stream import AirPlayStream, CliError
34
35START_UNIX_MS = 1_750_000_000_000
36AP2_FEATURES = "0x4A7FDFD5,0x3C177FDE"
37
38
39def _make_player() -> MagicMock:
40    """Build a mock AirPlay player with both discovery records present."""
41    player = MagicMock()
42    player.player_id = "apaabbccddeeff"
43    player.display_name = "Player A"
44    player.address = "192.168.1.50"
45    player.protocol = StreamingProtocol.AIRPLAY2
46    player.protocol_override = None
47    player.volume_level = 40
48    player.device_info.mac_address = "AA:BB:CC:DD:EE:FF"
49    player.device_info.ip_address = "192.168.1.50"
50    player.device_info.manufacturer = "Acme, Inc."
51    player.device_info.model = "Test1,1"
52    player.logger = logging.getLogger("test.airplay.player")
53    player.config.get_value = MagicMock(side_effect=lambda _key, default=None: default)
54    player.state.active_group = None
55
56    airplay_info = MagicMock()
57    airplay_info.port = 7000
58    airplay_info.server = "playera.local."
59    airplay_info.decoded_properties = {
60        "features": "0x5A7FFFF7,0x1E",
61        "flags": "0x4",
62        "model": "Test1,1",
63        "manufacturer": "Acme, Inc.",  # contains a space: must be skipped in --txt
64    }
65    player.airplay_discovery_info = airplay_info
66
67    raop_info = MagicMock()
68    raop_info.port = 5000
69    raop_info.name = "AABBCCDDEEFF@Player A._raop._tcp.local."
70    raop_info.decoded_properties = {"et": "0,4", "md": "0,1,2", "cn": "0,1"}
71    player.raop_discovery_info = raop_info
72
73    prov = MagicMock()
74    prov.dacp_id = "ABCDEF0123456789"
75    prov.ptp_daemon_ready = True
76    prov.logger = logging.getLogger("test.airplay.prov")
77    # auto-detected publish ip: reachable address of this host, but not the interface
78    # the stream to this device leaves from - so it is never handed to the binary
79    prov.mass.streams.publish_ip = "192.168.1.99"
80    prov.mass.streams.get_source_ip = AsyncMock(return_value="192.168.1.5")
81    prov.mass.streams.get_publish_ip = MagicMock(return_value=None)
82    player.provider = prov
83    return player
84
85
86async def _build_args(player: MagicMock) -> list[str]:
87    """Build the CLI args for the given player with the externals patched out."""
88    stream = AirPlayStream(player)
89    with patch(
90        "music_assistant.providers.airplay.stream.get_cli_binary",
91        return_value="/fake/cliairplay",
92    ):
93        return await stream._build_cli_args()
94
95
96def _arg_value(args: list[str], flag: str) -> Any:
97    """Return the value following the given flag in the argument list."""
98    return args[args.index(flag) + 1]
99
100
101def _acking_write_cli_command(
102    stream: AirPlayStream, at_unix_ms: int | None = None
103) -> Callable[[str], Coroutine[Any, Any, bool]]:
104    """
105    Build a ``_write_cli_command`` replacement that acks a START like the binary.
106
107    :param stream: The stream whose START commands are answered.
108    :param at_unix_ms: Scheduled audible instant to report back; defaults to the
109        commanded instant, i.e. the binary found it feasible.
110    """
111
112    async def write_command(command: str) -> bool:
113        # A START nothing acks otherwise pays the real ack timeout in wall clock;
114        # feeding the ack here leaves start()'s wait already released.
115        if "ACTION=START" in command:
116            requested = int(command.split("START_UNIX_MS=")[1].split("\n", 1)[0])
117            stream._handle_status_line(
118                f"[STATUS] started requested_unix_ms={requested} "
119                f"at_unix_ms={requested if at_unix_ms is None else at_unix_ms}"
120            )
121        return True
122
123    return write_command
124
125
126@pytest.mark.asyncio
127async def test_cli_args_default_auto() -> None:
128    """Default (no protocol override) passes --protocol auto with the full mDNS TXT."""
129    player = _make_player()
130    args = await _build_args(player)
131
132    assert _arg_value(args, "--protocol") == "auto"
133    assert "--start-unix-ms" not in args
134    # legacy timing args are gone
135    assert "--ntpstart" not in args
136    assert "--wait" not in args
137    # AirPlay 2 service is the connection target when it may be used
138    assert _arg_value(args, "--port") == "7000"
139    assert _arg_value(args, "--name") == "Player A"
140    assert _arg_value(args, "--hostname") == "playera.local."
141    # RAOP mDNS props still passed for the RAOP-based flows
142    assert _arg_value(args, "--udn") == "AABBCCDDEEFF@Player A._raop._tcp.local."
143    assert _arg_value(args, "--et") == "0,4"
144    assert _arg_value(args, "--cn") == "0,1"
145    # full TXT for route selection; pairs containing whitespace are skipped
146    txt = _arg_value(args, "--txt")
147    assert "features=0x5A7FFFF7,0x1E" in txt
148    assert "flags=0x4" in txt
149    assert "manufacturer" not in txt
150    # default format
151    assert _arg_value(args, "--samplerate") == "44100"
152    assert _arg_value(args, "--bitdepth") == "16"
153    # no explicit latency override configured
154    assert "--latency" not in args
155    # PTP daemon is running: stream attaches to the shared clock
156    assert "--ptp-shared" in args
157    # networking: the interface the timing packets leave from is pinned
158    assert _arg_value(args, "--if") == "192.168.1.5"
159    assert "--publish-ip" not in args
160    # the target is the only positional argument; PREPARE selects stdin
161    assert args[-1] == "192.168.1.50"
162    assert "-" not in args
163    assert "--cmdpipe" in args
164
165
166@pytest.mark.asyncio
167async def test_cli_args_auto_publish_ip_is_never_advertised() -> None:
168    """
169    An auto-detected publish IP must not reach --publish-ip.
170
171    The binary treats it as authoritative for the PTP timing-peer list, while the
172    timing packets leave from the resolved --if interface. A peer list naming any
173    other address makes the receiver discard our clock and play silence.
174    """
175    player = _make_player()
176    player.provider.mass.streams.publish_ip = "10.45.0.20"
177
178    args = await _build_args(player)
179
180    assert _arg_value(args, "--if") == "192.168.1.5"
181    assert "--publish-ip" not in args
182    assert "10.45.0.20" not in args
183
184
185@pytest.mark.asyncio
186async def test_cli_args_configured_publish_ip_is_advertised() -> None:
187    """An explicitly configured publish IP is a reachability statement and is passed on."""
188    player = _make_player()
189    player.provider.mass.streams.get_publish_ip = MagicMock(return_value="10.45.0.20")
190
191    args = await _build_args(player)
192
193    assert _arg_value(args, "--publish-ip") == "10.45.0.20"
194    player.provider.mass.streams.get_publish_ip.assert_called_once_with("192.168.1.50")
195
196
197@pytest.mark.asyncio
198async def test_cli_args_publish_ip_omitted_when_it_matches_the_interface() -> None:
199    """A publish IP identical to the bound interface adds nothing to the peer list."""
200    player = _make_player()
201    player.provider.mass.streams.get_publish_ip = MagicMock(return_value="192.168.1.5")
202
203    args = await _build_args(player)
204
205    assert _arg_value(args, "--if") == "192.168.1.5"
206    assert "--publish-ip" not in args
207
208
209@pytest.mark.asyncio
210async def test_cli_args_no_interface_pin_leaves_routing_to_the_binary() -> None:
211    """With no interface to pin, --if is dropped so the routing table decides."""
212    player = _make_player()
213    player.provider.mass.streams.get_source_ip = AsyncMock(return_value=None)
214
215    args = await _build_args(player)
216
217    assert "--if" not in args
218
219
220@pytest.mark.asyncio
221async def test_cli_args_log_the_pinned_interface(caplog: pytest.LogCaptureFixture) -> None:
222    """The resolved interface is stated outright, so a user's log shows what it bound to."""
223    player = _make_player()
224
225    with caplog.at_level(logging.DEBUG):
226        await _build_args(player)
227
228    assert (
229        "cliairplay network binding for player apaabbccddeeff: "
230        "if=192.168.1.5 publish_ip=<not configured>" in caplog.text
231    )
232
233
234@pytest.mark.asyncio
235async def test_cli_args_log_the_unpinned_interface_and_publish_ip(
236    caplog: pytest.LogCaptureFixture,
237) -> None:
238    """An unpinned interface and a configured publish IP are named for what they are."""
239    player = _make_player()
240    player.provider.mass.streams.get_source_ip = AsyncMock(return_value=None)
241    player.provider.mass.streams.get_publish_ip = MagicMock(return_value="10.45.0.20")
242
243    with caplog.at_level(logging.DEBUG):
244        await _build_args(player)
245
246    assert (
247        "cliairplay network binding for player apaabbccddeeff: "
248        "if=<all interfaces> publish_ip=10.45.0.20" in caplog.text
249    )
250
251
252@pytest.mark.asyncio
253async def test_cli_args_raop_override() -> None:
254    """A forced RAOP protocol targets the RAOP service and skips AP2-only args."""
255    player = _make_player()
256    player.protocol_override = StreamingProtocol.RAOP
257    args = await _build_args(player)
258
259    assert _arg_value(args, "--protocol") == "raop"
260    assert _arg_value(args, "--port") == "5000"
261    assert "--name" not in args
262    assert "--hostname" not in args
263    assert "--ptp-shared" not in args
264    assert "--encrypt" in args
265
266
267@pytest.mark.asyncio
268async def test_cli_args_raop_encryption_can_be_disabled() -> None:
269    """The legacy encryption preference remains available for incompatible receivers."""
270    player = _make_player()
271    player.protocol_override = StreamingProtocol.RAOP
272    player.config.get_value = MagicMock(
273        side_effect=lambda key, default=None: False if key == CONF_ENCRYPTION else default
274    )
275
276    args = await _build_args(player)
277
278    assert "--encrypt" not in args
279
280
281@pytest.mark.asyncio
282async def test_cli_args_no_ptp_shared_when_daemon_alive_but_not_ready() -> None:
283    """
284    A daemon that is merely alive must not get --ptp-shared.
285
286    Liveness does not mean the daemon is serving: until it publishes its clock
287    there is nothing to attach to, and a stream that asks anyway silently takes
288    its own timing instead - drifting away from the members that got the clock.
289    """
290    player = _make_player()
291    player.provider.ptp_daemon_running = True
292    player.provider.ptp_daemon_ready = False
293    args = await _build_args(player)
294    assert "--ptp-shared" not in args
295
296
297@pytest.mark.asyncio
298async def test_cli_args_linkplay_gets_deeper_buffer() -> None:
299    """
300    LinkPlay-family devices get the deep receiver queue from the family table.
301
302    Their pipeline starves at the stock depth - fully once the device is also
303    master of a native multiroom group - so the table maps them to 1750 ms.
304    Both platform generations must match: the newer names Linkplay as
305    manufacturer, the older only marks the platform in fv under OEM brands.
306    """
307    player = _make_player()
308    player.device_info.manufacturer = "Linkplay Technology Inc."
309    args = await _build_args(player)
310    assert _arg_value(args, "--latency") == "1750"
311    assert "--ptp-shared" in args
312
313    # Old platform: OEM brand, the Linkplay token only in fv.
314    player = _make_player()
315    player.device_info.manufacturer = "Edifier Inc"
316    player.airplay_discovery_info.decoded_properties["fv"] = "p20.Linkplay.4.6.430230"
317    args = await _build_args(player)
318    assert _arg_value(args, "--latency") == "1750"
319
320    # Non-LinkPlay devices stay on the binary's stock depth.
321    player = _make_player()
322    args = await _build_args(player)
323    assert "--latency" not in args
324
325
326@pytest.mark.asyncio
327async def test_cli_args_buffer_depth_config_overrides_auto() -> None:
328    """A configured buffer depth wins over the device-family default."""
329    player = _make_player()
330    player.device_info.manufacturer = "Linkplay Technology Inc."
331    player.config.get_value = MagicMock(
332        side_effect=lambda key, default=None: 1500 if key == CONF_BUFFER_DEPTH else default
333    )
334    args = await _build_args(player)
335    assert _arg_value(args, "--latency") == "1500"
336
337
338@pytest.mark.asyncio
339async def test_cli_args_no_latency_override() -> None:
340    """The playback lead/buffer is binary-managed; MA never passes --latency."""
341    player = _make_player()
342    args = await _build_args(player)
343    assert "--latency" not in args
344
345
346@pytest.mark.asyncio
347async def test_cli_args_hires_pcm_format() -> None:
348    """A 24-bit stream passes --bitdepth 24 while the pipe carries s32le samples."""
349    player = _make_player()
350    hires_format = AudioFormat(content_type=ContentType.PCM_S32LE, sample_rate=48000, bit_depth=24)
351    stream = AirPlayStream(player, pcm_format=hires_format)
352    with patch(
353        "music_assistant.providers.airplay.stream.get_cli_binary",
354        return_value="/fake/cliairplay",
355    ):
356        args = await stream._build_cli_args()
357
358    assert _arg_value(args, "--samplerate") == "48000"
359    assert _arg_value(args, "--bitdepth") == "24"
360    # the ffmpeg pipe format must be the 32-bit container (binary truncates to 24)
361    assert stream.pcm_format.content_type == ContentType.PCM_S32LE
362
363
364@pytest.mark.asyncio
365async def test_cli_args_raop_only_device() -> None:
366    """True legacy RAOP uses the command pipe and has no positional audio source."""
367    player = _make_player()
368    player.airplay_discovery_info = None
369    player.protocol = StreamingProtocol.RAOP
370    args = await _build_args(player)
371
372    assert _arg_value(args, "--protocol") == "auto"
373    assert _arg_value(args, "--port") == "5000"
374    assert "--txt" not in args
375    assert "--name" not in args
376    assert "--encrypt" in args
377    assert "--cmdpipe" in args
378    assert args[-1] == "192.168.1.50"
379    assert "-" not in args
380
381
382@pytest.mark.asyncio
383async def test_cli_args_auto_raop_uses_raop_service_port() -> None:
384    """Auto-selected legacy RAOP targets the RAOP service rather than the AP2 service."""
385    player = _make_player()
386    player.protocol = StreamingProtocol.RAOP
387    player.airplay_discovery_info.decoded_properties["features"] = "0x0"
388
389    args = await _build_args(player)
390
391    assert _arg_value(args, "--protocol") == "auto"
392    assert _arg_value(args, "--port") == "5000"
393    assert "--name" not in args
394
395
396@pytest.mark.asyncio
397async def test_cli_args_pass_raop_feature_fallback_to_auto_router() -> None:
398    """AP2 bits advertised only on _raop.ft still reach the binary route resolver."""
399    player = _make_player()
400    player.airplay_discovery_info.decoded_properties.pop("features")
401    player.raop_discovery_info.decoded_properties["ft"] = AP2_FEATURES
402
403    args = await _build_args(player)
404
405    assert _arg_value(args, "--protocol") == "auto"
406    assert _arg_value(args, "--port") == "7000"
407    assert f"ft={AP2_FEATURES}" in _arg_value(args, "--txt")
408
409
410@pytest.mark.asyncio
411async def test_cli_args_featureless_ap2_only_device_forces_airplay2() -> None:
412    """An AP2-only receiver without feature bits cannot fall through to legacy RAOP."""
413    player = _make_player()
414    player.raop_discovery_info = None
415    player.airplay_discovery_info.decoded_properties = {}
416
417    args = await _build_args(player)
418
419    assert _arg_value(args, "--protocol") == "airplay2"
420    assert _arg_value(args, "--port") == "7000"
421
422
423@pytest.mark.parametrize(
424    ("line", "expected_route"),
425    [
426        (
427            "[STATUS] route protocol=airplay2 flow=realtime timing=ptp buffered=0",
428            "AirPlay 2 (realtime, PTP)",
429        ),
430        # a buffered stream is named by its delivery, whatever flow it was requested as
431        (
432            "[STATUS] route protocol=airplay2 flow=realtime timing=ntp buffered=1",
433            "AirPlay 2 (buffered, NTP)",
434        ),
435        (
436            "[STATUS] route protocol=raop flow=realtime timing=ntp buffered=0",
437            "RAOP",
438        ),
439    ],
440    ids=["airplay2-realtime", "airplay2-buffered", "raop"],
441)
442def test_parse_route_status(
443    line: str, expected_route: str, caplog: pytest.LogCaptureFixture
444) -> None:
445    """The [STATUS] route line resolves the route this stream took and reports it."""
446    player = _make_player()
447    stream = AirPlayStream(player)
448
449    with caplog.at_level(logging.INFO):
450        stream._parse_route_status(line)
451
452    assert stream.active_route == expected_route
453    assert f"Streaming to Player A via {expected_route}" in caplog.text
454
455
456@pytest.mark.asyncio
457async def test_stdout_reader_dispatches_the_route_line() -> None:
458    """The CLI stdout reader hands the [STATUS] route line to the route parser."""
459    player = _make_player()
460    stream = AirPlayStream(player)
461    process = MagicMock()
462    process.read = AsyncMock(
463        side_effect=[
464            b"[STATUS] route protocol=airplay2 flow=realtime timing=ptp buffered=0\n",
465            b"",
466        ]
467    )
468    stream._cli_proc = process
469
470    await stream._stdout_reader()
471
472    assert stream.active_route == "AirPlay 2 (realtime, PTP)"
473
474
475def test_parse_latency_status() -> None:
476    """The [STATUS] latency line is parsed into the stream's latency attributes."""
477    player = _make_player()
478    stream = AirPlayStream(player)
479    stream._parse_latency_status(
480        "[STATUS] latency lead_ms=1750 device_min_frames=11025 device_max_frames=88200 "
481        "warm_lead_ms=1200"
482    )
483    assert stream.latency_lead_ms == 1750
484    assert stream.device_min_frames == 11025
485    assert stream.device_max_frames == 88200
486    assert stream.warm_lead_ms == 1200
487
488
489def test_parse_latency_status_reads_every_field_on_its_own() -> None:
490    """One unusable value must not leave the fields after it stale from the last report."""
491    player = _make_player()
492    stream = AirPlayStream(player)
493    stream._parse_latency_status(
494        "[STATUS] latency lead_ms=1750 device_min_frames=11025 device_max_frames=88200 "
495        "warm_lead_ms=1200"
496    )
497
498    stream._parse_latency_status(
499        "[STATUS] latency lead_ms=900 device_min_frames=garbage device_max_frames=44100 "
500        "warm_lead_ms=300"
501    )
502
503    assert stream.latency_lead_ms == 900
504    assert stream.device_min_frames == 0  # unusable, so unreported
505    assert stream.device_max_frames == 44100
506    assert stream.warm_lead_ms == 300
507
508
509def test_parse_latency_status_logs_the_warm_lead(caplog: pytest.LogCaptureFixture) -> None:
510    """The warm lead drives every warm group anchor, so it belongs in the line."""
511    stream = AirPlayStream(_make_player())
512
513    with caplog.at_level(logging.DEBUG):
514        stream._parse_latency_status(
515            "[STATUS] latency lead_ms=1750 device_min_frames=11025 device_max_frames=88200 "
516            "warm_lead_ms=1200"
517        )
518
519    assert "warm lead=1200ms" in caplog.text
520
521
522def test_mrp_push_accepted_is_not_logged_at_info(caplog: pytest.LogCaptureFixture) -> None:
523    """A push the device accepted is bookkeeping, not something to act on."""
524    stream = AirPlayStream(_make_player())
525
526    with caplog.at_level(logging.INFO):
527        stream._parse_mrp_status("[STATUS] mrp path=command status=200")
528
529    assert caplog.text == ""
530
531
532@pytest.mark.parametrize("status", [302, 403, 500], ids=["redirect", "forbidden", "server-error"])
533def test_mrp_push_rejection_is_reported(status: int, caplog: pytest.LogCaptureFixture) -> None:
534    """Anything but a 2xx is the device not taking the push, so it is reported."""
535    stream = AirPlayStream(_make_player())
536
537    with caplog.at_level(logging.WARNING):
538        stream._parse_mrp_status(f"[STATUS] mrp path=command status={status}")
539
540    assert "Player A" in caplog.text
541    assert str(status) in caplog.text
542
543
544def test_mrp_artwork_rejection_is_reported(caplog: pytest.LogCaptureFixture) -> None:
545    """An artwork rejection carries no path= or status=, and must not read as a plain push."""
546    stream = AirPlayStream(_make_player())
547
548    with caplog.at_level(logging.WARNING):
549        stream._parse_mrp_status(
550            "[STATUS] mrp artwork=rejected reason=progressive_jpeg bytes=48123 "
551            "width=512 height=512 precision=8 sof=0xc2 components=3 progressive=1 "
552            "clear_status=200 staging_max_bytes=131072"
553        )
554
555    assert "rejected the now-playing artwork" in caplog.text
556    assert "progressive_jpeg" in caplog.text
557    assert "HTTP ?" not in caplog.text
558
559
560def test_mrp_artwork_rejection_reaches_the_parser_from_the_stderr_reader(
561    caplog: pytest.LogCaptureFixture,
562) -> None:
563    """The binary reports artwork on stderr, so the status dispatcher must route it."""
564    stream = AirPlayStream(_make_player())
565
566    with caplog.at_level(logging.WARNING):
567        ends_the_loop = stream._handle_status_line(
568            "[STATUS] mrp artwork=rejected reason=progressive_jpeg bytes=48123 "
569            "width=512 height=512 precision=8 sof=0xc2 components=3 progressive=1 "
570            "clear_status=200 staging_max_bytes=131072"
571        )
572
573    assert ends_the_loop is False
574    assert "rejected the now-playing artwork" in caplog.text
575
576
577def test_mrp_channel_status_is_not_read_as_an_http_status(
578    caplog: pytest.LogCaptureFixture,
579) -> None:
580    """The data-channel line reports 0/1, which must not be warned about as a failed push."""
581    stream = AirPlayStream(_make_player())
582
583    with caplog.at_level(logging.WARNING):
584        stream._parse_mrp_status("[STATUS] mrp path=channel status=0")
585
586    assert caplog.text == ""
587
588
589@pytest.mark.parametrize(
590    ("line", "expected"),
591    [
592        # both tables published: the union is kept
593        (
594            "[STATUS] capabilities requested=0x80000 realtime_formats=0x40000 "
595            "realtime_known=1 buffered_formats=0x80000 buffered_known=1",
596            0xC0000,
597        ),
598        # the Apple TV publishes 24-bit for its buffered stream only
599        (
600            "[STATUS] capabilities requested=0x80000 realtime_formats=0x1440800 "
601            "realtime_known=1 buffered_formats=0xe80000 buffered_known=1",
602            0x1EC0800,
603        ),
604        # a table the device did not publish is ignored, even when non-zero
605        (
606            "[STATUS] capabilities requested=0x40000 realtime_formats=0x40000 "
607            "realtime_known=1 buffered_formats=0x80000 buffered_known=0",
608            0x40000,
609        ),
610        # RAOP-compat routes report nothing: the probed value survives
611        (
612            "[STATUS] capabilities requested=0x40000 realtime_formats=0x0 "
613            "realtime_known=0 buffered_formats=0x0 buffered_known=0",
614            0x1234,
615        ),
616        # a malformed mask is ignored rather than raising
617        (
618            "[STATUS] capabilities realtime_formats=nonsense realtime_known=1",
619            0x1234,
620        ),
621    ],
622)
623def test_parse_capabilities_status(line: str, expected: int) -> None:
624    """The [STATUS] capabilities line refreshes the formats the player advertises."""
625    player = _make_player()
626    player.advertised_audio_formats = 0x1234
627    stream = AirPlayStream(player)
628
629    stream._parse_capabilities_status(line)
630
631    assert player.advertised_audio_formats == expected
632
633
634@pytest.mark.parametrize(
635    ("value", "command"),
636    [
637        ("play", AirPlayRemoteCommand.PLAY),
638        ("pause", AirPlayRemoteCommand.PAUSE),
639        ("play_pause", AirPlayRemoteCommand.PLAY_PAUSE),
640        ("next", AirPlayRemoteCommand.NEXT),
641        ("previous", AirPlayRemoteCommand.PREVIOUS),
642    ],
643)
644def test_parse_remote_event(value: str, command: AirPlayRemoteCommand) -> None:
645    """A normalized CLI remote event is dispatched against its own player."""
646    player = _make_player()
647    stream = AirPlayStream(player)
648
649    stream._parse_remote_event(f"[EVENT] remote command={value}")
650
651    player.provider.handle_remote_command.assert_called_once_with(player, command)
652
653
654def test_parse_remote_event_rejects_unknown_command(caplog: pytest.LogCaptureFixture) -> None:
655    """An unknown CLI remote event is reported and ignored."""
656    player = _make_player()
657    stream = AirPlayStream(player)
658
659    with caplog.at_level(logging.WARNING):
660        stream._parse_remote_event("[EVENT] remote command=unsupported")
661
662    player.provider.handle_remote_command.assert_not_called()
663    assert "Ignoring unknown cliairplay remote command: unsupported" in caplog.text
664
665
666@pytest.mark.asyncio
667async def test_stdout_reader_dispatches_remote_events_once() -> None:
668    """The CLI stdout reader dispatches each normalized remote event exactly once."""
669    player = _make_player()
670    stream = AirPlayStream(player)
671    process = MagicMock()
672    output = "".join(f"[EVENT] remote command={command}\n" for command in AirPlayRemoteCommand)
673    process.read = AsyncMock(side_effect=[output.encode(), b""])
674    stream._cli_proc = process
675
676    await stream._stdout_reader()
677
678    assert player.provider.handle_remote_command.call_args_list == [
679        call(player, command) for command in AirPlayRemoteCommand
680    ]
681
682
683def test_command_pipe_paths_are_unique_per_stream() -> None:
684    """A stopped stream cannot remove a replacement stream's command pipe."""
685    player = _make_player()
686    first_stream = AirPlayStream(player)
687    second_stream = AirPlayStream(player)
688    assert first_stream.commands_pipe.path != second_stream.commands_pipe.path
689
690
691@pytest.mark.asyncio
692async def test_command_pipe_write_returns_false_without_reader(
693    tmp_path: Path, caplog: pytest.LogCaptureFixture
694) -> None:
695    """A command pipe write reports a missing reader without retrying or warning."""
696    pipe_path = tmp_path / "commands"
697    os.mkfifo(pipe_path)
698    writer = AsyncNamedPipeWriter(str(pipe_path))
699
700    with (
701        caplog.at_level(logging.WARNING),
702        patch("music_assistant.helpers.named_pipe.os.open", wraps=os.open) as open_pipe,
703    ):
704        assert await writer.write(b"ACTION=STANDBY\n") is False
705
706    # the missing reader is reported once, without retrying or crying wolf
707    open_pipe.assert_called_once()
708    assert not [record for record in caplog.records if record.levelno >= logging.WARNING]
709
710
711@pytest.mark.asyncio
712async def test_command_pipe_write_returns_true_for_complete_write() -> None:
713    """A complete command pipe write reports successful delivery."""
714    data = b"ACTION=STANDBY\n"
715    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
716    writer._write_fd = 42
717
718    with patch("music_assistant.helpers.named_pipe.os.write", return_value=len(data)) as write:
719        assert await writer.write(data) is True
720
721    write.assert_called_once_with(42, data)
722
723
724@pytest.mark.asyncio
725async def test_command_pipe_write_completes_short_write() -> None:
726    """A short command pipe write continues with the remaining data."""
727    data = b"ACTION=STANDBY\n"
728    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
729    writer._write_fd = 42
730
731    with patch(
732        "music_assistant.helpers.named_pipe.os.write",
733        side_effect=[5, len(data) - 5],
734    ) as write:
735        assert await writer.write(data) is True
736
737    assert write.call_args_list == [call(42, memoryview(data)), call(42, memoryview(data)[5:])]
738
739
740@pytest.mark.asyncio
741async def test_command_pipe_write_returns_false_and_resets_fd_on_epipe() -> None:
742    """A closed command pipe reader resets the writer for a later retry."""
743    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
744    writer._write_fd = 42
745
746    with (
747        patch(
748            "music_assistant.helpers.named_pipe.os.write",
749            side_effect=OSError(errno.EPIPE, "reader closed"),
750        ),
751        patch("music_assistant.helpers.named_pipe.os.close") as close_fd,
752    ):
753        assert await writer.write(b"ACTION=STANDBY\n") is False
754
755    close_fd.assert_called_once_with(42)
756    assert writer._write_fd is None
757
758
759@pytest.mark.asyncio
760async def test_command_pipe_write_resets_fd_when_epipe_follows_partial_write() -> None:
761    """An EPIPE after a short write reports failure and resets the writer."""
762    data = b"ACTION=STANDBY\n"
763    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
764    writer._write_fd = 42
765
766    with (
767        patch(
768            "music_assistant.helpers.named_pipe.os.write",
769            side_effect=[5, OSError(errno.EPIPE, "reader closed")],
770        ) as write,
771        patch("music_assistant.helpers.named_pipe.os.close") as close_fd,
772    ):
773        assert await writer.write(data) is False
774
775    assert write.call_args_list == [call(42, memoryview(data)), call(42, memoryview(data)[5:])]
776    close_fd.assert_called_once_with(42)
777    assert writer._write_fd is None
778
779
780@pytest.mark.asyncio
781async def test_command_pipe_writes_are_serialized() -> None:
782    """Concurrent command writes cannot interleave after a short write."""
783    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
784    writer._write_fd = 42
785    written_chunks: list[bytes] = []
786    first_chunk_written = threading.Event()
787    release_first_write = threading.Event()
788
789    def write_chunk(_fd: int, data: memoryview) -> int:
790        chunk = bytes(data)
791        if not written_chunks:
792            written_chunks.append(chunk[:2])
793            first_chunk_written.set()
794            assert release_first_write.wait(timeout=1)
795            return 2
796        written_chunks.append(chunk)
797        return len(chunk)
798
799    with patch("music_assistant.helpers.named_pipe.os.write", side_effect=write_chunk):
800        first_write = asyncio.create_task(writer.write(b"FIRST\n"))
801        assert await asyncio.to_thread(first_chunk_written.wait, 1)
802        second_write = asyncio.create_task(writer.write(b"SECOND\n"))
803        await asyncio.sleep(0)
804        release_first_write.set()
805        assert all(await asyncio.gather(first_write, second_write))
806
807    assert b"".join(written_chunks) == b"FIRST\nSECOND\n"
808
809
810@pytest.mark.asyncio
811async def test_command_pipe_write_propagates_non_epipe_errors() -> None:
812    """An unexpected command pipe write error remains visible to the caller."""
813    writer = AsyncNamedPipeWriter("/tmp/commands")  # noqa: S108
814    writer._write_fd = 42
815
816    with (
817        patch(
818            "music_assistant.helpers.named_pipe.os.write",
819            side_effect=OSError(errno.EBADF, "bad file descriptor"),
820        ),
821        pytest.raises(OSError, match="bad file descriptor"),
822    ):
823        await writer.write(b"ACTION=STANDBY\n")
824
825
826@pytest.mark.asyncio
827async def test_pipe_write_reports_a_stalled_reader_without_raising(
828    tmp_path: Path, caplog: pytest.LogCaptureFixture
829) -> None:
830    """A write bigger than the pipe buffer reports failure instead of raising."""
831    pipe_path = tmp_path / "audio"
832    os.mkfifo(pipe_path)
833    writer = AsyncNamedPipeWriter(str(pipe_path))
834    # a reader that never reads, so the pipe buffer fills up and stays full
835    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
836
837    try:
838        with (
839            caplog.at_level(logging.WARNING),
840            patch("music_assistant.helpers.named_pipe.WRITE_STALL_TIMEOUT", 0.05),
841        ):
842            assert await writer.write(b"\x00" * 176400) is False
843
844        # the reader is still attached, so the descriptor stays usable
845        assert writer._write_fd is not None
846        assert not [record for record in caplog.records if record.levelno >= logging.WARNING]
847    finally:
848        os.close(read_fd)
849        await writer.remove()
850
851
852@pytest.mark.asyncio
853async def test_pipe_write_completes_a_large_write_while_the_reader_drains(
854    tmp_path: Path,
855) -> None:
856    """A write bigger than the pipe buffer completes against a reader that keeps up."""
857    pipe_path = tmp_path / "audio"
858    os.mkfifo(pipe_path)
859    writer = AsyncNamedPipeWriter(str(pipe_path))
860    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
861    payload = b"\x00" * 176400
862
863    async def drain() -> bytes:
864        received = bytearray()
865        while len(received) < len(payload):
866            try:
867                chunk = os.read(read_fd, 65536)
868            except BlockingIOError:
869                chunk = b""
870            if not chunk:
871                # no writer attached yet or nothing buffered: yield instead of spinning
872                await asyncio.sleep(0.001)
873                continue
874            received += chunk
875        return bytes(received)
876
877    reader = asyncio.create_task(drain())
878    try:
879        async with asyncio.timeout(10):
880            assert await writer.write(payload) is True
881            # every byte arrives exactly once, so a resumed write never repeats itself
882            assert await reader == payload
883    finally:
884        reader.cancel()
885        with suppress(asyncio.CancelledError):
886            await reader
887        os.close(read_fd)
888        await writer.remove()
889
890
891@pytest.mark.asyncio
892async def test_pipe_write_outlasts_a_reader_slower_than_the_stall_timeout(
893    tmp_path: Path,
894) -> None:
895    """The stall budget covers a lack of progress, not the total time a write takes."""
896    pipe_path = tmp_path / "audio"
897    os.mkfifo(pipe_path)
898    writer = AsyncNamedPipeWriter(str(pipe_path))
899    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
900    payload = b"\x00" * 176400
901    stall_timeout = 0.15
902
903    async def drain_slowly() -> int:
904        received = 0
905        while received < len(payload):
906            await asyncio.sleep(0.02)
907            with suppress(BlockingIOError):
908                received += len(os.read(read_fd, 8192))
909        return received
910
911    reader = asyncio.create_task(drain_slowly())
912    loop = asyncio.get_running_loop()
913    started = loop.time()
914    try:
915        async with asyncio.timeout(30):
916            with patch("music_assistant.helpers.named_pipe.WRITE_STALL_TIMEOUT", stall_timeout):
917                assert await writer.write(payload) is True
918            assert await reader == len(payload)
919        # a reader this slow only completes because each pause resets the budget
920        assert loop.time() - started > stall_timeout
921    finally:
922        reader.cancel()
923        with suppress(asyncio.CancelledError):
924            await reader
925        os.close(read_fd)
926        await writer.remove()
927
928
929@pytest.mark.asyncio
930async def test_pipe_write_resumes_after_the_buffer_drains() -> None:
931    """A write that fills the pipe buffer continues once the reader catches up."""
932    data = b"\x00" * 10
933    writer = AsyncNamedPipeWriter("/tmp/audio")  # noqa: S108
934    writer._write_fd = 42
935
936    with (
937        patch(
938            "music_assistant.helpers.named_pipe.os.write",
939            side_effect=[4, BlockingIOError(errno.EAGAIN, "buffer full"), 6],
940        ) as write,
941        patch("music_assistant.helpers.named_pipe.select.poll") as poll,
942    ):
943        assert await writer.write(data) is True
944
945    # the retry picks up where the short write stopped, it never resends
946    assert write.call_args_list == [
947        call(42, memoryview(data)),
948        call(42, memoryview(data)[4:]),
949        call(42, memoryview(data)[4:]),
950    ]
951    poll.return_value.register.assert_called_once_with(42, select.POLLOUT)
952    poll.return_value.poll.assert_called_once_with(WRITE_POLL_INTERVAL_MS)
953
954
955@pytest.mark.asyncio
956async def test_pipe_write_gives_up_when_the_pipe_goes_before_the_first_write() -> None:
957    """A pipe removed the instant a write starts fails cleanly rather than blowing up."""
958    writer = AsyncNamedPipeWriter("/tmp/audio")  # noqa: S108
959
960    with (
961        patch.object(AsyncNamedPipeWriter, "_ensure_write_fd", return_value=True),
962        patch("music_assistant.helpers.named_pipe.os.write") as write,
963    ):
964        assert await writer.write(b"\x00" * 10) is False
965
966    write.assert_not_called()
967
968
969@pytest.mark.asyncio
970async def test_pipe_write_stops_when_the_descriptor_is_taken_mid_stall() -> None:
971    """A stalled write gives up once the pipe is removed out from under it."""
972    writer = AsyncNamedPipeWriter("/tmp/audio")  # noqa: S108
973    writer._write_fd = 42
974
975    def take_descriptor(pipe_writer: AsyncNamedPipeWriter, _write_fd: int) -> None:
976        pipe_writer._write_fd = None
977
978    with (
979        patch(
980            "music_assistant.helpers.named_pipe.os.write",
981            side_effect=[4, BlockingIOError(errno.EAGAIN, "buffer full")],
982        ) as write,
983        patch.object(AsyncNamedPipeWriter, "_wait_writable", take_descriptor),
984    ):
985        assert await writer.write(b"\x00" * 10) is False
986
987    # writing on would land in whatever reopened that descriptor number
988    assert write.call_count == 2
989
990
991@pytest.mark.asyncio
992async def test_pipe_write_resets_fd_when_the_reader_closes_during_a_stall(
993    tmp_path: Path,
994) -> None:
995    """A reader that goes away while the buffer is full still resets the writer."""
996    pipe_path = tmp_path / "audio"
997    os.mkfifo(pipe_path)
998    writer = AsyncNamedPipeWriter(str(pipe_path))
999    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
1000
1001    async def close_reader_mid_stall() -> None:
1002        await asyncio.sleep(0.05)
1003        os.close(read_fd)
1004
1005    closer = asyncio.create_task(close_reader_mid_stall())
1006    try:
1007        async with asyncio.timeout(30):
1008            assert await writer.write(b"\x00" * 176400) is False
1009            await closer
1010        # the departed reader is what ends the write, so it must not be left mid-stall
1011        assert writer._write_fd is None
1012    finally:
1013        closer.cancel()
1014        with suppress(asyncio.CancelledError):
1015            await closer
1016        with suppress(OSError):
1017            os.close(read_fd)
1018        await writer.remove()
1019
1020
1021@pytest.mark.asyncio
1022async def test_command_pipe_wait_for_reader_resolves_when_the_reader_attaches(
1023    tmp_path: Path,
1024) -> None:
1025    """The reader wait ends as soon as the binary opens its end of the command pipe."""
1026    pipe_path = tmp_path / "commands"
1027    os.mkfifo(pipe_path)
1028    writer = AsyncNamedPipeWriter(str(pipe_path))
1029    read_fds: list[int] = []
1030
1031    async def attach_reader() -> None:
1032        await asyncio.sleep(0.01)
1033        read_fds.append(os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK))
1034
1035    reader = asyncio.create_task(attach_reader())
1036    try:
1037        assert await writer.wait_for_reader(timeout=1) is True
1038        assert read_fds  # resolved on the attach, not before it
1039    finally:
1040        await reader
1041        for read_fd in read_fds:
1042            os.close(read_fd)
1043        await writer.remove()
1044
1045
1046@pytest.mark.asyncio
1047async def test_command_pipe_wait_for_reader_gives_up_at_its_timeout(tmp_path: Path) -> None:
1048    """A command pipe nothing ever reads fails the wait within the requested window."""
1049    pipe_path = tmp_path / "commands"
1050    os.mkfifo(pipe_path)
1051    writer = AsyncNamedPipeWriter(str(pipe_path))
1052    loop = asyncio.get_running_loop()
1053
1054    started = loop.time()
1055    assert await writer.wait_for_reader(timeout=0.1) is False
1056
1057    assert 0.1 <= loop.time() - started < 1
1058
1059
1060@pytest.mark.asyncio
1061async def test_command_pipe_wait_for_reader_uses_no_worker_thread(tmp_path: Path) -> None:
1062    """Waiting for the binary's reader never strands a blocking worker thread."""
1063    pipe_path = tmp_path / "commands"
1064    os.mkfifo(pipe_path)
1065    writer = AsyncNamedPipeWriter(str(pipe_path))
1066
1067    with (
1068        patch(
1069            "music_assistant.helpers.named_pipe.os.open",
1070            side_effect=OSError(errno.ENXIO, "no reader"),
1071        ),
1072        patch(
1073            "music_assistant.helpers.named_pipe.asyncio.to_thread",
1074            new_callable=AsyncMock,
1075        ) as to_thread,
1076    ):
1077        assert await writer.wait_for_reader(timeout=0.1) is False
1078
1079    to_thread.assert_not_awaited()
1080
1081
1082@pytest.mark.asyncio
1083async def test_command_pipe_wait_for_reader_defers_to_an_in_flight_write(tmp_path: Path) -> None:
1084    """The reader wait leaves the descriptor to a write that is already under way."""
1085    pipe_path = tmp_path / "commands"
1086    os.mkfifo(pipe_path)
1087    writer = AsyncNamedPipeWriter(str(pipe_path))
1088    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
1089
1090    try:
1091        await writer._write_lock.acquire()
1092        wait = asyncio.create_task(writer.wait_for_reader(timeout=1))
1093        await asyncio.sleep(0)
1094
1095        # a reader is attached the whole time, so only the write lock holds this up
1096        assert not wait.done()
1097
1098        writer._write_lock.release()
1099        assert await wait is True
1100    finally:
1101        os.close(read_fd)
1102        await writer.remove()
1103
1104
1105@pytest.mark.asyncio
1106async def test_command_pipe_write_reuses_the_fd_from_the_reader_wait(tmp_path: Path) -> None:
1107    """A command written after the reader wait rides the descriptor that wait opened."""
1108    pipe_path = tmp_path / "commands"
1109    os.mkfifo(pipe_path)
1110    writer = AsyncNamedPipeWriter(str(pipe_path))
1111    read_fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
1112
1113    try:
1114        assert await writer.wait_for_reader(timeout=1) is True
1115
1116        with patch("music_assistant.helpers.named_pipe.os.open") as open_pipe:
1117            assert await writer.write(b"ACTION=STANDBY\n") is True
1118
1119        open_pipe.assert_not_called()
1120        assert os.read(read_fd, 64) == b"ACTION=STANDBY\n"
1121    finally:
1122        os.close(read_fd)
1123        await writer.remove()
1124
1125
1126@pytest.mark.asyncio
1127async def test_cli_command_updates_timestamp_after_successful_delivery() -> None:
1128    """A delivered command updates the player's last command timestamp."""
1129    player = _make_player()
1130    player.last_command_sent = 10.0
1131    stream = AirPlayStream(player)
1132    stream._cli_proc = MagicMock(closed=False)
1133
1134    with (
1135        patch.object(stream.commands_pipe, "write", new=AsyncMock(return_value=True)),
1136        patch("music_assistant.providers.airplay.stream.time.time", return_value=20.0),
1137    ):
1138        assert await stream.send_cli_command("ACTION=STANDBY") is True
1139
1140    assert player.last_command_sent == 20.0
1141
1142
1143@pytest.mark.asyncio
1144async def test_cli_command_preserves_timestamp_when_delivery_fails() -> None:
1145    """A dropped command leaves the player's last command timestamp unchanged."""
1146    player = _make_player()
1147    player.last_command_sent = 10.0
1148    stream = AirPlayStream(player)
1149    stream._cli_proc = MagicMock(closed=False)
1150
1151    with patch.object(stream.commands_pipe, "write", new=AsyncMock(return_value=False)):
1152        assert await stream.send_cli_command("ACTION=STANDBY") is False
1153
1154    assert player.last_command_sent == 10.0
1155
1156
1157@pytest.mark.asyncio
1158async def test_cli_command_preserves_timestamp_when_delivery_raises() -> None:
1159    """A command write error leaves the player's last command timestamp unchanged."""
1160    player = _make_player()
1161    player.last_command_sent = 10.0
1162    stream = AirPlayStream(player)
1163    stream._cli_proc = MagicMock(closed=False)
1164
1165    with (
1166        patch.object(
1167            stream.commands_pipe,
1168            "write",
1169            new=AsyncMock(side_effect=OSError("command pipe failed")),
1170        ),
1171        pytest.raises(OSError, match="command pipe failed"),
1172    ):
1173        await stream.send_cli_command("ACTION=STANDBY")
1174
1175    assert player.last_command_sent == 10.0
1176
1177
1178@pytest.mark.asyncio
1179async def test_connect_does_not_write_before_the_binary_can_read() -> None:
1180    """Connecting lays out the command pipe and starts cliairplay, pushing nothing into it."""
1181    player = _make_player()
1182    stream = AirPlayStream(player)
1183    # media is available, so a push that never happens is by design and not a missing source
1184    player.current_media = MagicMock(corrected_elapsed_time=12.5)
1185    process = MagicMock(closed=False)
1186    process.start = AsyncMock(return_value=None)
1187    operation_order: list[str] = []
1188
1189    async def create_pipe() -> None:
1190        operation_order.append("pipe")
1191
1192    async def start_process() -> None:
1193        operation_order.append("process")
1194
1195    def consume_task(awaitable: Any) -> MagicMock:
1196        awaitable.close()
1197        task = MagicMock()
1198        task.done.return_value = True
1199        return task
1200
1201    process.start.side_effect = start_process
1202    player.provider.mass.create_task.side_effect = consume_task
1203    with (
1204        patch.object(stream, "_build_cli_args", new_callable=AsyncMock, return_value=["binary"]),
1205        patch(
1206            "music_assistant.providers.airplay.stream.AsyncProcess",
1207            return_value=process,
1208        ),
1209        patch.object(stream.commands_pipe, "create", side_effect=create_pipe),
1210        patch.object(stream, "send_metadata", new_callable=AsyncMock) as send_metadata_mock,
1211    ):
1212        await stream.connect()
1213
1214    assert operation_order == ["pipe", "process"]
1215    send_metadata_mock.assert_not_awaited()
1216
1217
1218@pytest.mark.asyncio
1219async def test_connect_failure_cleans_up_process_and_pipe() -> None:
1220    """A cliairplay process that fails to start cannot leave a live process or FIFO."""
1221    player = _make_player()
1222    stream = AirPlayStream(player)
1223    process = MagicMock(closed=False)
1224    process.start = AsyncMock(side_effect=OSError("process start failed"))
1225    process.kill = AsyncMock()
1226
1227    with (
1228        patch.object(stream, "_build_cli_args", new_callable=AsyncMock, return_value=["binary"]),
1229        patch(
1230            "music_assistant.providers.airplay.stream.AsyncProcess",
1231            return_value=process,
1232        ),
1233        patch.object(stream.commands_pipe, "create", new_callable=AsyncMock),
1234        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
1235        pytest.raises(OSError, match="process start failed"),
1236    ):
1237        await stream.connect()
1238
1239    process.kill.assert_awaited_once()
1240    remove_pipe.assert_awaited_once()
1241    assert stream._cli_proc is None
1242    assert stream._cleanup_complete is True
1243
1244
1245@pytest.mark.asyncio
1246async def test_start_sends_command_and_stamps_position() -> None:
1247    """START is delivered over the command pipe and stamps the media position."""
1248    player = _make_player()
1249    stream = AirPlayStream(player)
1250    stream._cli_proc = MagicMock(closed=False)
1251    stream._connected.set()
1252
1253    with patch.object(
1254        stream,
1255        "_write_cli_command",
1256        new_callable=AsyncMock,
1257        side_effect=_acking_write_cli_command(stream),
1258    ) as write_command:
1259        assert await stream.start(START_UNIX_MS, 12_000) == START_UNIX_MS
1260
1261    write_command.assert_awaited_once_with(f"START_UNIX_MS={START_UNIX_MS}\nACTION=START")
1262    assert stream._start_position == 12.0
1263    player.set_state_from_stream.assert_called_once_with(elapsed_time=12.0, stream=stream)
1264
1265
1266@pytest.mark.asyncio
1267async def test_start_join_marks_the_command() -> None:
1268    """A late-join START carries START_JOIN=1 so the binary enforces clock readiness."""
1269    player = _make_player()
1270    stream = AirPlayStream(player)
1271    stream._cli_proc = MagicMock(closed=False)
1272    stream._connected.set()
1273
1274    with patch.object(
1275        stream,
1276        "_write_cli_command",
1277        new_callable=AsyncMock,
1278        side_effect=_acking_write_cli_command(stream),
1279    ) as write_command:
1280        assert await stream.start(START_UNIX_MS, 0, join=True) == START_UNIX_MS
1281
1282    write_command.assert_awaited_once_with(
1283        f"START_UNIX_MS={START_UNIX_MS}\nSTART_JOIN=1\nACTION=START"
1284    )
1285
1286
1287@pytest.mark.asyncio
1288@pytest.mark.parametrize(
1289    "complete_before_start",
1290    [True, False],
1291    ids=["delivered", "rendering"],
1292)
1293async def test_start_transition_artwork_settled_or_retried(complete_before_start: bool) -> None:
1294    """START keeps already-delivered transition artwork settled and retries a superseded render."""
1295    player = _make_player()
1296    stream = AirPlayStream(player)
1297    stream._cli_proc = MagicMock(closed=False)
1298    stream._connected.set()
1299    metadata = MagicMock(
1300        corrected_elapsed_time=0,
1301        queue_item_id="item-new",
1302        title="New track",
1303        artist="Artist",
1304        album="Album",
1305        duration=180,
1306        image_url="new-image",
1307    )
1308    stream.session = MagicMock(media=metadata)
1309    render_started = asyncio.Event()
1310    release_render = asyncio.Event()
1311    metadata_tasks: list[asyncio.Task[Any]] = []
1312    render_count = 0
1313
1314    def create_task(target: Any, **_kwargs: Any) -> asyncio.Task[Any]:
1315        task = asyncio.create_task(target())
1316        metadata_tasks.append(task)
1317        return task
1318
1319    async def prepare_artwork(_image_url: str, _generation: int) -> str:
1320        nonlocal render_count
1321        render_count += 1
1322        if render_count == 1:
1323            render_started.set()
1324            if not complete_before_start:
1325                await release_render.wait()
1326            return "/cache/pretransition.jpg"
1327        return "/cache/posttransition.jpg"
1328
1329    player.provider.mass.create_task.side_effect = create_task
1330    with (
1331        patch.object(
1332            stream,
1333            "_write_cli_command",
1334            new_callable=AsyncMock,
1335            side_effect=_acking_write_cli_command(stream),
1336        ) as write_command,
1337        patch.object(
1338            stream,
1339            "_prepare_artwork",
1340            new_callable=AsyncMock,
1341            side_effect=prepare_artwork,
1342        ),
1343    ):
1344        pretransition_task = asyncio.create_task(stream.send_metadata(None, metadata))
1345        await render_started.wait()
1346        if complete_before_start:
1347            await pretransition_task
1348        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
1349        await asyncio.gather(*metadata_tasks)
1350        release_render.set()
1351        await pretransition_task
1352
1353    commands = [args.args[0] for args in write_command.await_args_list]
1354    start_command = f"START_UNIX_MS={START_UNIX_MS}\nACTION=START"
1355    if complete_before_start:
1356        # artwork delivered before the anchor stays settled; re-pushing it
1357        # around the START would make an Apple TV re-render its screen
1358        assert commands[-1] == start_command
1359        assert "ARTWORK=/cache/pretransition.jpg" in commands
1360    else:
1361        # the anchor superseded the in-flight render; the post-anchor push
1362        # renders again and delivers the artwork once
1363        assert commands[-2:] == [start_command, "ARTWORK=/cache/posttransition.jpg"]
1364        assert "ARTWORK=/cache/pretransition.jpg" not in commands
1365    assert stream._metadata_generation == 2
1366    assert stream._metadata_artwork_checksum == "new-image"
1367
1368
1369@pytest.mark.asyncio
1370async def test_start_requires_connected_process() -> None:
1371    """START is rejected without a connected cliairplay process."""
1372    stream = AirPlayStream(_make_player())
1373    stream._cli_proc = MagicMock(closed=False)
1374    # not connected: _connected event never set
1375
1376    with (
1377        patch.object(stream, "_write_cli_command", new_callable=AsyncMock) as write_command,
1378        pytest.raises(RuntimeError, match="without a connected cliairplay process"),
1379    ):
1380        await stream.start(START_UNIX_MS, 0)
1381
1382    write_command.assert_not_awaited()
1383
1384
1385@pytest.mark.asyncio
1386async def test_flush_sends_command_and_awaits_ack() -> None:
1387    """FLUSH is delivered and resolves once the binary reports it flushed."""
1388    stream = AirPlayStream(_make_player())
1389    stream._cli_proc = MagicMock(closed=False)
1390    stream._connected.set()
1391
1392    with patch.object(
1393        stream, "_write_cli_command", new_callable=AsyncMock, return_value=True
1394    ) as write_command:
1395        flush_task = asyncio.create_task(stream.flush())
1396        await asyncio.sleep(0)
1397        assert stream._handle_status_line("[STATUS] flushed") is False
1398        assert await flush_task is True
1399
1400    write_command.assert_awaited_once_with("ACTION=FLUSH")
1401
1402
1403@pytest.mark.asyncio
1404async def test_flush_times_out_without_ack() -> None:
1405    """FLUSH returns False when the binary never acknowledges it."""
1406    stream = AirPlayStream(_make_player())
1407    stream._cli_proc = MagicMock(closed=False)
1408    stream._connected.set()
1409
1410    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1411        assert await stream.flush(timeout=0) is False
1412
1413
1414@pytest.mark.asyncio
1415async def test_flush_returns_false_when_command_not_delivered() -> None:
1416    """FLUSH reports failure when the command cannot be delivered."""
1417    stream = AirPlayStream(_make_player())
1418    stream._cli_proc = MagicMock(closed=False)
1419    stream._connected.set()
1420
1421    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=False):
1422        assert await stream.flush() is False
1423
1424
1425@pytest.mark.asyncio
1426async def test_start_raises_when_command_not_delivered() -> None:
1427    """A dropped START surfaces as an error so the caller can fall back cold."""
1428    stream = AirPlayStream(_make_player())
1429    stream._cli_proc = MagicMock(closed=False)
1430    stream._connected.set()
1431
1432    with (
1433        patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=False),
1434        pytest.raises(PlayerCommandFailed, match="Could not deliver START"),
1435    ):
1436        await stream.start(1_750_000_000_000, 0)
1437
1438
1439@pytest.mark.asyncio
1440async def test_start_fails_fast_on_reported_start_failure() -> None:
1441    """A reported start failure ends the ack wait at once instead of timing out."""
1442    stream = AirPlayStream(_make_player())
1443    stream._cli_proc = MagicMock(closed=False)
1444    stream._connected.set()
1445
1446    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1447        start_task = asyncio.create_task(stream.start(START_UNIX_MS, 0, join=True))
1448        await asyncio.sleep(0)
1449        stream._handle_status_line(
1450            '[STATUS] error code=start_failed http=0 detail="no live session to start"'
1451        )
1452        with pytest.raises(PlayerCommandFailed, match="no live session to start"):
1453            await start_task
1454
1455    # a command failure must not poison how a NEW connection is reported
1456    assert stream._connect_error is None
1457
1458
1459@pytest.mark.asyncio
1460async def test_flush_fails_fast_on_reported_flush_failure() -> None:
1461    """A reported flush failure resolves the ack wait as a failure, not a timeout."""
1462    stream = AirPlayStream(_make_player())
1463    stream._cli_proc = MagicMock(closed=False)
1464    stream._connected.set()
1465
1466    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1467        flush_task = asyncio.create_task(stream.flush())
1468        await asyncio.sleep(0)
1469        stream._handle_status_line(
1470            '[STATUS] error code=flush_failed http=0 detail="session rejected the flush"'
1471        )
1472        assert await flush_task is False
1473
1474    assert stream._connect_error is None
1475
1476
1477@pytest.mark.asyncio
1478async def test_start_failure_does_not_outlive_its_command() -> None:
1479    """A failed START leaves no error behind that would fail the next one."""
1480    stream = AirPlayStream(_make_player())
1481    stream._cli_proc = MagicMock(closed=False)
1482    stream._connected.set()
1483    stream._handle_status_line("[STATUS] error code=start_failed")
1484
1485    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1486        start_task = asyncio.create_task(stream.start(START_UNIX_MS, 0))
1487        await asyncio.sleep(0)
1488        stream._handle_status_line(
1489            f"[STATUS] started requested_unix_ms={START_UNIX_MS} at_unix_ms={START_UNIX_MS}"
1490        )
1491        assert await start_task == START_UNIX_MS
1492
1493
1494@pytest.mark.asyncio
1495async def test_start_returns_the_instant_the_binary_scheduled() -> None:
1496    """A corrected ack, not the commanded instant, is what the caller maps content onto."""
1497    stream = AirPlayStream(_make_player())
1498    stream._cli_proc = MagicMock(closed=False)
1499    stream._connected.set()
1500    corrected = START_UNIX_MS + 700
1501
1502    with patch.object(
1503        stream,
1504        "_write_cli_command",
1505        new_callable=AsyncMock,
1506        side_effect=_acking_write_cli_command(stream, corrected),
1507    ):
1508        assert await stream.start(START_UNIX_MS, 0) == corrected
1509
1510
1511@pytest.mark.asyncio
1512async def test_start_fails_when_the_ack_never_arrives() -> None:
1513    """An unacknowledged START fails: nothing may be mapped onto an unconfirmed instant."""
1514    stream = AirPlayStream(_make_player())
1515    stream._cli_proc = MagicMock(closed=False)
1516    stream._connected.set()
1517
1518    with (
1519        patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True),
1520        patch("music_assistant.providers.airplay.stream.AIRPLAY_START_ACK_TIMEOUT_MS", 10),
1521        pytest.raises(PlayerCommandFailed, match="did not acknowledge its start") as err,
1522    ):
1523        await stream.start(START_UNIX_MS, 0)
1524
1525    # the player and the instant nothing confirmed are the whole diagnostic
1526    assert "Player A" in str(err.value)
1527    assert str(START_UNIX_MS) in str(err.value)
1528
1529
1530@pytest.mark.asyncio
1531async def test_start_accepts_a_malformed_ack_as_the_commanded_instant() -> None:
1532    """An ack that cannot be parsed still answered the START, so the commanded instant stands."""
1533    stream = AirPlayStream(_make_player())
1534    stream._cli_proc = MagicMock(closed=False)
1535    stream._connected.set()
1536
1537    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1538        start_task = asyncio.create_task(stream.start(START_UNIX_MS, 0))
1539        await asyncio.sleep(0)
1540        stream._handle_status_line("[STATUS] started requested_unix_ms=nonsense at_unix_ms=")
1541        assert await start_task == START_UNIX_MS
1542
1543
1544@pytest.mark.asyncio
1545async def test_start_treats_a_missing_scheduled_instant_as_malformed() -> None:
1546    """An ack without at_unix_ms parses cleanly to 0, which must never be returned."""
1547    stream = AirPlayStream(_make_player())
1548    stream._cli_proc = MagicMock(closed=False)
1549    stream._connected.set()
1550
1551    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True):
1552        start_task = asyncio.create_task(stream.start(START_UNIX_MS, 0))
1553        await asyncio.sleep(0)
1554        stream._handle_status_line(f"[STATUS] started requested_unix_ms={START_UNIX_MS}")
1555        assert await start_task == START_UNIX_MS
1556
1557
1558@pytest.mark.asyncio
1559@pytest.mark.parametrize(
1560    ("join", "expected_timeout"),
1561    [
1562        (True, AIRPLAY_JOIN_START_ACK_TIMEOUT_MS / 1000),
1563        (False, AIRPLAY_START_ACK_TIMEOUT_MS / 1000),
1564    ],
1565    ids=["join", "plain"],
1566)
1567async def test_start_ack_window_matches_the_arm(join: bool, expected_timeout: float) -> None:
1568    """A join's ack is held for clock verification, so it waits far longer than a plain start."""
1569    stream = AirPlayStream(_make_player())
1570    stream._cli_proc = MagicMock(closed=False)
1571    stream._connected.set()
1572    timeouts: list[float] = []
1573
1574    async def record_timeout(awaitable: Any, timeout: float) -> None:
1575        timeouts.append(timeout)
1576        awaitable.close()
1577        raise TimeoutError
1578
1579    with (
1580        patch.object(stream, "_write_cli_command", new_callable=AsyncMock, return_value=True),
1581        patch(
1582            "music_assistant.providers.airplay.stream.asyncio.wait_for",
1583            side_effect=record_timeout,
1584        ),
1585        pytest.raises(PlayerCommandFailed),
1586    ):
1587        await stream.start(START_UNIX_MS, 0, join=join)
1588
1589    assert timeouts == [expected_timeout]
1590
1591
1592@pytest.mark.asyncio
1593async def test_flush_returns_false_when_not_connected() -> None:
1594    """FLUSH is a no-op returning False before the device connects."""
1595    stream = AirPlayStream(_make_player())
1596    stream._cli_proc = MagicMock(closed=False)
1597    # not connected
1598
1599    with patch.object(stream, "_write_cli_command", new_callable=AsyncMock) as write_command:
1600        assert await stream.flush() is False
1601
1602    write_command.assert_not_awaited()
1603
1604
1605def test_flushed_status_sets_flush_event() -> None:
1606    """The [STATUS] flushed line releases the flush acknowledgement event."""
1607    stream = AirPlayStream(_make_player())
1608    assert not stream._flushed.is_set()
1609
1610    assert stream._handle_status_line("[STATUS] flushed") is False
1611
1612    assert stream._flushed.is_set()
1613
1614
1615@pytest.mark.asyncio
1616async def test_clock_ready_projection_resolves_the_wait() -> None:
1617    """A probing receiver reports when its clock becomes usable, from its first probe."""
1618    stream = AirPlayStream(_make_player())
1619
1620    assert (
1621        stream._handle_status_line(
1622            "[STATUS] clock_ready mode=ptp state=probing streak_ms=0 exchanges=1 "
1623            f"ready_in_ms=2300 ready_at_unix_ms={START_UNIX_MS}"
1624        )
1625        is False
1626    )
1627
1628    assert await stream.wait_clock_ready(timeout=0.01) == (
1629        ClockReadiness.PROJECTED,
1630        START_UNIX_MS,
1631    )
1632
1633
1634@pytest.mark.asyncio
1635async def test_clock_ready_cold_line_keeps_waiting_for_a_projection() -> None:
1636    """A receiver that has not probed yet carries no projection, so the wait goes on."""
1637    stream = AirPlayStream(_make_player())
1638
1639    stream._handle_status_line(
1640        "[STATUS] clock_ready mode=ptp state=cold streak_ms=0 exchanges=0 "
1641        "ready_in_ms=0 ready_at_unix_ms=0"
1642    )
1643
1644    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.UNREPORTED, 0)
1645    assert not stream._clock_ready.is_set()
1646
1647    stream._handle_status_line(
1648        "[STATUS] clock_ready mode=ptp state=ready streak_ms=2400 exchanges=9 "
1649        f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
1650    )
1651
1652    assert await stream.wait_clock_ready(timeout=0.01) == (
1653        ClockReadiness.PROJECTED,
1654        START_UNIX_MS,
1655    )
1656
1657
1658@pytest.mark.asyncio
1659async def test_clock_ready_ntp_resolves_without_a_projection() -> None:
1660    """NTP timing has no receiver clock to wait for, so the wait ends with nothing."""
1661    stream = AirPlayStream(_make_player())
1662
1663    stream._handle_status_line(
1664        "[STATUS] clock_ready mode=ntp state=ready streak_ms=0 exchanges=0 "
1665        f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
1666    )
1667
1668    assert stream._clock_ready.is_set()
1669    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.NOT_APPLICABLE, 0)
1670
1671
1672@pytest.mark.asyncio
1673async def test_wait_clock_ready_times_out_for_a_binary_that_never_reports() -> None:
1674    """A binary that does not report readiness is told apart from one that answered."""
1675    stream = AirPlayStream(_make_player())
1676
1677    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.UNREPORTED, 0)
1678
1679
1680@pytest.mark.asyncio
1681async def test_clock_ready_stalled_state_warns_once(caplog: pytest.LogCaptureFixture) -> None:
1682    """A receiver that never answers our clock is reported loudly once and carries no projection."""
1683    stream = AirPlayStream(_make_player())
1684
1685    with caplog.at_level(logging.DEBUG):
1686        ended = stream._handle_status_line(
1687            "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
1688            "ready_in_ms=0 ready_at_unix_ms=0"
1689        )
1690        stream._handle_status_line(
1691            "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
1692            "ready_in_ms=0 ready_at_unix_ms=0"
1693        )
1694
1695    warnings = [record for record in caplog.records if record.levelno == logging.WARNING]
1696    assert ended is False
1697    assert len(warnings) == 1
1698    assert "Player A" in warnings[0].getMessage()
1699    assert "319/320" in warnings[0].getMessage()
1700    assert stream._clock_ready.is_set()
1701    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.STALLED, 0)
1702
1703
1704def test_clock_ready_stall_warning_is_ptp_only(caplog: pytest.LogCaptureFixture) -> None:
1705    """An NTP-timed session has no clock of ours to answer, so it never reads as a stall."""
1706    stream = AirPlayStream(_make_player())
1707
1708    with caplog.at_level(logging.DEBUG):
1709        stream._handle_status_line(
1710            "[STATUS] clock_ready mode=ntp state=stalled streak_ms=0 exchanges=0 "
1711            f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
1712        )
1713
1714    assert [record for record in caplog.records if record.levelno >= logging.WARNING] == []
1715    assert stream._clock_ready_at_unix_ms == 0
1716
1717
1718@pytest.mark.parametrize("state", ["cold", "probing", "ready"])
1719def test_clock_ready_handshake_states_do_not_warn(
1720    state: str, caplog: pytest.LogCaptureFixture
1721) -> None:
1722    """Any state but a stall is a normal step of the clock handshake and stays quiet."""
1723    stream = AirPlayStream(_make_player())
1724
1725    with caplog.at_level(logging.DEBUG):
1726        ended = stream._handle_status_line(
1727            f"[STATUS] clock_ready mode=ptp state={state} streak_ms=900 exchanges=4 "
1728            f"ready_in_ms=940 ready_at_unix_ms={START_UNIX_MS}"
1729        )
1730
1731    assert ended is False
1732    assert [record for record in caplog.records if record.levelno >= logging.WARNING] == []
1733
1734
1735def test_elapsed_includes_start_position() -> None:
1736    """Reported progress is the current anchor's media base plus the binary delta."""
1737    player = _make_player()
1738    stream = AirPlayStream(player)
1739    stream._start_position = 12.0
1740
1741    stream._update_elapsed(1.5)
1742
1743    player.set_state_from_stream.assert_called_once_with(
1744        state=PlaybackState.PLAYING,
1745        elapsed_time=13.5,
1746        stream=stream,
1747    )
1748
1749
1750def test_reanchor_status_sets_the_cumulative_total() -> None:
1751    """Each [STATUS] REANCHOR line carries the authoritative total, so it SETS the shift."""
1752    stream = AirPlayStream(_make_player())
1753    assert stream.cumulative_shift_seconds == 0.0
1754
1755    assert (
1756        stream._handle_status_line(
1757            "[STATUS] REANCHOR shifted_frames=67870 total_shifted_frames=67870 sample_rate=44100"
1758        )
1759        is False
1760    )
1761    assert stream.cumulative_shift_seconds == pytest.approx(67870 / 44100)
1762
1763    # the second event reports the running total, never a delta to add
1764    stream._handle_status_line(
1765        "[STATUS] REANCHOR shifted_frames=67870 total_shifted_frames=135740 sample_rate=44100"
1766    )
1767    assert stream.cumulative_shift_seconds == pytest.approx(135740 / 44100)
1768
1769
1770def test_human_readable_reanchor_warn_is_not_counted() -> None:
1771    """The binary's warn line accompanies the status line; counting it would double it."""
1772    stream = AirPlayStream(_make_player())
1773
1774    ended = stream._handle_status_line(
1775        "[AP2] Re-anchored after PCM starvation: shifted_frames=67870 lead_frames=77175 count=1"
1776    )
1777
1778    assert ended is False
1779    assert stream.cumulative_shift_seconds == 0.0
1780
1781
1782def test_reanchor_status_prefers_line_sample_rate() -> None:
1783    """The sample rate carried on the [STATUS] REANCHOR line wins over the stream format."""
1784    player = _make_player()
1785    stream = AirPlayStream(player)
1786    stream.pcm_format = AudioFormat(
1787        content_type=ContentType.PCM_S16LE, sample_rate=48000, bit_depth=16
1788    )
1789
1790    stream._handle_status_line(
1791        "[STATUS] REANCHOR shifted_frames=44100 total_shifted_frames=44100 sample_rate=44100"
1792    )
1793
1794    # converts at 44100 (from the line), not 48000 (the stream format) -> exactly 1.0s
1795    assert stream.cumulative_shift_seconds == pytest.approx(1.0)
1796
1797
1798def test_reanchor_status_ignores_line_without_total() -> None:
1799    """A [STATUS] REANCHOR line missing the total leaves the shift unchanged."""
1800    stream = AirPlayStream(_make_player())
1801    stream.cumulative_shift_seconds = 1.5
1802
1803    stream._handle_status_line("[STATUS] REANCHOR shifted_frames=67870 sample_rate=44100")
1804
1805    assert stream.cumulative_shift_seconds == 1.5
1806
1807
1808@pytest.mark.asyncio
1809async def test_start_resets_reanchor_shift() -> None:
1810    """A START re-anchors from scratch, clearing the accumulated shift."""
1811    stream = AirPlayStream(_make_player())
1812    stream._cli_proc = MagicMock(closed=False)
1813    stream._connected.set()
1814    stream.cumulative_shift_seconds = 3.078
1815
1816    with patch.object(
1817        stream,
1818        "_write_cli_command",
1819        new_callable=AsyncMock,
1820        side_effect=_acking_write_cli_command(stream),
1821    ):
1822        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
1823
1824    assert stream.cumulative_shift_seconds == 0.0
1825
1826
1827@pytest.mark.asyncio
1828async def test_connect_resets_accumulated_shift() -> None:
1829    """A fresh cliairplay process starts from a zero playout shift."""
1830    player = _make_player()
1831    stream = AirPlayStream(player)
1832    stream.cumulative_shift_seconds = 5.0
1833    process = MagicMock(closed=False)
1834    process.start = AsyncMock(return_value=None)
1835
1836    def consume_task(awaitable: Any) -> MagicMock:
1837        awaitable.close()
1838        task = MagicMock()
1839        task.done.return_value = True
1840        return task
1841
1842    player.provider.mass.create_task.side_effect = consume_task
1843    with (
1844        patch.object(stream, "_build_cli_args", new_callable=AsyncMock, return_value=["binary"]),
1845        patch("music_assistant.providers.airplay.stream.AsyncProcess", return_value=process),
1846        patch.object(stream.commands_pipe, "create", new_callable=AsyncMock),
1847    ):
1848        await stream.connect()
1849
1850    assert stream.cumulative_shift_seconds == 0.0
1851
1852
1853@pytest.mark.asyncio
1854async def test_initial_metadata_skips_artwork() -> None:
1855    """The pre-connect metadata push cannot delay setup on artwork rendering."""
1856    player = _make_player()
1857    stream = AirPlayStream(player)
1858    stream._cli_proc = MagicMock(closed=False)
1859    metadata = MagicMock(
1860        title="Track",
1861        artist="Artist",
1862        album="Album",
1863        duration=180,
1864        image_url="image",
1865    )
1866
1867    with (
1868        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
1869        patch.object(
1870            stream,
1871            "_render_and_send_artwork",
1872            new_callable=AsyncMock,
1873        ) as send_artwork,
1874    ):
1875        await stream.send_metadata(0, metadata, send_artwork=False)
1876
1877    assert send_command.await_count == 2
1878    commands = [args.args[0] for args in send_command.await_args_list]
1879    assert "TITLE=Track" in commands[0]
1880    # the progress correction rides along right after the metadata push
1881    assert commands[1].endswith("PROGRESS=0")
1882    send_artwork.assert_not_awaited()
1883
1884
1885@pytest.mark.asyncio
1886async def test_wait_for_connection_pushes_metadata_immediately() -> None:
1887    """
1888    Track metadata is pushed the instant the binary can receive it.
1889
1890    Receivers that gate audio rendering on receiving timeline-anchored metadata
1891    (e.g. Sonos over native AirPlay 2) must not be left silent while a deferred
1892    push is pending, so the metadata callback runs synchronously on connect
1893    while only the volume resend stays on the delayed path.
1894    """
1895    player = _make_player()
1896    player.volume_muted = False
1897    stream = AirPlayStream(player)
1898    stream._connected.set()  # connection already established
1899    player.provider.mass.call_later = MagicMock()
1900    operation_order: list[str] = []
1901
1902    async def wait_for_reader(_timeout: float) -> bool:
1903        operation_order.append("reader")
1904        return True
1905
1906    async def send_current_metadata(**_kwargs: Any) -> None:
1907        operation_order.append("metadata")
1908
1909    with (
1910        patch.object(stream, "_cli_proc", MagicMock()),  # non-None so the method proceeds
1911        patch.object(stream.commands_pipe, "wait_for_reader", side_effect=wait_for_reader),
1912        patch.object(stream, "_send_current_metadata", side_effect=send_current_metadata),
1913        patch.object(stream, "send_cli_command", return_value=None),  # avoid a real coroutine
1914    ):
1915        await stream.wait_for_connection()
1916
1917    # Nothing is written before the binary has a reader on the command pipe.
1918    assert operation_order == ["reader", "metadata"]
1919    # Metadata pushed synchronously on connect...
1920    player._on_player_media_updated.assert_called_once_with()
1921    # ...and never routed through the delayed call_later path.
1922    deferred_callables = [call.args[1] for call in player.provider.mass.call_later.call_args_list]
1923    assert player._on_player_media_updated not in deferred_callables
1924    # The volume resend is still deferred (existing behavior preserved).
1925    assert player.provider.mass.call_later.call_count == 1
1926    assert player.provider.mass.call_later.call_args_list[0].args[0] == 2
1927
1928
1929@pytest.mark.asyncio
1930async def test_deferred_volume_resend_reads_the_state_when_it_fires() -> None:
1931    """The repeated volume push must carry the volume at fire time, not at connect time."""
1932    player = _make_player()
1933    player.volume_muted = False
1934    player.volume_level = 40
1935    stream = AirPlayStream(player)
1936    stream._connected.set()  # connection already established
1937    player.provider.mass.call_later = MagicMock()
1938
1939    with (
1940        patch.object(stream, "_cli_proc", MagicMock()),  # non-None so the method proceeds
1941        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=True)),
1942        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
1943        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
1944    ):
1945        await stream.wait_for_connection()
1946        send_command.assert_awaited_with("VOLUME=40")
1947        deferred = player.provider.mass.call_later.call_args_list[0].args[1]
1948
1949        # a volume change between the connect and the resend firing must survive
1950        player.volume_level = 75
1951        await deferred()
1952        send_command.assert_awaited_with("VOLUME=75")
1953
1954        # ...and so must a mute
1955        player.volume_muted = True
1956        await deferred()
1957
1958    send_command.assert_awaited_with("VOLUME=0")
1959
1960
1961@pytest.mark.asyncio
1962async def test_wait_for_connection_fails_on_an_unread_command_pipe() -> None:
1963    """A binary that never attaches to the command pipe can never be anchored: fail the connect."""
1964    player = _make_player()
1965    player.logger = MagicMock()
1966    player.volume_muted = False
1967    stream = AirPlayStream(player)
1968    stream._connected.set()
1969    stream._cli_proc = MagicMock(closed=False)
1970
1971    with (
1972        patch.object(
1973            stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=False)
1974        ) as wait_for_reader,
1975        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
1976        patch.object(stream, "send_cli_command", return_value=None),
1977        pytest.raises(PlayerCommandFailed, match="command pipe") as err,
1978    ):
1979        await stream.wait_for_connection()
1980
1981    wait_for_reader.assert_awaited_once()
1982    assert player.display_name in str(err.value)
1983
1984
1985@pytest.mark.asyncio
1986async def test_wait_for_connection_stays_quiet_about_a_stopped_stream() -> None:
1987    """A stream torn down while connecting took its command pipe along, which is no fault."""
1988    player = _make_player()
1989    player.logger = MagicMock()
1990    player.volume_muted = False
1991    stream = AirPlayStream(player)
1992    stream._connected.set()
1993    stream._cli_proc = MagicMock(closed=False)
1994    stream._stopping = True
1995
1996    with (
1997        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=False)),
1998        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
1999        patch.object(stream, "send_cli_command", return_value=None),
2000    ):
2001        await stream.wait_for_connection()
2002
2003    player.logger.warning.assert_not_called()
2004
2005
2006@pytest.mark.asyncio
2007async def test_prepare_artwork_returns_cache_path() -> None:
2008    """Artwork preparation returns the shared cache path without a per-player copy."""
2009    player = _make_player()
2010    stream = AirPlayStream(player)
2011    image_url = "https://example.com/artwork.png"
2012    cached_path = "/cache/thumbnails/artwork_flat.jpg"
2013
2014    with patch(
2015        "music_assistant.providers.airplay.stream.get_image_thumb_path",
2016        new=AsyncMock(return_value=cached_path),
2017    ) as get_thumb_path:
2018        result = await stream._prepare_artwork(image_url, 1)
2019
2020    assert result == cached_path
2021    assert not hasattr(stream, "_artwork_paths")
2022    get_thumb_path.assert_awaited_once_with(
2023        stream.mass,
2024        image_url,
2025        AIRPLAY_ARTWORK_SIZE,
2026        "",
2027        image_format="JPEG",
2028        flatten_transparency=True,
2029    )
2030
2031
2032@pytest.mark.asyncio
2033async def test_stop_cleans_up_when_stop_command_fails() -> None:
2034    """A command-pipe failure cannot skip process and stream cleanup."""
2035    player = _make_player()
2036    stream = AirPlayStream(player)
2037    process = MagicMock()
2038    process.closed = False
2039    process.kill = AsyncMock()
2040    stream._cli_proc = process
2041
2042    with (
2043        patch.object(
2044            stream.commands_pipe,
2045            "write",
2046            new_callable=AsyncMock,
2047            side_effect=OSError("command pipe failed"),
2048        ),
2049        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
2050        pytest.raises(OSError, match="command pipe failed"),
2051    ):
2052        await stream.stop(force=True)
2053
2054    assert stream._stopped is True
2055    assert stream._cleanup_complete is True
2056    remove_pipe.assert_awaited_once()
2057    process.kill.assert_awaited_once()
2058    player.set_state_from_stream.assert_called_once_with(
2059        state=PlaybackState.IDLE,
2060        elapsed_time=0,
2061        stream=stream,
2062    )
2063
2064
2065@pytest.mark.asyncio
2066async def test_stop_awaits_cancelled_stdout_reader() -> None:
2067    """Stream teardown waits for the stdout reader to release process resources."""
2068    player = _make_player()
2069    stream = AirPlayStream(player)
2070    process = MagicMock()
2071    process.closed = False
2072    process.kill = AsyncMock()
2073    stream._cli_proc = process
2074    reader_started = asyncio.Event()
2075
2076    async def _stdout_reader() -> None:
2077        reader_started.set()
2078        await asyncio.Event().wait()
2079
2080    reader_task = asyncio.create_task(_stdout_reader())
2081    stream._stdout_reader_task = reader_task
2082    await reader_started.wait()
2083
2084    with (
2085        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock),
2086        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2087    ):
2088        await stream.stop(force=True)
2089
2090    assert reader_task.cancelled()
2091    process.kill.assert_awaited_once()
2092
2093
2094@pytest.mark.asyncio
2095async def test_force_stop_does_not_wait_for_artwork_render() -> None:
2096    """Force-stop tears down immediately while remote artwork rendering finishes."""
2097    player = _make_player()
2098    stream = AirPlayStream(player)
2099    process = MagicMock()
2100    process.closed = False
2101    process.kill = AsyncMock()
2102    stream._cli_proc = process
2103    metadata = MagicMock(
2104        title="Track",
2105        artist="Artist",
2106        album="Album",
2107        duration=180,
2108        image_url="slow-image",
2109    )
2110    artwork_started = asyncio.Event()
2111    release_artwork = asyncio.Event()
2112
2113    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2114        artwork_started.set()
2115        await release_artwork.wait()
2116        return "late.jpg"
2117
2118    with (
2119        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock),
2120        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2121        patch.object(
2122            stream,
2123            "_prepare_artwork",
2124            new_callable=AsyncMock,
2125            side_effect=_prepare_artwork,
2126        ),
2127    ):
2128        metadata_task = asyncio.create_task(stream.send_metadata(0, metadata))
2129        await artwork_started.wait()
2130        await asyncio.wait_for(stream.stop(force=True), timeout=0.5)
2131        release_artwork.set()
2132        await metadata_task
2133
2134    assert stream._cleanup_complete is True
2135    process.kill.assert_awaited_once()
2136
2137
2138@pytest.mark.asyncio
2139async def test_process_eof_cleans_up_command_pipe() -> None:
2140    """A naturally ended CLI stream removes its command pipe."""
2141    player = _make_player()
2142    stream = AirPlayStream(player)
2143    process = MagicMock()
2144    stream._cli_proc = process
2145
2146    async def _stderr_lines() -> AsyncGenerator[str]:
2147        yield "[STATUS] eof"
2148
2149    with (
2150        patch.object(process, "iter_stderr", return_value=_stderr_lines()),
2151        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
2152    ):
2153        await stream._stderr_reader()
2154
2155    assert stream._stopped is True
2156    remove_pipe.assert_awaited_once()
2157    player.schedule_group_rejoin.assert_not_called()
2158
2159
2160async def _run_unexpected_process_death(player: MagicMock) -> AirPlayStream:
2161    """Drive the stderr reader through an unexpected process exit."""
2162    stream = AirPlayStream(player)
2163    process = MagicMock()
2164    stream._cli_proc = process
2165
2166    async def _stderr_lines() -> AsyncGenerator[str]:
2167        yield "some final log line"
2168
2169    with (
2170        patch.object(process, "iter_stderr", return_value=_stderr_lines()),
2171        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2172    ):
2173        await stream._stderr_reader()
2174    return stream
2175
2176
2177@pytest.mark.asyncio
2178async def test_unexpected_death_of_synced_child_schedules_rejoin() -> None:
2179    """A grouped member whose process dies unexpectedly gets a re-join scheduled."""
2180    player = _make_player()
2181    player.synced_to = "leader"
2182    player.group_members = []
2183    # the leader's other members are captured as fallback candidates in case
2184    # leadership transfers while the re-join backoff runs
2185    leader = MagicMock()
2186    leader.group_members = ["leader", player.player_id, "sibling"]
2187    player.provider.mass.players.get_player.return_value = leader
2188
2189    stream = await _run_unexpected_process_death(player)
2190
2191    player.schedule_group_rejoin.assert_called_once_with(["leader", "sibling"])
2192    player.set_state_from_stream.assert_called_once_with(
2193        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2194    )
2195
2196
2197@pytest.mark.asyncio
2198async def test_unexpected_death_of_leader_schedules_rejoin_to_members() -> None:
2199    """A dying leader re-joins towards its surviving members (leadership transfers)."""
2200    player = _make_player()
2201    player.synced_to = None
2202    player.group_members = [player.player_id, "child1", "child2"]
2203
2204    await _run_unexpected_process_death(player)
2205
2206    player.schedule_group_rejoin.assert_called_once_with(["child1", "child2"])
2207    # the controller sets the leader's final state (transfer or dissolve)
2208    player.set_state_from_stream.assert_not_called()
2209
2210
2211@pytest.mark.asyncio
2212async def test_unexpected_death_of_solo_player_schedules_no_rejoin() -> None:
2213    """An ungrouped player's process death only marks the player idle."""
2214    player = _make_player()
2215    player.synced_to = None
2216    player.group_members = []
2217
2218    stream = await _run_unexpected_process_death(player)
2219
2220    player.schedule_group_rejoin.assert_not_called()
2221    player.set_state_from_stream.assert_called_once_with(
2222        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2223    )
2224
2225
2226@pytest.mark.asyncio
2227async def test_unexpected_death_of_static_group_member_drops_member_only() -> None:
2228    """A static group member's death drops just that member, never the whole group."""
2229    player = _make_player()
2230    player.synced_to = "leader"
2231    player.group_members = []
2232    # the player is a static member of an actively playing group player, for
2233    # which cmd_ungroup would release (stop) the WHOLE group
2234    player.state.active_group = "syncgroup1"
2235    group_player = MagicMock()
2236    group_player.static_group_members = ["leader", player.player_id]
2237    leader = MagicMock()
2238    leader.group_members = ["leader", player.player_id]
2239    player.provider.mass.players.get_player.side_effect = lambda player_id: {
2240        "syncgroup1": group_player,
2241        "leader": leader,
2242    }.get(player_id)
2243
2244    await _run_unexpected_process_death(player)
2245
2246    players_controller = player.provider.mass.players
2247    players_controller.cmd_set_members.assert_called_once_with(
2248        "leader", player_ids_to_remove=[player.player_id]
2249    )
2250    players_controller.cmd_ungroup.assert_not_called()
2251    player.schedule_group_rejoin.assert_called_once_with(["leader"])
2252
2253
2254@pytest.mark.asyncio
2255async def test_process_eof_during_render_does_not_send_artwork() -> None:
2256    """A cache lookup finishing after EOF cannot send stale artwork."""
2257    player = _make_player()
2258    stream = AirPlayStream(player)
2259    render_started = asyncio.Event()
2260    release_render = asyncio.Event()
2261
2262    async def _get_image_thumb_path(*_args: Any, **_kwargs: Any) -> str:
2263        render_started.set()
2264        await release_render.wait()
2265        return "/cache/thumbnails/artwork.jpg"
2266
2267    with (
2268        patch(
2269            "music_assistant.providers.airplay.stream.get_image_thumb_path",
2270            new_callable=AsyncMock,
2271            side_effect=_get_image_thumb_path,
2272        ),
2273        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2274    ):
2275        render_task = asyncio.create_task(stream._render_and_send_artwork("image", 1))
2276        await render_started.wait()
2277        stream._stopped = True
2278        release_render.set()
2279        await render_task
2280
2281    send_command.assert_not_awaited()
2282
2283
2284@pytest.mark.asyncio
2285async def test_concurrent_metadata_updates_only_send_latest_artwork() -> None:
2286    """An older slow artwork render cannot overwrite a newer track update."""
2287    player = _make_player()
2288    stream = AirPlayStream(player)
2289    process = MagicMock()
2290    process.closed = False
2291    stream._cli_proc = process
2292    first_render_started = asyncio.Event()
2293    release_first_render = asyncio.Event()
2294
2295    old_metadata = MagicMock(
2296        title="Old track",
2297        artist="Artist",
2298        album="Album",
2299        duration=180,
2300        image_url="old-image",
2301    )
2302    new_metadata = MagicMock(
2303        title="New track",
2304        artist="Artist",
2305        album="Album",
2306        duration=180,
2307        image_url="new-image",
2308    )
2309
2310    async def _prepare_artwork(image_url: str, _generation: int) -> str:
2311        if image_url == "old-image":
2312            first_render_started.set()
2313            await release_first_render.wait()
2314            return "old.jpg"
2315        return "new.jpg"
2316
2317    with (
2318        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2319        patch.object(
2320            stream,
2321            "_prepare_artwork",
2322            new_callable=AsyncMock,
2323            side_effect=_prepare_artwork,
2324        ),
2325    ):
2326        old_task = asyncio.create_task(stream.send_metadata(0, old_metadata))
2327        await first_render_started.wait()
2328        new_task = asyncio.create_task(stream.send_metadata(0, new_metadata))
2329        await asyncio.sleep(0)
2330        assert stream._metadata_generation == 2
2331        release_first_render.set()
2332        await asyncio.gather(old_task, new_task)
2333
2334    commands = [call.args[0].decode() for call in write_command.await_args_list]
2335    assert any("TITLE=New track" in command for command in commands)
2336    assert not any("ARTWORK=old.jpg" in command for command in commands)
2337    assert commands[-1] == "ARTWORK=new.jpg\n"
2338
2339
2340@pytest.mark.asyncio
2341async def test_metadata_revert_resends_text_after_superseded_artwork() -> None:
2342    """Reverting while artwork renders restores the previously displayed track text."""
2343    player = _make_player()
2344    stream = AirPlayStream(player)
2345    process = MagicMock()
2346    process.closed = False
2347    stream._cli_proc = process
2348    first_metadata = MagicMock(
2349        title="First track",
2350        artist="Artist",
2351        album="Album",
2352        duration=180,
2353        image_url=None,
2354    )
2355    second_metadata = MagicMock(
2356        title="Second track",
2357        artist="Artist",
2358        album="Album",
2359        duration=180,
2360        image_url="second-image",
2361    )
2362    first_checksum = "First track|Artist|Album|180|None"
2363    stream._metadata_text_checksum = first_checksum
2364    stream._pending_metadata_checksum = first_checksum
2365    artwork_started = asyncio.Event()
2366    release_artwork = asyncio.Event()
2367
2368    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2369        artwork_started.set()
2370        await release_artwork.wait()
2371        return "second.jpg"
2372
2373    with (
2374        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2375        patch.object(
2376            stream,
2377            "_prepare_artwork",
2378            new_callable=AsyncMock,
2379            side_effect=_prepare_artwork,
2380        ),
2381    ):
2382        second_task = asyncio.create_task(stream.send_metadata(0, second_metadata))
2383        await artwork_started.wait()
2384        revert_task = asyncio.create_task(stream.send_metadata(0, first_metadata))
2385        await asyncio.sleep(0)
2386        release_artwork.set()
2387        await asyncio.gather(second_task, revert_task)
2388
2389    metadata_commands = [
2390        call.args[0].decode()
2391        for call in write_command.await_args_list
2392        if "ACTION=SENDMETA" in call.args[0].decode()
2393    ]
2394    assert "TITLE=Second track" in metadata_commands[0]
2395    assert "TITLE=First track" in metadata_commands[-1]
2396
2397
2398@pytest.mark.asyncio
2399async def test_repeated_metadata_retries_superseded_artwork() -> None:
2400    """A B-to-C-to-B update sequence still applies B artwork after supersession."""
2401    player = _make_player()
2402    stream = AirPlayStream(player)
2403    process = MagicMock()
2404    process.closed = False
2405    stream._cli_proc = process
2406    initial_text_checksum = "item-initial|Initial|Artist|Album"
2407    initial_checksum = f"{initial_text_checksum}|initial-image"
2408    stream._metadata_artwork_checksum = "initial-image"
2409    stream._metadata_text_checksum = initial_text_checksum
2410    stream._pending_metadata_checksum = initial_checksum
2411    metadata_b = MagicMock(
2412        queue_item_id="item-b",
2413        title="Track B",
2414        artist="Artist",
2415        album="Album",
2416        duration=180,
2417        image_url="b-image",
2418    )
2419    metadata_c = MagicMock(
2420        queue_item_id="item-c",
2421        title="Track C",
2422        artist="Artist",
2423        album="Album",
2424        duration=180,
2425        image_url="c-image",
2426    )
2427    first_artwork_started = asyncio.Event()
2428    release_first_artwork = asyncio.Event()
2429    c_artwork_started = asyncio.Event()
2430    release_c_artwork = asyncio.Event()
2431    b_render_count = 0
2432
2433    async def _prepare_artwork(image_url: str, _generation: int) -> str:
2434        nonlocal b_render_count
2435        if image_url == "b-image":
2436            b_render_count += 1
2437            if b_render_count == 1:
2438                first_artwork_started.set()
2439                await release_first_artwork.wait()
2440                return "b-stale.jpg"
2441            return "b-final.jpg"
2442        c_artwork_started.set()
2443        await release_c_artwork.wait()
2444        return "c.jpg"
2445
2446    with (
2447        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2448        patch.object(
2449            stream,
2450            "_prepare_artwork",
2451            new_callable=AsyncMock,
2452            side_effect=_prepare_artwork,
2453        ) as prepare_artwork,
2454    ):
2455        first_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
2456        await first_artwork_started.wait()
2457        c_task = asyncio.create_task(stream.send_metadata(0, metadata_c))
2458        await c_artwork_started.wait()
2459        final_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
2460        await final_b_task
2461        release_first_artwork.set()
2462        release_c_artwork.set()
2463        await asyncio.gather(first_b_task, c_task)
2464
2465    rendered_images = [args.args[0] for args in prepare_artwork.await_args_list]
2466    commands = [args.args[0].decode() for args in write_command.await_args_list]
2467    assert rendered_images == ["b-image", "c-image", "b-image"]
2468    assert "ARTWORK=b-stale.jpg\n" not in commands
2469    assert "ARTWORK=c.jpg\n" not in commands
2470    assert commands[-1] == "ARTWORK=b-final.jpg\n"
2471    assert stream._metadata_artwork_checksum == "b-image"
2472
2473
2474@pytest.mark.asyncio
2475async def test_send_metadata_passes_cached_artwork_path_to_binary() -> None:
2476    """The ARTWORK command passes the absolute cache path returned by preparation."""
2477    player = _make_player()
2478    stream = AirPlayStream(player)
2479    metadata = MagicMock(
2480        duration=180,
2481        title="Track",
2482        artist="Artist",
2483        album="Album",
2484        image_url="https://example.com/artwork.png",
2485    )
2486    cached_path = "/cache/thumbnails/artwork_flat.jpg"
2487    send_command = AsyncMock()
2488
2489    with (
2490        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value=cached_path)),
2491        patch.object(stream, "send_cli_command", new=send_command),
2492    ):
2493        await stream.send_metadata(None, metadata)
2494
2495    assert send_command.await_args_list[-1] == call(f"ARTWORK={cached_path}")
2496
2497
2498@pytest.mark.asyncio
2499async def test_failed_artwork_delivery_is_retried() -> None:
2500    """A dropped ARTWORK command remains pending for the next metadata update."""
2501    stream = AirPlayStream(_make_player())
2502    metadata = MagicMock(
2503        queue_item_id="item-1",
2504        duration=180,
2505        title="Track",
2506        artist="Artist",
2507        album="Album",
2508        image_url="image",
2509    )
2510    artwork_path = "/cache/thumbnails/artwork.jpg"
2511
2512    with (
2513        patch.object(
2514            stream,
2515            "_prepare_artwork",
2516            new_callable=AsyncMock,
2517            return_value=artwork_path,
2518        ) as prepare_artwork,
2519        patch.object(
2520            stream,
2521            "send_cli_command",
2522            new_callable=AsyncMock,
2523            side_effect=[True, False, True],
2524        ) as send_command,
2525    ):
2526        await stream.send_metadata(None, metadata)
2527        assert stream._metadata_artwork_checksum == ""
2528        await stream.send_metadata(None, metadata)
2529
2530    assert prepare_artwork.await_count == 2
2531    assert [args.args[0] for args in send_command.await_args_list].count(
2532        f"ARTWORK={artwork_path}"
2533    ) == 2
2534    assert stream._metadata_artwork_checksum == "image"
2535
2536
2537@pytest.mark.asyncio
2538async def test_text_refinement_keeps_delivered_artwork_settled() -> None:
2539    """A text-only metadata update after delivery does not re-render unchanged art."""
2540    stream = AirPlayStream(_make_player())
2541    metadata = MagicMock(
2542        duration=180,
2543        title="Track",
2544        artist="Artist",
2545        album="Album",
2546        image_url="image",
2547    )
2548    refined = MagicMock(
2549        queue_item_id=metadata.queue_item_id,
2550        duration=180,
2551        title="Track (Remastered)",
2552        artist="Artist",
2553        album="Album",
2554        image_url="image",
2555    )
2556    artwork_path = "/cache/thumbnails/artwork.jpg"
2557
2558    with (
2559        patch.object(
2560            stream,
2561            "_prepare_artwork",
2562            new_callable=AsyncMock,
2563            return_value=artwork_path,
2564        ) as prepare_artwork,
2565        patch.object(
2566            stream,
2567            "send_cli_command",
2568            new_callable=AsyncMock,
2569            return_value=True,
2570        ) as send_command,
2571    ):
2572        await stream.send_metadata(None, metadata)
2573        assert stream._metadata_artwork_checksum == "image"
2574        # the refinement bumps the metadata generation (pending identity
2575        # changed), which before the identity settle re-armed the artwork
2576        await stream.send_metadata(None, refined)
2577
2578    prepare_artwork.assert_awaited_once()
2579    commands = [args.args[0] for args in send_command.await_args_list]
2580    assert commands.count(f"ARTWORK={artwork_path}") == 1
2581    assert "TITLE=Track (Remastered)" in commands[-1]
2582
2583
2584# --- Structured connect failures reported by the binary ---
2585
2586
2587@pytest.mark.asyncio
2588async def test_connect_error_status_line_is_parsed() -> None:
2589    """The machine-readable failure line is captured with all of its fields."""
2590    stream = AirPlayStream(_make_player())
2591
2592    stream._handle_status_line(
2593        '[STATUS] error code=auth_required http=401 detail="RTSP setup rejected"'
2594    )
2595
2596    assert stream._connect_error == CliError("auth_required", 401, "RTSP setup rejected")
2597
2598
2599@pytest.mark.asyncio
2600async def test_started_ack_status_line_parsing() -> None:
2601    """A started ack releases the START wait; a malformed one carries no details."""
2602    stream = AirPlayStream(_make_player())
2603
2604    stream._handle_status_line(
2605        "[STATUS] started requested_unix_ms=1750000000000 at_unix_ms=1750000000004"
2606    )
2607    assert stream._started.is_set()
2608    assert stream._start_ack == (1750000000000, 1750000000004)
2609
2610    stream._started.clear()
2611    stream._start_ack = None
2612    stream._handle_status_line("[STATUS] started requested_unix_ms=garbage at_unix_ms=1")
2613    assert stream._started.is_set()
2614    assert stream._start_ack is None
2615
2616
2617# --- Post-commit anchor verification ---
2618
2619
2620def test_anchor_corrected_status_line_rebases_the_position() -> None:
2621    """A correction carrying a content cut moves the reported-position base by it."""
2622    stream = AirPlayStream(_make_player())
2623    stream._start_position = 12.0
2624
2625    ended = stream._handle_status_line(
2626        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
2627        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
2628    )
2629
2630    assert ended is False
2631    assert stream._start_position == 12.5
2632
2633
2634def test_anchor_corrected_status_line_logs_a_warning(caplog: pytest.LogCaptureFixture) -> None:
2635    """The correction is logged loudly, including the display name and the delta."""
2636    stream = AirPlayStream(_make_player())
2637
2638    with caplog.at_level(logging.WARNING):
2639        stream._handle_status_line(
2640            "[STATUS] anchor_corrected requested_unix_ms=0 "
2641            "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
2642        )
2643
2644    assert "Player A" in caplog.text
2645    assert "+500 ms" in caplog.text
2646
2647
2648def test_anchor_corrected_status_line_tolerates_malformed_line() -> None:
2649    """A malformed anchor_corrected line is dropped instead of raising or rebasing."""
2650    stream = AirPlayStream(_make_player())
2651    stream._start_position = 12.0
2652
2653    ended = stream._handle_status_line("[STATUS] anchor_corrected requested_unix_ms=garbage")
2654
2655    assert ended is False
2656    assert stream._start_position == 12.0
2657
2658
2659def test_content_cut_short_rebases_the_position_and_warns(
2660    caplog: pytest.LogCaptureFixture,
2661) -> None:
2662    """A cut that ended early gives back the ms the correction over-advanced the base by."""
2663    stream = AirPlayStream(_make_player())
2664    stream._start_position = 12.0
2665    stream._handle_status_line(
2666        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
2667        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
2668    )
2669    assert stream._start_position == 12.5
2670
2671    with caplog.at_level(logging.WARNING):
2672        ended = stream._handle_status_line(
2673            "[STATUS] content_cut requested_ms=500 cut_ms=180 cut_bytes=31752 drain_ms=210"
2674        )
2675
2676    assert ended is False
2677    assert stream._start_position == pytest.approx(12.18)
2678    assert "AirPlay content cut" in caplog.text
2679    assert "Player A" in caplog.text
2680    assert "320 ms short" in caplog.text
2681
2682
2683def test_content_cut_in_full_leaves_the_position_alone(caplog: pytest.LogCaptureFixture) -> None:
2684    """A cut that took what it asked for needs no correction and stays quiet."""
2685    stream = AirPlayStream(_make_player())
2686    stream._start_position = 12.0
2687    stream._handle_status_line(
2688        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
2689        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
2690    )
2691
2692    caplog.clear()
2693    with caplog.at_level(logging.WARNING):
2694        # a few ms below the request is byte quantization, not a short cut
2695        stream._handle_status_line(
2696            "[STATUS] content_cut requested_ms=500 cut_ms=498 cut_bytes=87887 drain_ms=505"
2697        )
2698
2699    assert stream._start_position == 12.5
2700    assert caplog.text == ""
2701
2702
2703def test_content_cut_after_a_new_anchor_is_not_reconciled() -> None:
2704    """A cut settling after a START must not be taken off that START's absolute base."""
2705    stream = AirPlayStream(_make_player())
2706    stream._start_position = 12.0
2707    stream._handle_status_line(
2708        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
2709        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
2710    )
2711    stream.rebase_position(30_000)
2712
2713    stream._handle_status_line(
2714        "[STATUS] content_cut requested_ms=500 cut_ms=0 cut_bytes=0 drain_ms=12"
2715    )
2716
2717    assert stream._start_position == 30.0
2718
2719
2720def test_content_cut_status_line_tolerates_malformed_line() -> None:
2721    """A malformed content_cut line is dropped instead of raising or rebasing."""
2722    stream = AirPlayStream(_make_player())
2723    stream._start_position = 12.0
2724    stream._handle_status_line(
2725        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
2726        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
2727    )
2728
2729    ended = stream._handle_status_line("[STATUS] content_cut requested_ms=500 cut_ms=garbage")
2730
2731    assert ended is False
2732    assert stream._start_position == 12.5
2733
2734
2735def test_clock_verified_status_line_is_debug_logged(caplog: pytest.LogCaptureFixture) -> None:
2736    """A clock_verified line needs no server action beyond a debug note of the margin."""
2737    stream = AirPlayStream(_make_player())
2738
2739    with caplog.at_level(logging.DEBUG):
2740        ended = stream._handle_status_line("[STATUS] clock_verified margin_ms=42")
2741
2742    assert ended is False
2743    assert "42" in caplog.text
2744
2745
2746@pytest.mark.asyncio
2747async def test_connect_error_status_line_tolerates_missing_fields() -> None:
2748    """A failure line without http/detail still yields the reported code."""
2749    stream = AirPlayStream(_make_player())
2750
2751    stream._handle_status_line("[STATUS] error code=connect_failed")
2752
2753    assert stream._connect_error == CliError("connect_failed", 0, "")
2754
2755
2756@pytest.mark.asyncio
2757async def test_auth_required_surfaces_password_required_error() -> None:
2758    """A device asking for a password produces an actionable, translated error."""
2759    stream = AirPlayStream(_make_player())
2760    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
2761    stream._process_ended.set()
2762
2763    with (
2764        patch.object(stream, "_cli_proc", MagicMock()),
2765        pytest.raises(PlayerCommandFailed) as err,
2766    ):
2767        await stream.wait_for_connection()
2768
2769    assert err.value.translation_key == "password_required"
2770
2771
2772@pytest.mark.asyncio
2773async def test_auth_failed_surfaces_authentication_failed_error() -> None:
2774    """A rejected password is reported as an authentication failure, not a timeout."""
2775    stream = AirPlayStream(_make_player())
2776    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
2777    stream._process_ended.set()
2778
2779    with (
2780        patch.object(stream, "_cli_proc", MagicMock()),
2781        pytest.raises(PlayerCommandFailed) as err,
2782    ):
2783        await stream.wait_for_connection()
2784
2785    assert err.value.translation_key == "authentication_failed"
2786
2787
2788@pytest.mark.asyncio
2789async def test_generic_connect_failure_keeps_the_timeout_semantics() -> None:
2790    """A non-auth failure keeps raising the plain timeout its callers already handle."""
2791    stream = AirPlayStream(_make_player())
2792    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
2793    stream._process_ended.set()
2794
2795    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
2796        await stream.wait_for_connection()
2797
2798
2799@pytest.mark.asyncio
2800async def test_dead_process_fails_the_connect_wait_immediately() -> None:
2801    """
2802    A binary that reports no reason at all leaves the wait its plain timeout error.
2803
2804    It must still end the moment the process is gone instead of running out the
2805    full connect timeout.
2806    """
2807    stream = AirPlayStream(_make_player())
2808    stream._process_ended.set()  # process died without emitting a [STATUS] error line
2809
2810    started = asyncio.get_running_loop().time()
2811    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
2812        await stream.wait_for_connection()
2813
2814    assert asyncio.get_running_loop().time() - started < 1
2815
2816
2817@pytest.mark.asyncio
2818async def test_auth_failed_marks_the_stored_password_invalid() -> None:
2819    """A rejected password is persisted so the player keeps offering its setup action."""
2820    player = _make_player()
2821    stream = AirPlayStream(player)
2822
2823    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
2824
2825    player.set_password_invalid.assert_called_once_with(True)
2826
2827
2828@pytest.mark.asyncio
2829async def test_auth_required_also_marks_the_password_as_needed() -> None:
2830    """
2831    A device that demanded a password we could not supply flips into setup.
2832
2833    Devices can enforce a password without announcing it (stale TXT records), so
2834    the runtime signal must set the marker too - it is the only reliable one.
2835    """
2836    player = _make_player()
2837    stream = AirPlayStream(player)
2838
2839    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
2840
2841    player.set_password_invalid.assert_called_once_with(True)
2842
2843
2844@pytest.mark.asyncio
2845async def test_plain_connect_failures_leave_the_password_marker_alone() -> None:
2846    """A non-authentication failure says nothing about the stored password."""
2847    player = _make_player()
2848    stream = AirPlayStream(player)
2849
2850    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
2851
2852    player.set_password_invalid.assert_not_called()
2853
2854
2855@pytest.mark.asyncio
2856async def test_successful_connect_clears_the_password_marker() -> None:
2857    """Whatever the device accepted is a working password."""
2858    player = _make_player()
2859    stream = AirPlayStream(player)
2860
2861    stream._handle_status_line("[STATUS] connected")
2862
2863    assert stream.connected is True
2864    player.set_password_invalid.assert_called_once_with(False)
2865
2866
2867# --- Password preflight ---
2868
2869
2870@pytest.mark.asyncio
2871async def test_connect_refuses_password_device_without_password_or_credentials() -> None:
2872    """A password-protected AirPlay 2 device with nothing to authenticate never spawns a process."""
2873    player = _make_player()
2874    player.password_required = True
2875    player.get_setup_value = MagicMock(return_value=None)
2876    stream = AirPlayStream(player)
2877
2878    with pytest.raises(PlayerCommandFailed) as err:
2879        await stream.connect()
2880
2881    assert err.value.translation_key == "password_required"
2882    assert stream._cli_proc is None
2883
2884
2885@pytest.mark.asyncio
2886async def test_password_preflight_passes_with_password_or_credentials() -> None:
2887    """Either a configured password or stored credentials let the connect proceed."""
2888    player = _make_player()
2889    player.password_required = True
2890    player.get_setup_value = MagicMock(return_value=None)
2891    player.config.get_value = MagicMock(
2892        side_effect=lambda key, default=None: "s3cret" if key == CONF_PASSWORD else default
2893    )
2894    AirPlayStream(player)._check_password_preflight()
2895
2896    # credentials alone are enough: the binary's pair-verify leg may still succeed
2897    player.config.get_value = MagicMock(side_effect=lambda _key, default=None: default)
2898    player.get_setup_value = MagicMock(
2899        side_effect=lambda key, default=None: (
2900            "ab" * 96 if key == CONF_AIRPLAY_CREDENTIALS else default
2901        )
2902    )
2903    AirPlayStream(player)._check_password_preflight()
2904
2905
2906@pytest.mark.asyncio
2907async def test_password_preflight_skipped_for_raop() -> None:
2908    """The preflight only guards the native AirPlay 2 flow; RAOP carries its own password."""
2909    player = _make_player()
2910    player.protocol = StreamingProtocol.RAOP
2911    player.password_required = True
2912    player.get_setup_value = MagicMock(return_value=None)
2913
2914    AirPlayStream(player)._check_password_preflight()
2915