music-assistant-server

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