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