music-assistant-server

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