music-assistant-server

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