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