music-assistant-server

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