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