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