music-assistant-server

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