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_fails_on_an_unread_command_pipe() -> None:
2255    """A binary that never attaches to the command pipe can never be anchored: fail the connect."""
2256    player = _make_player()
2257    player.logger = MagicMock()
2258    player.volume_muted = False
2259    stream = AirPlayStream(player)
2260    stream._connected.set()
2261    stream._cli_proc = _make_cli_proc()
2262
2263    with (
2264        patch.object(
2265            stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=False)
2266        ) as wait_for_reader,
2267        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2268        patch.object(stream, "send_cli_command", return_value=None),
2269        pytest.raises(PlayerCommandFailed, match="command pipe") as err,
2270    ):
2271        await stream.wait_for_connection()
2272
2273    wait_for_reader.assert_awaited_once()
2274    assert player.display_name in str(err.value)
2275
2276
2277@pytest.mark.asyncio
2278async def test_wait_for_connection_stays_quiet_about_a_stopped_stream() -> None:
2279    """A stream torn down while connecting took its command pipe along, which is no fault."""
2280    player = _make_player()
2281    player.logger = MagicMock()
2282    player.volume_muted = False
2283    stream = AirPlayStream(player)
2284    stream._connected.set()
2285    stream._cli_proc = _make_cli_proc()
2286    stream._stopping = True
2287
2288    with (
2289        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=False)),
2290        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2291        patch.object(stream, "send_cli_command", return_value=None),
2292    ):
2293        await stream.wait_for_connection()
2294
2295    player.logger.warning.assert_not_called()
2296
2297
2298@pytest.mark.asyncio
2299async def test_prepare_artwork_returns_cache_path() -> None:
2300    """Artwork preparation returns the shared cache path without a per-player copy."""
2301    player = _make_player()
2302    stream = AirPlayStream(player)
2303    image_url = "https://example.com/artwork.png"
2304    cached_path = "/cache/thumbnails/artwork_flat.jpg"
2305
2306    with patch(
2307        "music_assistant.providers.airplay.stream.get_image_thumb_path",
2308        new=AsyncMock(return_value=cached_path),
2309    ) as get_thumb_path:
2310        result = await stream._prepare_artwork(image_url, 1)
2311
2312    assert result == cached_path
2313    assert not hasattr(stream, "_artwork_paths")
2314    get_thumb_path.assert_awaited_once_with(
2315        stream.mass,
2316        image_url,
2317        AIRPLAY_ARTWORK_SIZE,
2318        "",
2319        image_format="JPEG",
2320        flatten_transparency=True,
2321    )
2322
2323
2324@pytest.mark.asyncio
2325async def test_stop_cleans_up_when_stop_command_fails() -> None:
2326    """A command-pipe failure cannot skip process and stream cleanup."""
2327    player = _make_player()
2328    stream = AirPlayStream(player)
2329    process = MagicMock()
2330    process.closed = False
2331    process.kill = AsyncMock()
2332    stream._cli_proc = process
2333
2334    with (
2335        patch.object(
2336            stream.commands_pipe,
2337            "write",
2338            new_callable=AsyncMock,
2339            side_effect=OSError("command pipe failed"),
2340        ),
2341        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
2342        pytest.raises(OSError, match="command pipe failed"),
2343    ):
2344        await stream.stop(force=True)
2345
2346    assert stream._stopped is True
2347    assert stream._cleanup_complete is True
2348    remove_pipe.assert_awaited_once()
2349    process.kill.assert_awaited_once()
2350    player.set_state_from_stream.assert_called_once_with(
2351        state=PlaybackState.IDLE,
2352        elapsed_time=0,
2353        stream=stream,
2354    )
2355
2356
2357@pytest.mark.asyncio
2358async def test_stop_awaits_cancelled_stdout_reader() -> None:
2359    """Stream teardown waits for the stdout reader to release process resources."""
2360    player = _make_player()
2361    stream = AirPlayStream(player)
2362    process = MagicMock()
2363    process.closed = False
2364    process.kill = AsyncMock()
2365    stream._cli_proc = process
2366    reader_started = asyncio.Event()
2367
2368    async def _stdout_reader() -> None:
2369        reader_started.set()
2370        await asyncio.Event().wait()
2371
2372    reader_task = asyncio.create_task(_stdout_reader())
2373    stream._stdout_reader_task = reader_task
2374    await reader_started.wait()
2375
2376    with (
2377        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock),
2378        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2379    ):
2380        await stream.stop(force=True)
2381
2382    assert reader_task.cancelled()
2383    process.kill.assert_awaited_once()
2384
2385
2386@pytest.mark.asyncio
2387async def test_force_stop_does_not_wait_for_artwork_render() -> None:
2388    """Force-stop tears down immediately while remote artwork rendering finishes."""
2389    player = _make_player()
2390    stream = AirPlayStream(player)
2391    process = MagicMock()
2392    process.closed = False
2393    process.kill = AsyncMock()
2394    stream._cli_proc = process
2395    metadata = MagicMock(
2396        title="Track",
2397        artist="Artist",
2398        album="Album",
2399        duration=180,
2400        image_url="slow-image",
2401    )
2402    artwork_started = asyncio.Event()
2403    release_artwork = asyncio.Event()
2404
2405    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2406        artwork_started.set()
2407        await release_artwork.wait()
2408        return "late.jpg"
2409
2410    with (
2411        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock),
2412        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2413        patch.object(
2414            stream,
2415            "_prepare_artwork",
2416            new_callable=AsyncMock,
2417            side_effect=_prepare_artwork,
2418        ),
2419    ):
2420        metadata_task = asyncio.create_task(stream.send_metadata(0, metadata))
2421        await artwork_started.wait()
2422        await asyncio.wait_for(stream.stop(force=True), timeout=0.5)
2423        release_artwork.set()
2424        await metadata_task
2425
2426    assert stream._cleanup_complete is True
2427    process.kill.assert_awaited_once()
2428
2429
2430@pytest.mark.asyncio
2431async def test_process_eof_cleans_up_command_pipe() -> None:
2432    """A naturally ended CLI stream removes its command pipe."""
2433    player = _make_player()
2434    stream = AirPlayStream(player)
2435    process = MagicMock()
2436    stream._cli_proc = process
2437
2438    async def _stderr_lines() -> AsyncGenerator[str]:
2439        yield "[STATUS] eof"
2440
2441    with (
2442        patch.object(process, "iter_stderr", return_value=_stderr_lines()),
2443        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
2444    ):
2445        await stream._stderr_reader()
2446
2447    assert stream._stopped is True
2448    remove_pipe.assert_awaited_once()
2449    player.schedule_group_rejoin.assert_not_called()
2450
2451
2452async def _run_unexpected_process_death(player: MagicMock) -> AirPlayStream:
2453    """Drive the stderr reader through an unexpected process exit."""
2454    stream = AirPlayStream(player)
2455    process = MagicMock()
2456    stream._cli_proc = process
2457
2458    async def _stderr_lines() -> AsyncGenerator[str]:
2459        yield "some final log line"
2460
2461    with (
2462        patch.object(process, "iter_stderr", return_value=_stderr_lines()),
2463        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2464    ):
2465        await stream._stderr_reader()
2466    return stream
2467
2468
2469@pytest.mark.asyncio
2470async def test_unexpected_death_of_synced_child_schedules_rejoin() -> None:
2471    """A grouped member whose process dies unexpectedly gets a re-join scheduled."""
2472    player = _make_player()
2473    player.synced_to = "leader"
2474    player.group_members = []
2475    # the leader's other members are captured as fallback candidates in case
2476    # leadership transfers while the re-join backoff runs
2477    leader = MagicMock()
2478    leader.group_members = ["leader", player.player_id, "sibling"]
2479    player.provider.mass.players.get_player.return_value = leader
2480
2481    stream = await _run_unexpected_process_death(player)
2482
2483    player.schedule_group_rejoin.assert_called_once_with(["leader", "sibling"])
2484    player.set_state_from_stream.assert_called_once_with(
2485        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2486    )
2487
2488
2489@pytest.mark.asyncio
2490async def test_unexpected_death_of_leader_schedules_rejoin_to_members() -> None:
2491    """A dying leader re-joins towards its surviving members (leadership transfers)."""
2492    player = _make_player()
2493    player.synced_to = None
2494    player.group_members = [player.player_id, "child1", "child2"]
2495
2496    await _run_unexpected_process_death(player)
2497
2498    player.schedule_group_rejoin.assert_called_once_with(["child1", "child2"])
2499    # the controller sets the leader's final state (transfer or dissolve)
2500    player.set_state_from_stream.assert_not_called()
2501
2502
2503@pytest.mark.asyncio
2504async def test_unexpected_death_of_solo_player_schedules_no_rejoin() -> None:
2505    """An ungrouped player's process death only marks the player idle."""
2506    player = _make_player()
2507    player.synced_to = None
2508    player.group_members = []
2509
2510    stream = await _run_unexpected_process_death(player)
2511
2512    player.schedule_group_rejoin.assert_not_called()
2513    player.set_state_from_stream.assert_called_once_with(
2514        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2515    )
2516
2517
2518@pytest.mark.asyncio
2519async def test_unexpected_death_of_static_group_member_drops_member_only() -> None:
2520    """A static group member's death drops just that member, never the whole group."""
2521    player = _make_player()
2522    player.synced_to = "leader"
2523    player.group_members = []
2524    # the player is a static member of an actively playing group player, for
2525    # which cmd_ungroup would release (stop) the WHOLE group
2526    player.state.active_group = "syncgroup1"
2527    group_player = MagicMock()
2528    group_player.static_group_members = ["leader", player.player_id]
2529    leader = MagicMock()
2530    leader.group_members = ["leader", player.player_id]
2531    player.provider.mass.players.get_player.side_effect = lambda player_id: {
2532        "syncgroup1": group_player,
2533        "leader": leader,
2534    }.get(player_id)
2535
2536    await _run_unexpected_process_death(player)
2537
2538    players_controller = player.provider.mass.players
2539    players_controller.cmd_set_members.assert_called_once_with(
2540        "leader", player_ids_to_remove=[player.player_id]
2541    )
2542    players_controller.cmd_ungroup.assert_not_called()
2543    player.schedule_group_rejoin.assert_called_once_with(["leader"])
2544
2545
2546@pytest.mark.asyncio
2547async def test_process_eof_during_render_does_not_send_artwork() -> None:
2548    """A cache lookup finishing after EOF cannot send stale artwork."""
2549    player = _make_player()
2550    stream = AirPlayStream(player)
2551    render_started = asyncio.Event()
2552    release_render = asyncio.Event()
2553
2554    async def _get_image_thumb_path(*_args: Any, **_kwargs: Any) -> str:
2555        render_started.set()
2556        await release_render.wait()
2557        return "/cache/thumbnails/artwork.jpg"
2558
2559    with (
2560        patch(
2561            "music_assistant.providers.airplay.stream.get_image_thumb_path",
2562            new_callable=AsyncMock,
2563            side_effect=_get_image_thumb_path,
2564        ),
2565        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2566    ):
2567        render_task = asyncio.create_task(stream._render_and_send_artwork("image", 1))
2568        await render_started.wait()
2569        stream._stopped = True
2570        release_render.set()
2571        await render_task
2572
2573    send_command.assert_not_awaited()
2574
2575
2576@pytest.mark.asyncio
2577async def test_concurrent_metadata_updates_only_send_latest_artwork() -> None:
2578    """An older slow artwork render cannot overwrite a newer track update."""
2579    player = _make_player()
2580    stream = AirPlayStream(player)
2581    process = MagicMock()
2582    process.closed = False
2583    stream._cli_proc = process
2584    first_render_started = asyncio.Event()
2585    release_first_render = asyncio.Event()
2586
2587    old_metadata = MagicMock(
2588        title="Old track",
2589        artist="Artist",
2590        album="Album",
2591        duration=180,
2592        image_url="old-image",
2593    )
2594    new_metadata = MagicMock(
2595        title="New track",
2596        artist="Artist",
2597        album="Album",
2598        duration=180,
2599        image_url="new-image",
2600    )
2601
2602    async def _prepare_artwork(image_url: str, _generation: int) -> str:
2603        if image_url == "old-image":
2604            first_render_started.set()
2605            await release_first_render.wait()
2606            return "old.jpg"
2607        return "new.jpg"
2608
2609    with (
2610        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2611        patch.object(
2612            stream,
2613            "_prepare_artwork",
2614            new_callable=AsyncMock,
2615            side_effect=_prepare_artwork,
2616        ),
2617        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
2618    ):
2619        old_task = asyncio.create_task(stream.send_metadata(0, old_metadata))
2620        await first_render_started.wait()
2621        new_task = asyncio.create_task(stream.send_metadata(0, new_metadata))
2622        # the new update supersedes the old render once the old push's render
2623        # budget lapses and the metadata lock is released
2624        await new_task
2625        assert stream._metadata_generation == 2
2626        release_first_render.set()
2627        await old_task
2628
2629    commands = [call.args[0].decode() for call in write_command.await_args_list]
2630    assert any("TITLE=New track" in command for command in commands)
2631    assert not any("ARTWORK=old.jpg" in command for command in commands)
2632    assert "ARTWORKFILE=new.jpg\n" in commands[-1]
2633    assert commands[-1].endswith("ACTION=SENDMETA\n")
2634
2635
2636@pytest.mark.asyncio
2637async def test_artwork_url_form_change_does_not_resend_artwork() -> None:
2638    """Alternating URL forms of the same imageproxy image send artwork only once."""
2639    player = _make_player()
2640    stream = AirPlayStream(player)
2641    stream._cli_proc = _make_cli_proc()
2642
2643    def make_metadata(image_url: str) -> MagicMock:
2644        return MagicMock(
2645            corrected_elapsed_time=0,
2646            queue_item_id="item-1",
2647            title="Track",
2648            artist="Artist",
2649            album="Album",
2650            duration=180,
2651            image_url=image_url,
2652        )
2653
2654    # the queue session builds the image URL on the stream server base, the
2655    # player state on the webserver base - same image id behind both forms
2656    image_id = "ab" * 32
2657    other_image_id = "cd" * 32
2658    session_media = make_metadata(
2659        f"http://192.168.1.5:8097/imageproxy/{image_id}?size=512&fmt=jpeg"
2660    )
2661    state_media = make_metadata(f"http://192.168.1.5:8095/imageproxy/{image_id}?size=512&fmt=png")
2662    other_image_media = make_metadata(
2663        f"http://192.168.1.5:8095/imageproxy/{other_image_id}?size=512&fmt=png"
2664    )
2665
2666    with (
2667        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2668        patch.object(
2669            stream,
2670            "_prepare_artwork",
2671            new_callable=AsyncMock,
2672            return_value="/cache/thumb.jpg",
2673        ) as prepare_artwork,
2674    ):
2675        await stream.send_metadata(None, session_media)
2676        generation_after_first_send = stream._metadata_generation
2677        # the post-START push (session media) and the media-updated push
2678        # (player state) alternate on every seek
2679        await stream.send_metadata(None, state_media)
2680        await stream.send_metadata(None, session_media)
2681        assert stream._metadata_generation == generation_after_first_send
2682        await stream.send_metadata(None, other_image_media)
2683
2684    commands = [call.args[0].decode() for call in write_command.await_args_list]
2685    bundled = [command for command in commands if "ARTWORKFILE=" in command]
2686    resends = [command for command in commands if command.startswith("ARTWORK=")]
2687    assert len(bundled) == 1
2688    assert resends == ["ARTWORK=/cache/thumb.jpg\n"]
2689    assert prepare_artwork.await_count == 2
2690    assert stream._metadata_artwork_checksum == other_image_id
2691
2692
2693@pytest.mark.asyncio
2694async def test_metadata_revert_resends_text_after_superseded_artwork() -> None:
2695    """Reverting while artwork renders restores the previously displayed track text."""
2696    player = _make_player()
2697    stream = AirPlayStream(player)
2698    process = MagicMock()
2699    process.closed = False
2700    stream._cli_proc = process
2701    first_metadata = MagicMock(
2702        title="First track",
2703        artist="Artist",
2704        album="Album",
2705        duration=180,
2706        image_url=None,
2707    )
2708    second_metadata = MagicMock(
2709        title="Second track",
2710        artist="Artist",
2711        album="Album",
2712        duration=180,
2713        image_url="second-image",
2714    )
2715    first_checksum = "First track|Artist|Album|180|None"
2716    stream._metadata_text_checksum = first_checksum
2717    stream._pending_metadata_checksum = first_checksum
2718    artwork_started = asyncio.Event()
2719    release_artwork = asyncio.Event()
2720
2721    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2722        artwork_started.set()
2723        await release_artwork.wait()
2724        return "second.jpg"
2725
2726    with (
2727        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2728        patch.object(
2729            stream,
2730            "_prepare_artwork",
2731            new_callable=AsyncMock,
2732            side_effect=_prepare_artwork,
2733        ),
2734    ):
2735        second_task = asyncio.create_task(stream.send_metadata(0, second_metadata))
2736        await artwork_started.wait()
2737        revert_task = asyncio.create_task(stream.send_metadata(0, first_metadata))
2738        await asyncio.sleep(0)
2739        release_artwork.set()
2740        await asyncio.gather(second_task, revert_task)
2741
2742    metadata_commands = [
2743        call.args[0].decode()
2744        for call in write_command.await_args_list
2745        if "ACTION=SENDMETA" in call.args[0].decode()
2746    ]
2747    assert "TITLE=Second track" in metadata_commands[0]
2748    assert "TITLE=First track" in metadata_commands[-1]
2749
2750
2751@pytest.mark.asyncio
2752async def test_repeated_metadata_retries_superseded_artwork() -> None:
2753    """A B-to-C-to-B update sequence still applies B artwork after supersession."""
2754    player = _make_player()
2755    stream = AirPlayStream(player)
2756    process = MagicMock()
2757    process.closed = False
2758    stream._cli_proc = process
2759    initial_text_checksum = "item-initial|Initial|Artist|Album"
2760    initial_checksum = f"{initial_text_checksum}|initial-image"
2761    stream._metadata_artwork_checksum = "initial-image"
2762    stream._metadata_text_checksum = initial_text_checksum
2763    stream._pending_metadata_checksum = initial_checksum
2764    metadata_b = MagicMock(
2765        queue_item_id="item-b",
2766        title="Track B",
2767        artist="Artist",
2768        album="Album",
2769        duration=180,
2770        image_url="b-image",
2771    )
2772    metadata_c = MagicMock(
2773        queue_item_id="item-c",
2774        title="Track C",
2775        artist="Artist",
2776        album="Album",
2777        duration=180,
2778        image_url="c-image",
2779    )
2780    first_artwork_started = asyncio.Event()
2781    release_first_artwork = asyncio.Event()
2782    c_artwork_started = asyncio.Event()
2783    release_c_artwork = asyncio.Event()
2784    b_render_count = 0
2785
2786    async def _prepare_artwork(image_url: str, _generation: int) -> str:
2787        nonlocal b_render_count
2788        if image_url == "b-image":
2789            b_render_count += 1
2790            if b_render_count == 1:
2791                first_artwork_started.set()
2792                await release_first_artwork.wait()
2793                return "b-stale.jpg"
2794            return "b-final.jpg"
2795        c_artwork_started.set()
2796        await release_c_artwork.wait()
2797        return "c.jpg"
2798
2799    with (
2800        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2801        patch.object(
2802            stream,
2803            "_prepare_artwork",
2804            new_callable=AsyncMock,
2805            side_effect=_prepare_artwork,
2806        ) as prepare_artwork,
2807        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
2808    ):
2809        first_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
2810        await first_artwork_started.wait()
2811        c_task = asyncio.create_task(stream.send_metadata(0, metadata_c))
2812        await c_artwork_started.wait()
2813        final_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
2814        await final_b_task
2815        release_first_artwork.set()
2816        release_c_artwork.set()
2817        await asyncio.gather(first_b_task, c_task)
2818
2819    rendered_images = [args.args[0] for args in prepare_artwork.await_args_list]
2820    commands = [args.args[0].decode() for args in write_command.await_args_list]
2821    assert rendered_images == ["b-image", "c-image", "b-image"]
2822    assert "ARTWORK=b-stale.jpg\n" not in commands
2823    assert "ARTWORK=c.jpg\n" not in commands
2824    # the final B render completed within the budget, so it rides the bundle
2825    assert "ARTWORKFILE=b-final.jpg\n" in commands[-1]
2826    assert commands[-1].endswith("ACTION=SENDMETA\n")
2827    assert stream._metadata_artwork_checksum == "b-image"
2828
2829
2830@pytest.mark.asyncio
2831async def test_send_metadata_passes_cached_artwork_path_to_binary() -> None:
2832    """The staged artwork carries the absolute cache path returned by preparation."""
2833    player = _make_player()
2834    stream = AirPlayStream(player)
2835    metadata = MagicMock(
2836        duration=180,
2837        title="Track",
2838        artist="Artist",
2839        album="Album",
2840        image_url="https://example.com/artwork.png",
2841    )
2842    cached_path = "/cache/thumbnails/artwork_flat.jpg"
2843    send_command = AsyncMock()
2844
2845    with (
2846        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value=cached_path)),
2847        patch.object(stream, "send_cli_command", new=send_command),
2848    ):
2849        await stream.send_metadata(None, metadata)
2850
2851    assert f"ARTWORKFILE={cached_path}\n" in send_command.await_args_list[-1].args[0]
2852
2853
2854@pytest.mark.asyncio
2855async def test_track_change_bundles_ready_artwork_into_a_single_push() -> None:
2856    """A track change whose artwork renders within budget lands as ONE bundled write."""
2857    stream = AirPlayStream(_make_player())
2858    stream._cli_proc = _make_cli_proc()
2859    metadata = MagicMock(
2860        queue_item_id="item-1",
2861        duration=180,
2862        title="Track",
2863        artist="Artist",
2864        album="Album",
2865        image_url="image",
2866    )
2867
2868    with (
2869        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2870        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
2871    ):
2872        await stream.send_metadata(0, metadata)
2873
2874    assert write_command.await_count == 1
2875    lines = write_command.await_args_list[0].args[0].decode().splitlines()
2876    assert "TITLE=Track" in lines
2877    assert "ITEMID=item-1" in lines
2878    # the artwork is staged before the SENDMETA applies the whole bundle
2879    assert lines[-2:] == ["ARTWORKFILE=/cache/art.jpg", "ACTION=SENDMETA"]
2880    # the push resets the device position to zero: no PROGRESS correction
2881    assert stream._last_progress_sent == 0
2882    assert stream._metadata_artwork_checksum == "image"
2883
2884
2885@pytest.mark.asyncio
2886async def test_track_change_artwork_missing_the_budget_follows_as_artwork_command() -> None:
2887    """A render missing the bundling budget still delivers via ARTWORK once it completes."""
2888    stream = AirPlayStream(_make_player())
2889    stream._cli_proc = _make_cli_proc()
2890    metadata = MagicMock(
2891        queue_item_id="item-1",
2892        duration=180,
2893        title="Track",
2894        artist="Artist",
2895        album="Album",
2896        image_url="image",
2897    )
2898    release_render = asyncio.Event()
2899
2900    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2901        await release_render.wait()
2902        return "/cache/late.jpg"
2903
2904    with (
2905        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2906        patch.object(
2907            stream, "_prepare_artwork", new_callable=AsyncMock, side_effect=_prepare_artwork
2908        ),
2909        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.01),
2910    ):
2911        push = asyncio.create_task(stream.send_metadata(0, metadata))
2912        async with asyncio.timeout(2):
2913            while write_command.await_count == 0:
2914                await asyncio.sleep(0)
2915        # the identity bundle went out without artwork once the budget lapsed
2916        assert stream._metadata_artwork_checksum == ""
2917        release_render.set()
2918        await push
2919
2920    commands = [args.args[0].decode() for args in write_command.await_args_list]
2921    assert "ARTWORKFILE" not in commands[0]
2922    assert commands[0].endswith("ACTION=SENDMETA\n")
2923    assert [command for command in commands if command.startswith("ARTWORK=")] == [
2924        "ARTWORK=/cache/late.jpg\n"
2925    ]
2926    assert stream._metadata_artwork_checksum == "image"
2927
2928
2929@pytest.mark.asyncio
2930async def test_pending_start_interrupts_the_artwork_wait() -> None:
2931    """A pending START releases the bounded artwork wait instead of queueing behind it."""
2932    player = _make_player()
2933    stream = AirPlayStream(player)
2934    stream._cli_proc = _make_cli_proc()
2935    stream._connected.set()
2936    metadata = MagicMock(
2937        queue_item_id="item-1",
2938        duration=180,
2939        title="Track",
2940        artist="Artist",
2941        album="Album",
2942        image_url="image",
2943    )
2944    release_render = asyncio.Event()
2945    render_started = asyncio.Event()
2946
2947    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2948        render_started.set()
2949        await release_render.wait()
2950        return "/cache/late.jpg"
2951
2952    with (
2953        patch.object(
2954            stream,
2955            "_write_cli_command",
2956            new_callable=AsyncMock,
2957            side_effect=_acking_write_cli_command(stream),
2958        ) as write_command,
2959        patch.object(
2960            stream, "_prepare_artwork", new_callable=AsyncMock, side_effect=_prepare_artwork
2961        ),
2962    ):
2963        push = asyncio.create_task(stream.send_metadata(0, metadata))
2964        await render_started.wait()
2965        # the metadata push sits in its render budget holding the lock; the
2966        # START must release that wait instead of losing its anchor lead to it
2967        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
2968        release_render.set()
2969        await push
2970
2971    commands = [args.args[0] for args in write_command.await_args_list]
2972    assert commands[0].endswith("ACTION=SENDMETA\n")
2973    assert "ARTWORKFILE" not in commands[0]
2974    assert commands[1].startswith(f"START_UNIX_MS={START_UNIX_MS}")
2975
2976
2977@pytest.mark.asyncio
2978async def test_track_change_starting_mid_track_sends_a_progress_correction() -> None:
2979    """A track change landing mid-position corrects the timeline after the bundle."""
2980    stream = AirPlayStream(_make_player())
2981    stream._cli_proc = _make_cli_proc()
2982    metadata = MagicMock(
2983        queue_item_id="item-1",
2984        duration=180,
2985        title="Track",
2986        artist="Artist",
2987        album="Album",
2988        image_url="image",
2989    )
2990
2991    with (
2992        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2993        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
2994    ):
2995        await stream.send_metadata(120, metadata)
2996
2997    commands = [args.args[0].decode() for args in write_command.await_args_list]
2998    assert len(commands) == 2
2999    assert commands[0].endswith("ACTION=SENDMETA\n")
3000    # the push reset the device position to zero, so the mid-track start is
3001    # corrected right after
3002    assert commands[1].endswith("PROGRESS=120\n")
3003    assert stream._last_progress_sent == 120
3004
3005
3006@pytest.mark.asyncio
3007async def test_failed_artwork_delivery_is_retried() -> None:
3008    """A dropped ARTWORK command remains pending for the next metadata update."""
3009    stream = AirPlayStream(_make_player())
3010    metadata = MagicMock(
3011        queue_item_id="item-1",
3012        duration=180,
3013        title="Track",
3014        artist="Artist",
3015        album="Album",
3016        image_url="image",
3017    )
3018    artwork_path = "/cache/thumbnails/artwork.jpg"
3019    release_render = asyncio.Event()
3020
3021    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3022        # the first render misses the bundling budget, so the artwork goes
3023        # out through the stand-alone ARTWORK command
3024        if not release_render.is_set():
3025            await release_render.wait()
3026        return artwork_path
3027
3028    with (
3029        patch.object(
3030            stream,
3031            "_prepare_artwork",
3032            new_callable=AsyncMock,
3033            side_effect=_prepare_artwork,
3034        ) as prepare_artwork,
3035        patch.object(
3036            stream,
3037            "send_cli_command",
3038            new_callable=AsyncMock,
3039            side_effect=[True, False, True],
3040        ) as send_command,
3041        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.01),
3042    ):
3043        first_push = asyncio.create_task(stream.send_metadata(None, metadata))
3044        async with asyncio.timeout(2):
3045            while send_command.await_count == 0:
3046                await asyncio.sleep(0)
3047        release_render.set()
3048        await first_push
3049        assert stream._metadata_artwork_checksum == ""
3050        await stream.send_metadata(None, metadata)
3051
3052    assert prepare_artwork.await_count == 2
3053    assert [args.args[0] for args in send_command.await_args_list].count(
3054        f"ARTWORK={artwork_path}"
3055    ) == 2
3056    assert stream._metadata_artwork_checksum == "image"
3057
3058
3059@pytest.mark.asyncio
3060async def test_text_refinement_keeps_delivered_artwork_settled() -> None:
3061    """A text-only metadata update after delivery does not re-render unchanged art."""
3062    stream = AirPlayStream(_make_player())
3063    metadata = MagicMock(
3064        duration=180,
3065        title="Track",
3066        artist="Artist",
3067        album="Album",
3068        image_url="image",
3069    )
3070    refined = MagicMock(
3071        queue_item_id=metadata.queue_item_id,
3072        duration=180,
3073        title="Track (Remastered)",
3074        artist="Artist",
3075        album="Album",
3076        image_url="image",
3077    )
3078    artwork_path = "/cache/thumbnails/artwork.jpg"
3079
3080    with (
3081        patch.object(
3082            stream,
3083            "_prepare_artwork",
3084            new_callable=AsyncMock,
3085            return_value=artwork_path,
3086        ) as prepare_artwork,
3087        patch.object(
3088            stream,
3089            "send_cli_command",
3090            new_callable=AsyncMock,
3091            return_value=True,
3092        ) as send_command,
3093    ):
3094        await stream.send_metadata(None, metadata)
3095        assert stream._metadata_artwork_checksum == "image"
3096        # the refinement bumps the metadata generation (pending identity
3097        # changed), which before the identity settle re-armed the artwork
3098        await stream.send_metadata(None, refined)
3099
3100    prepare_artwork.assert_awaited_once()
3101    commands = [args.args[0] for args in send_command.await_args_list]
3102    assert sum(f"ARTWORKFILE={artwork_path}\n" in command for command in commands) == 1
3103    assert "ARTWORKFILE" not in commands[-1]
3104    assert "TITLE=Track (Remastered)" in commands[-1]
3105
3106
3107# --- Structured connect failures reported by the binary ---
3108
3109
3110@pytest.mark.asyncio
3111async def test_connect_error_status_line_is_parsed() -> None:
3112    """The machine-readable failure line is captured with all of its fields."""
3113    stream = AirPlayStream(_make_player())
3114
3115    stream._handle_status_line(
3116        '[STATUS] error code=auth_required http=401 detail="RTSP setup rejected"'
3117    )
3118
3119    assert stream._connect_error == CliError("auth_required", 401, "RTSP setup rejected")
3120
3121
3122@pytest.mark.asyncio
3123async def test_started_ack_status_line_parsing() -> None:
3124    """A started ack releases the START wait; a malformed one carries no details."""
3125    stream = AirPlayStream(_make_player())
3126
3127    stream._handle_status_line(
3128        "[STATUS] started requested_unix_ms=1750000000000 at_unix_ms=1750000000004"
3129    )
3130    assert stream._started.is_set()
3131    assert stream._start_ack == (1750000000000, 1750000000004)
3132
3133    stream._started.clear()
3134    stream._start_ack = None
3135    stream._handle_status_line("[STATUS] started requested_unix_ms=garbage at_unix_ms=1")
3136    assert stream._started.is_set()
3137    assert stream._start_ack is None
3138
3139
3140# --- Post-commit anchor verification ---
3141
3142
3143def test_anchor_corrected_status_line_rebases_the_position() -> None:
3144    """A correction carrying a content cut moves the reported-position base by it."""
3145    stream = AirPlayStream(_make_player())
3146    stream._start_position = 12.0
3147
3148    ended = stream._handle_status_line(
3149        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3150        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3151    )
3152
3153    assert ended is False
3154    assert stream._start_position == 12.5
3155
3156
3157def test_anchor_corrected_status_line_logs_a_warning(caplog: pytest.LogCaptureFixture) -> None:
3158    """The correction is logged loudly, including the display name and the delta."""
3159    stream = AirPlayStream(_make_player())
3160
3161    with caplog.at_level(logging.WARNING):
3162        stream._handle_status_line(
3163            "[STATUS] anchor_corrected requested_unix_ms=0 "
3164            "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3165        )
3166
3167    assert "Player A" in caplog.text
3168    assert "+500 ms" in caplog.text
3169
3170
3171def test_anchor_corrected_status_line_tolerates_malformed_line() -> None:
3172    """A malformed anchor_corrected line is dropped instead of raising or rebasing."""
3173    stream = AirPlayStream(_make_player())
3174    stream._start_position = 12.0
3175
3176    ended = stream._handle_status_line("[STATUS] anchor_corrected requested_unix_ms=garbage")
3177
3178    assert ended is False
3179    assert stream._start_position == 12.0
3180
3181
3182def test_content_cut_short_rebases_the_position_and_warns(
3183    caplog: pytest.LogCaptureFixture,
3184) -> None:
3185    """A cut that ended early gives back the ms the correction over-advanced the base by."""
3186    stream = AirPlayStream(_make_player())
3187    stream._start_position = 12.0
3188    stream._handle_status_line(
3189        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3190        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3191    )
3192    assert stream._start_position == 12.5
3193
3194    with caplog.at_level(logging.WARNING):
3195        ended = stream._handle_status_line(
3196            "[STATUS] content_cut requested_ms=500 cut_ms=180 cut_bytes=31752 drain_ms=210"
3197        )
3198
3199    assert ended is False
3200    assert stream._start_position == pytest.approx(12.18)
3201    assert "AirPlay content cut" in caplog.text
3202    assert "Player A" in caplog.text
3203    assert "320 ms short" in caplog.text
3204
3205
3206def test_content_cut_in_full_leaves_the_position_alone(caplog: pytest.LogCaptureFixture) -> None:
3207    """A cut that took what it asked for needs no correction and stays quiet."""
3208    stream = AirPlayStream(_make_player())
3209    stream._start_position = 12.0
3210    stream._handle_status_line(
3211        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3212        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3213    )
3214
3215    caplog.clear()
3216    with caplog.at_level(logging.WARNING):
3217        # a few ms below the request is byte quantization, not a short cut
3218        stream._handle_status_line(
3219            "[STATUS] content_cut requested_ms=500 cut_ms=498 cut_bytes=87887 drain_ms=505"
3220        )
3221
3222    assert stream._start_position == 12.5
3223    assert caplog.text == ""
3224
3225
3226def test_content_cut_after_a_new_anchor_is_not_reconciled() -> None:
3227    """A cut settling after a START must not be taken off that START's absolute base."""
3228    stream = AirPlayStream(_make_player())
3229    stream._start_position = 12.0
3230    stream._handle_status_line(
3231        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3232        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3233    )
3234    stream.rebase_position(30_000)
3235
3236    stream._handle_status_line(
3237        "[STATUS] content_cut requested_ms=500 cut_ms=0 cut_bytes=0 drain_ms=12"
3238    )
3239
3240    assert stream._start_position == 30.0
3241
3242
3243def test_content_cut_status_line_tolerates_malformed_line() -> None:
3244    """A malformed content_cut line is dropped instead of raising or rebasing."""
3245    stream = AirPlayStream(_make_player())
3246    stream._start_position = 12.0
3247    stream._handle_status_line(
3248        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3249        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3250    )
3251
3252    ended = stream._handle_status_line("[STATUS] content_cut requested_ms=500 cut_ms=garbage")
3253
3254    assert ended is False
3255    assert stream._start_position == 12.5
3256
3257
3258def test_clock_verified_status_line_is_debug_logged(caplog: pytest.LogCaptureFixture) -> None:
3259    """A clock_verified line needs no server action beyond a debug note of the margin."""
3260    stream = AirPlayStream(_make_player())
3261
3262    with caplog.at_level(logging.DEBUG):
3263        ended = stream._handle_status_line("[STATUS] clock_verified margin_ms=42")
3264
3265    assert ended is False
3266    assert "42" in caplog.text
3267
3268
3269@pytest.mark.asyncio
3270async def test_connect_error_status_line_tolerates_missing_fields() -> None:
3271    """A failure line without http/detail still yields the reported code."""
3272    stream = AirPlayStream(_make_player())
3273
3274    stream._handle_status_line("[STATUS] error code=connect_failed")
3275
3276    assert stream._connect_error == CliError("connect_failed", 0, "")
3277
3278
3279@pytest.mark.asyncio
3280async def test_auth_required_surfaces_password_required_error() -> None:
3281    """A device asking for a password produces an actionable, translated error."""
3282    stream = AirPlayStream(_make_player())
3283    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
3284    stream._process_ended.set()
3285
3286    with (
3287        patch.object(stream, "_cli_proc", MagicMock()),
3288        pytest.raises(PlayerCommandFailed) as err,
3289    ):
3290        await stream.wait_for_connection()
3291
3292    assert err.value.translation_key == "password_required"
3293
3294
3295@pytest.mark.asyncio
3296async def test_auth_failed_surfaces_authentication_failed_error() -> None:
3297    """A rejected password is reported as an authentication failure, not a timeout."""
3298    stream = AirPlayStream(_make_player())
3299    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
3300    stream._process_ended.set()
3301
3302    with (
3303        patch.object(stream, "_cli_proc", MagicMock()),
3304        pytest.raises(PlayerCommandFailed) as err,
3305    ):
3306        await stream.wait_for_connection()
3307
3308    assert err.value.translation_key == "authentication_failed"
3309
3310
3311@pytest.mark.asyncio
3312async def test_generic_connect_failure_keeps_the_timeout_semantics() -> None:
3313    """A non-auth failure keeps raising the plain timeout its callers already handle."""
3314    stream = AirPlayStream(_make_player())
3315    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
3316    stream._process_ended.set()
3317
3318    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
3319        await stream.wait_for_connection()
3320
3321
3322@pytest.mark.asyncio
3323async def test_dead_process_fails_the_connect_wait_immediately() -> None:
3324    """
3325    A binary that reports no reason at all leaves the wait its plain timeout error.
3326
3327    It must still end the moment the process is gone instead of running out the
3328    full connect timeout.
3329    """
3330    stream = AirPlayStream(_make_player())
3331    stream._process_ended.set()  # process died without emitting a [STATUS] error line
3332
3333    started = asyncio.get_running_loop().time()
3334    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
3335        await stream.wait_for_connection()
3336
3337    assert asyncio.get_running_loop().time() - started < 1
3338
3339
3340@pytest.mark.asyncio
3341async def test_auth_failed_marks_the_stored_password_invalid() -> None:
3342    """A rejected password is persisted so the player keeps offering its setup action."""
3343    player = _make_player()
3344    stream = AirPlayStream(player)
3345
3346    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
3347
3348    player.set_password_invalid.assert_called_once_with(True)
3349
3350
3351@pytest.mark.asyncio
3352async def test_auth_required_also_marks_the_password_as_needed() -> None:
3353    """
3354    A device that demanded a password we could not supply flips into setup.
3355
3356    Devices can enforce a password without announcing it (stale TXT records), so
3357    the runtime signal must set the marker too - it is the only reliable one.
3358    """
3359    player = _make_player()
3360    stream = AirPlayStream(player)
3361
3362    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
3363
3364    player.set_password_invalid.assert_called_once_with(True)
3365
3366
3367@pytest.mark.asyncio
3368async def test_plain_connect_failures_leave_the_password_marker_alone() -> None:
3369    """A non-authentication failure says nothing about the stored password."""
3370    player = _make_player()
3371    stream = AirPlayStream(player)
3372
3373    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
3374
3375    player.set_password_invalid.assert_not_called()
3376
3377
3378@pytest.mark.asyncio
3379async def test_successful_connect_clears_the_password_marker() -> None:
3380    """Whatever the device accepted is a working password."""
3381    player = _make_player()
3382    stream = AirPlayStream(player)
3383
3384    stream._handle_status_line("[STATUS] connected")
3385
3386    assert stream.connected is True
3387    player.set_password_invalid.assert_called_once_with(False)
3388
3389
3390# --- Password preflight ---
3391
3392
3393@pytest.mark.asyncio
3394async def test_connect_refuses_password_device_without_password_or_credentials() -> None:
3395    """A password-protected AirPlay 2 device with nothing to authenticate never spawns a process."""
3396    player = _make_player()
3397    player.password_required = True
3398    player.get_setup_value = MagicMock(return_value=None)
3399    stream = AirPlayStream(player)
3400
3401    with pytest.raises(PlayerCommandFailed) as err:
3402        await stream.connect()
3403
3404    assert err.value.translation_key == "password_required"
3405    assert stream._cli_proc is None
3406
3407
3408@pytest.mark.asyncio
3409async def test_password_preflight_passes_with_password_or_credentials() -> None:
3410    """Either a configured password or stored credentials let the connect proceed."""
3411    player = _make_player()
3412    player.password_required = True
3413    player.get_setup_value = MagicMock(return_value=None)
3414    player.config.get_value = MagicMock(
3415        side_effect=lambda key, default=None: "s3cret" if key == CONF_PASSWORD else default
3416    )
3417    AirPlayStream(player)._check_password_preflight()
3418
3419    # credentials alone are enough: the binary's pair-verify leg may still succeed
3420    player.config.get_value = MagicMock(side_effect=lambda _key, default=None: default)
3421    player.get_setup_value = MagicMock(
3422        side_effect=lambda key, default=None: (
3423            "ab" * 96 if key == CONF_AIRPLAY_CREDENTIALS else default
3424        )
3425    )
3426    AirPlayStream(player)._check_password_preflight()
3427
3428
3429@pytest.mark.asyncio
3430async def test_password_preflight_skipped_for_raop() -> None:
3431    """The preflight only guards the native AirPlay 2 flow; RAOP carries its own password."""
3432    player = _make_player()
3433    player.protocol = StreamingProtocol.RAOP
3434    player.password_required = True
3435    player.get_setup_value = MagicMock(return_value=None)
3436
3437    AirPlayStream(player)._check_password_preflight()
3438