music-assistant-server

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