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