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    # the member is dropped from its native sync leader directly: cmd_ungroup
2836    # would resolve a linked protocol player to its visible parent and act at
2837    # the group level, removing the member from its (sync)group over one dead
2838    # transport
2839    players_controller = player.provider.mass.players
2840    players_controller.cmd_set_members.assert_called_once_with(
2841        "leader", player_ids_to_remove=[player.player_id]
2842    )
2843    players_controller.cmd_ungroup.assert_not_called()
2844    player.schedule_group_rejoin.assert_called_once_with(["leader", "sibling"])
2845    player.set_state_from_stream.assert_called_once_with(
2846        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2847    )
2848
2849
2850@pytest.mark.asyncio
2851async def test_unexpected_death_of_leader_schedules_rejoin_to_members() -> None:
2852    """A dying leader re-joins towards its surviving members (leadership transfers)."""
2853    player = _make_player()
2854    player.synced_to = None
2855    player.group_members = [player.player_id, "child1", "child2"]
2856
2857    await _run_unexpected_process_death(player)
2858
2859    player.schedule_group_rejoin.assert_called_once_with(["child1", "child2"])
2860    # the controller sets the leader's final state (transfer or dissolve)
2861    player.set_state_from_stream.assert_not_called()
2862
2863
2864@pytest.mark.asyncio
2865async def test_unexpected_death_of_solo_player_schedules_no_rejoin() -> None:
2866    """An ungrouped player's process death only marks the player idle."""
2867    player = _make_player()
2868    player.synced_to = None
2869    player.group_members = []
2870
2871    stream = await _run_unexpected_process_death(player)
2872
2873    player.schedule_group_rejoin.assert_not_called()
2874    player.set_state_from_stream.assert_called_once_with(
2875        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2876    )
2877
2878
2879@pytest.mark.asyncio
2880async def test_unexpected_death_of_static_group_member_drops_member_only() -> None:
2881    """A static group member's death drops just that member, never the whole group."""
2882    player = _make_player()
2883    player.synced_to = "leader"
2884    player.group_members = []
2885    # the player is a static member of an actively playing group player, for
2886    # which cmd_ungroup would release (stop) the WHOLE group
2887    player.state.active_group = "syncgroup1"
2888    group_player = MagicMock()
2889    group_player.static_group_members = ["leader", player.player_id]
2890    leader = MagicMock()
2891    leader.group_members = ["leader", player.player_id]
2892    player.provider.mass.players.get_player.side_effect = lambda player_id: {
2893        "syncgroup1": group_player,
2894        "leader": leader,
2895    }.get(player_id)
2896
2897    await _run_unexpected_process_death(player)
2898
2899    players_controller = player.provider.mass.players
2900    players_controller.cmd_set_members.assert_called_once_with(
2901        "leader", player_ids_to_remove=[player.player_id]
2902    )
2903    players_controller.cmd_ungroup.assert_not_called()
2904    player.schedule_group_rejoin.assert_called_once_with(["leader"])
2905
2906
2907@pytest.mark.asyncio
2908async def test_unexpected_death_of_superseded_stream_leaves_the_group_alone() -> None:
2909    """
2910    A superseded process dying does not ungroup the player or schedule a re-join.
2911
2912    Its death is the newer session taking the receiver over, so acting on it
2913    would tear down the healthy session that replaced it.
2914    """
2915    player = _make_player()
2916    player.synced_to = "leader"
2917    player.group_members = []
2918    player.stream = AirPlayStream(player)
2919
2920    stream = await _run_unexpected_process_death(player, still_owned=False)
2921
2922    players_controller = player.provider.mass.players
2923    players_controller.cmd_ungroup.assert_not_called()
2924    players_controller.cmd_set_members.assert_not_called()
2925    player.schedule_group_rejoin.assert_not_called()
2926    player.set_state_from_stream.assert_not_called()
2927    assert stream._stopped is True
2928
2929
2930@pytest.mark.asyncio
2931async def test_process_eof_during_render_does_not_send_artwork() -> None:
2932    """A cache lookup finishing after EOF cannot send stale artwork."""
2933    player = _make_player()
2934    stream = AirPlayStream(player)
2935    render_started = asyncio.Event()
2936    release_render = asyncio.Event()
2937
2938    async def _get_image_thumb_path(*_args: Any, **_kwargs: Any) -> str:
2939        render_started.set()
2940        await release_render.wait()
2941        return "/cache/thumbnails/artwork.jpg"
2942
2943    with (
2944        patch(
2945            "music_assistant.providers.airplay.stream.get_image_thumb_path",
2946            new_callable=AsyncMock,
2947            side_effect=_get_image_thumb_path,
2948        ),
2949        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2950    ):
2951        render_task = asyncio.create_task(stream._render_and_send_artwork("image", 1))
2952        await render_started.wait()
2953        stream._stopped = True
2954        release_render.set()
2955        await render_task
2956
2957    send_command.assert_not_awaited()
2958
2959
2960@pytest.mark.asyncio
2961async def test_concurrent_metadata_updates_only_send_latest_artwork() -> None:
2962    """An older slow artwork render cannot overwrite a newer track update."""
2963    player = _make_player()
2964    stream = AirPlayStream(player)
2965    process = MagicMock()
2966    process.closed = False
2967    stream._cli_proc = process
2968    first_render_started = asyncio.Event()
2969    release_first_render = asyncio.Event()
2970
2971    old_metadata = MagicMock(
2972        title="Old track",
2973        artist="Artist",
2974        album="Album",
2975        duration=180,
2976        image_url="old-image",
2977    )
2978    new_metadata = MagicMock(
2979        title="New track",
2980        artist="Artist",
2981        album="Album",
2982        duration=180,
2983        image_url="new-image",
2984    )
2985
2986    async def _prepare_artwork(image_url: str, _generation: int) -> str:
2987        if image_url == "old-image":
2988            first_render_started.set()
2989            await release_first_render.wait()
2990            return "old.jpg"
2991        return "new.jpg"
2992
2993    with (
2994        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2995        patch.object(
2996            stream,
2997            "_prepare_artwork",
2998            new_callable=AsyncMock,
2999            side_effect=_prepare_artwork,
3000        ),
3001        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
3002    ):
3003        old_task = asyncio.create_task(stream.send_metadata(0, old_metadata))
3004        await first_render_started.wait()
3005        new_task = asyncio.create_task(stream.send_metadata(0, new_metadata))
3006        # the new update supersedes the old render once the old push's render
3007        # budget lapses and the metadata lock is released
3008        await new_task
3009        assert stream._metadata_generation == 2
3010        release_first_render.set()
3011        await old_task
3012
3013    commands = [call.args[0].decode() for call in write_command.await_args_list]
3014    assert any("TITLE=New track" in command for command in commands)
3015    assert not any("ARTWORK=old.jpg" in command for command in commands)
3016    last_bundle = [command for command in commands if command.endswith("ACTION=SENDMETA\n")][-1]
3017    assert "ARTWORKFILE=new.jpg\n" in last_bundle
3018
3019
3020@pytest.mark.asyncio
3021async def test_artwork_url_form_change_does_not_resend_artwork() -> None:
3022    """Alternating URL forms of the same imageproxy image send artwork only once."""
3023    player = _make_player()
3024    stream = AirPlayStream(player)
3025    stream._cli_proc = _make_cli_proc()
3026
3027    def make_metadata(image_url: str) -> MagicMock:
3028        return MagicMock(
3029            corrected_elapsed_time=0,
3030            queue_item_id="item-1",
3031            title="Track",
3032            artist="Artist",
3033            album="Album",
3034            duration=180,
3035            image_url=image_url,
3036        )
3037
3038    # the queue session builds the image URL on the stream server base, the
3039    # player state on the webserver base - same image id behind both forms
3040    image_id = "ab" * 32
3041    other_image_id = "cd" * 32
3042    session_media = make_metadata(
3043        f"http://192.168.1.5:8097/imageproxy/{image_id}?size=512&fmt=jpeg"
3044    )
3045    state_media = make_metadata(f"http://192.168.1.5:8095/imageproxy/{image_id}?size=512&fmt=png")
3046    other_image_media = make_metadata(
3047        f"http://192.168.1.5:8095/imageproxy/{other_image_id}?size=512&fmt=png"
3048    )
3049
3050    with (
3051        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3052        patch.object(
3053            stream,
3054            "_prepare_artwork",
3055            new_callable=AsyncMock,
3056            return_value="/cache/thumb.jpg",
3057        ) as prepare_artwork,
3058    ):
3059        await stream.send_metadata(None, session_media)
3060        generation_after_first_send = stream._metadata_generation
3061        # the post-START push (session media) and the media-updated push
3062        # (player state) alternate on every seek
3063        await stream.send_metadata(None, state_media)
3064        await stream.send_metadata(None, session_media)
3065        assert stream._metadata_generation == generation_after_first_send
3066        await stream.send_metadata(None, other_image_media)
3067
3068    commands = [call.args[0].decode() for call in write_command.await_args_list]
3069    bundled = [command for command in commands if "ARTWORKFILE=" in command]
3070    resends = [command for command in commands if command.startswith("ARTWORK=")]
3071    assert len(bundled) == 1
3072    assert resends == ["ARTWORK=/cache/thumb.jpg\n"]
3073    assert prepare_artwork.await_count == 2
3074    assert stream._metadata_artwork_checksum == other_image_id
3075
3076
3077@pytest.mark.asyncio
3078async def test_metadata_revert_resends_text_after_superseded_artwork() -> None:
3079    """Reverting while artwork renders restores the previously displayed track text."""
3080    player = _make_player()
3081    stream = AirPlayStream(player)
3082    process = MagicMock()
3083    process.closed = False
3084    stream._cli_proc = process
3085    first_metadata = MagicMock(
3086        title="First track",
3087        artist="Artist",
3088        album="Album",
3089        duration=180,
3090        image_url=None,
3091    )
3092    second_metadata = MagicMock(
3093        title="Second track",
3094        artist="Artist",
3095        album="Album",
3096        duration=180,
3097        image_url="second-image",
3098    )
3099    first_checksum = "First track|Artist|Album|180|None"
3100    stream._metadata_text_checksum = first_checksum
3101    stream._pending_metadata_checksum = first_checksum
3102    artwork_started = asyncio.Event()
3103    release_artwork = asyncio.Event()
3104
3105    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3106        artwork_started.set()
3107        await release_artwork.wait()
3108        return "second.jpg"
3109
3110    with (
3111        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3112        patch.object(
3113            stream,
3114            "_prepare_artwork",
3115            new_callable=AsyncMock,
3116            side_effect=_prepare_artwork,
3117        ),
3118    ):
3119        second_task = asyncio.create_task(stream.send_metadata(0, second_metadata))
3120        await artwork_started.wait()
3121        revert_task = asyncio.create_task(stream.send_metadata(0, first_metadata))
3122        await asyncio.sleep(0)
3123        release_artwork.set()
3124        await asyncio.gather(second_task, revert_task)
3125
3126    metadata_commands = [
3127        call.args[0].decode()
3128        for call in write_command.await_args_list
3129        if "ACTION=SENDMETA" in call.args[0].decode()
3130    ]
3131    assert "TITLE=Second track" in metadata_commands[0]
3132    assert "TITLE=First track" in metadata_commands[-1]
3133
3134
3135@pytest.mark.asyncio
3136async def test_repeated_metadata_retries_superseded_artwork() -> None:
3137    """A B-to-C-to-B update sequence still applies B artwork after supersession."""
3138    player = _make_player()
3139    stream = AirPlayStream(player)
3140    process = MagicMock()
3141    process.closed = False
3142    stream._cli_proc = process
3143    initial_text_checksum = "item-initial|Initial|Artist|Album"
3144    initial_checksum = f"{initial_text_checksum}|initial-image"
3145    stream._metadata_artwork_checksum = "initial-image"
3146    stream._metadata_text_checksum = initial_text_checksum
3147    stream._pending_metadata_checksum = initial_checksum
3148    metadata_b = MagicMock(
3149        queue_item_id="item-b",
3150        title="Track B",
3151        artist="Artist",
3152        album="Album",
3153        duration=180,
3154        image_url="b-image",
3155    )
3156    metadata_c = MagicMock(
3157        queue_item_id="item-c",
3158        title="Track C",
3159        artist="Artist",
3160        album="Album",
3161        duration=180,
3162        image_url="c-image",
3163    )
3164    first_artwork_started = asyncio.Event()
3165    release_first_artwork = asyncio.Event()
3166    c_artwork_started = asyncio.Event()
3167    release_c_artwork = asyncio.Event()
3168    b_render_count = 0
3169
3170    async def _prepare_artwork(image_url: str, _generation: int) -> str:
3171        nonlocal b_render_count
3172        if image_url == "b-image":
3173            b_render_count += 1
3174            if b_render_count == 1:
3175                first_artwork_started.set()
3176                await release_first_artwork.wait()
3177                return "b-stale.jpg"
3178            return "b-final.jpg"
3179        c_artwork_started.set()
3180        await release_c_artwork.wait()
3181        return "c.jpg"
3182
3183    with (
3184        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3185        patch.object(
3186            stream,
3187            "_prepare_artwork",
3188            new_callable=AsyncMock,
3189            side_effect=_prepare_artwork,
3190        ) as prepare_artwork,
3191        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
3192    ):
3193        first_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
3194        await first_artwork_started.wait()
3195        c_task = asyncio.create_task(stream.send_metadata(0, metadata_c))
3196        await c_artwork_started.wait()
3197        final_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
3198        await final_b_task
3199        release_first_artwork.set()
3200        release_c_artwork.set()
3201        await asyncio.gather(first_b_task, c_task)
3202
3203    rendered_images = [args.args[0] for args in prepare_artwork.await_args_list]
3204    commands = [args.args[0].decode() for args in write_command.await_args_list]
3205    assert rendered_images == ["b-image", "c-image", "b-image"]
3206    assert "ARTWORK=b-stale.jpg\n" not in commands
3207    assert "ARTWORK=c.jpg\n" not in commands
3208    # the final B render completed within the budget, so it rides the bundle
3209    last_bundle = [command for command in commands if command.endswith("ACTION=SENDMETA\n")][-1]
3210    assert "ARTWORKFILE=b-final.jpg\n" in last_bundle
3211    assert stream._metadata_artwork_checksum == "b-image"
3212
3213
3214@pytest.mark.asyncio
3215async def test_send_metadata_passes_cached_artwork_path_to_binary() -> None:
3216    """The staged artwork carries the absolute cache path returned by preparation."""
3217    player = _make_player()
3218    stream = AirPlayStream(player)
3219    metadata = MagicMock(
3220        duration=180,
3221        title="Track",
3222        artist="Artist",
3223        album="Album",
3224        image_url="https://example.com/artwork.png",
3225    )
3226    cached_path = "/cache/thumbnails/artwork_flat.jpg"
3227    send_command = AsyncMock()
3228
3229    with (
3230        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value=cached_path)),
3231        patch.object(stream, "send_cli_command", new=send_command),
3232    ):
3233        await stream.send_metadata(None, metadata)
3234
3235    assert f"ARTWORKFILE={cached_path}\n" in send_command.await_args_list[-1].args[0]
3236
3237
3238@pytest.mark.asyncio
3239async def test_track_change_bundles_ready_artwork_into_a_single_push() -> None:
3240    """A track change whose artwork renders within budget lands as ONE bundled write."""
3241    stream = AirPlayStream(_make_player())
3242    stream._cli_proc = _make_cli_proc()
3243    metadata = MagicMock(
3244        queue_item_id="item-1",
3245        duration=180,
3246        title="Track",
3247        artist="Artist",
3248        album="Album",
3249        image_url="image",
3250    )
3251
3252    with (
3253        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3254        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
3255    ):
3256        await stream.send_metadata(0, metadata)
3257
3258    assert write_command.await_count == 2
3259    lines = write_command.await_args_list[0].args[0].decode().splitlines()
3260    assert "TITLE=Track" in lines
3261    assert "ITEMID=item-1" in lines
3262    # the artwork is staged before the SENDMETA applies the whole bundle
3263    assert lines[-2:] == ["ARTWORKFILE=/cache/art.jpg", "ACTION=SENDMETA"]
3264    # the bundle is one write; the explicit progress anchor follows separately
3265    assert write_command.await_args_list[1].args[0].decode().endswith("PROGRESS=0\n")
3266    assert stream._last_progress_sent == 0
3267    assert stream._metadata_artwork_checksum == "image"
3268
3269
3270@pytest.mark.asyncio
3271async def test_track_change_artwork_missing_the_budget_follows_as_artwork_command() -> None:
3272    """A render missing the bundling budget still delivers via ARTWORK once it completes."""
3273    stream = AirPlayStream(_make_player())
3274    stream._cli_proc = _make_cli_proc()
3275    metadata = MagicMock(
3276        queue_item_id="item-1",
3277        duration=180,
3278        title="Track",
3279        artist="Artist",
3280        album="Album",
3281        image_url="image",
3282    )
3283    release_render = asyncio.Event()
3284
3285    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3286        await release_render.wait()
3287        return "/cache/late.jpg"
3288
3289    with (
3290        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3291        patch.object(
3292            stream, "_prepare_artwork", new_callable=AsyncMock, side_effect=_prepare_artwork
3293        ),
3294        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.01),
3295    ):
3296        push = asyncio.create_task(stream.send_metadata(0, metadata))
3297        async with asyncio.timeout(2):
3298            while write_command.await_count == 0:
3299                await asyncio.sleep(0)
3300        # the identity bundle went out without artwork once the budget lapsed
3301        assert stream._metadata_artwork_checksum == ""
3302        release_render.set()
3303        await push
3304
3305    commands = [args.args[0].decode() for args in write_command.await_args_list]
3306    assert "ARTWORKFILE" not in commands[0]
3307    assert commands[0].endswith("ACTION=SENDMETA\n")
3308    assert [command for command in commands if command.startswith("ARTWORK=")] == [
3309        "ARTWORK=/cache/late.jpg\n"
3310    ]
3311    assert stream._metadata_artwork_checksum == "image"
3312
3313
3314@pytest.mark.asyncio
3315async def test_pending_start_interrupts_the_artwork_wait() -> None:
3316    """A pending START releases the bounded artwork wait instead of queueing behind it."""
3317    player = _make_player()
3318    stream = AirPlayStream(player)
3319    stream._cli_proc = _make_cli_proc()
3320    stream._connected.set()
3321    metadata = MagicMock(
3322        queue_item_id="item-1",
3323        duration=180,
3324        title="Track",
3325        artist="Artist",
3326        album="Album",
3327        image_url="image",
3328    )
3329    release_render = asyncio.Event()
3330    render_started = asyncio.Event()
3331
3332    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3333        render_started.set()
3334        await release_render.wait()
3335        return "/cache/late.jpg"
3336
3337    with (
3338        patch.object(
3339            stream,
3340            "_write_cli_command",
3341            new_callable=AsyncMock,
3342            side_effect=_acking_write_cli_command(stream),
3343        ) as write_command,
3344        patch.object(
3345            stream, "_prepare_artwork", new_callable=AsyncMock, side_effect=_prepare_artwork
3346        ),
3347    ):
3348        push = asyncio.create_task(stream.send_metadata(0, metadata))
3349        await render_started.wait()
3350        # the metadata push sits in its render budget holding the lock; the
3351        # START must release that wait instead of losing its anchor lead to it
3352        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
3353        release_render.set()
3354        await push
3355
3356    commands = [args.args[0] for args in write_command.await_args_list]
3357    assert commands[0].endswith("ACTION=SENDMETA\n")
3358    assert "ARTWORKFILE" not in commands[0]
3359    # the START may only queue behind the push's quick pipe writes (the
3360    # progress anchor), never behind the artwork render itself
3361    start_index = next(
3362        index
3363        for index, command in enumerate(commands)
3364        if command.startswith(f"START_UNIX_MS={START_UNIX_MS}")
3365    )
3366    assert start_index <= 2
3367    assert not any("late.jpg" in command for command in commands[:start_index])
3368
3369
3370@pytest.mark.asyncio
3371async def test_track_change_starting_mid_track_sends_a_progress_correction() -> None:
3372    """A track change landing mid-position corrects the timeline after the bundle."""
3373    stream = AirPlayStream(_make_player())
3374    stream._cli_proc = _make_cli_proc()
3375    metadata = MagicMock(
3376        queue_item_id="item-1",
3377        duration=180,
3378        title="Track",
3379        artist="Artist",
3380        album="Album",
3381        image_url="image",
3382    )
3383
3384    with (
3385        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3386        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
3387    ):
3388        await stream.send_metadata(120, metadata)
3389
3390    commands = [args.args[0].decode() for args in write_command.await_args_list]
3391    assert len(commands) == 2
3392    assert commands[0].endswith("ACTION=SENDMETA\n")
3393    # the push reset the device position to zero, so the mid-track start is
3394    # corrected right after
3395    assert commands[1].endswith("PROGRESS=120\n")
3396    assert stream._last_progress_sent == 120
3397
3398
3399@pytest.mark.asyncio
3400async def test_track_change_at_position_zero_still_sends_a_progress_anchor() -> None:
3401    """
3402    A track starting at zero still gets an explicit PROGRESS anchor.
3403
3404    Some receivers gate their rendering on an explicit timeline anchor: a WiiM
3405    Amp mutes a flushed-and-restarted session a couple of minutes in when no
3406    PROGRESS ever follows the metadata push, and un-mutes the instant one
3407    arrives. Relying on SENDMETA's implicit reset to zero is not enough.
3408    """
3409    stream = AirPlayStream(_make_player())
3410    stream._cli_proc = _make_cli_proc()
3411    metadata = MagicMock(
3412        queue_item_id="item-1",
3413        duration=180,
3414        title="Track",
3415        artist="Artist",
3416        album="Album",
3417        image_url="image",
3418    )
3419
3420    with (
3421        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3422        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
3423    ):
3424        await stream.send_metadata(0, metadata)
3425
3426    commands = [args.args[0].decode() for args in write_command.await_args_list]
3427    assert len(commands) == 2
3428    assert commands[0].endswith("ACTION=SENDMETA\n")
3429    assert commands[1].endswith("PROGRESS=0\n")
3430    assert stream._last_progress_sent == 0
3431
3432
3433@pytest.mark.asyncio
3434async def test_failed_artwork_delivery_is_retried() -> None:
3435    """A dropped ARTWORK command remains pending for the next metadata update."""
3436    stream = AirPlayStream(_make_player())
3437    metadata = MagicMock(
3438        queue_item_id="item-1",
3439        duration=180,
3440        title="Track",
3441        artist="Artist",
3442        album="Album",
3443        image_url="image",
3444    )
3445    artwork_path = "/cache/thumbnails/artwork.jpg"
3446    release_render = asyncio.Event()
3447
3448    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3449        # the first render misses the bundling budget, so the artwork goes
3450        # out through the stand-alone ARTWORK command
3451        if not release_render.is_set():
3452            await release_render.wait()
3453        return artwork_path
3454
3455    with (
3456        patch.object(
3457            stream,
3458            "_prepare_artwork",
3459            new_callable=AsyncMock,
3460            side_effect=_prepare_artwork,
3461        ) as prepare_artwork,
3462        patch.object(
3463            stream,
3464            "send_cli_command",
3465            new_callable=AsyncMock,
3466            side_effect=[True, False, True],
3467        ) as send_command,
3468        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.01),
3469    ):
3470        first_push = asyncio.create_task(stream.send_metadata(None, metadata))
3471        async with asyncio.timeout(2):
3472            while send_command.await_count == 0:
3473                await asyncio.sleep(0)
3474        release_render.set()
3475        await first_push
3476        assert stream._metadata_artwork_checksum == ""
3477        await stream.send_metadata(None, metadata)
3478
3479    assert prepare_artwork.await_count == 2
3480    assert [args.args[0] for args in send_command.await_args_list].count(
3481        f"ARTWORK={artwork_path}"
3482    ) == 2
3483    assert stream._metadata_artwork_checksum == "image"
3484
3485
3486@pytest.mark.asyncio
3487async def test_text_refinement_keeps_delivered_artwork_settled() -> None:
3488    """A text-only metadata update after delivery does not re-render unchanged art."""
3489    stream = AirPlayStream(_make_player())
3490    metadata = MagicMock(
3491        duration=180,
3492        title="Track",
3493        artist="Artist",
3494        album="Album",
3495        image_url="image",
3496    )
3497    refined = MagicMock(
3498        queue_item_id=metadata.queue_item_id,
3499        duration=180,
3500        title="Track (Remastered)",
3501        artist="Artist",
3502        album="Album",
3503        image_url="image",
3504    )
3505    artwork_path = "/cache/thumbnails/artwork.jpg"
3506
3507    with (
3508        patch.object(
3509            stream,
3510            "_prepare_artwork",
3511            new_callable=AsyncMock,
3512            return_value=artwork_path,
3513        ) as prepare_artwork,
3514        patch.object(
3515            stream,
3516            "send_cli_command",
3517            new_callable=AsyncMock,
3518            return_value=True,
3519        ) as send_command,
3520    ):
3521        await stream.send_metadata(None, metadata)
3522        assert stream._metadata_artwork_checksum == "image"
3523        # the refinement bumps the metadata generation (pending identity
3524        # changed), which before the identity settle re-armed the artwork
3525        await stream.send_metadata(None, refined)
3526
3527    prepare_artwork.assert_awaited_once()
3528    commands = [args.args[0] for args in send_command.await_args_list]
3529    assert sum(f"ARTWORKFILE={artwork_path}\n" in command for command in commands) == 1
3530    assert "ARTWORKFILE" not in commands[-1]
3531    assert "TITLE=Track (Remastered)" in commands[-1]
3532
3533
3534# --- Structured connect failures reported by the binary ---
3535
3536
3537@pytest.mark.asyncio
3538async def test_connect_error_status_line_is_parsed() -> None:
3539    """The machine-readable failure line is captured with all of its fields."""
3540    stream = AirPlayStream(_make_player())
3541
3542    stream._handle_status_line(
3543        '[STATUS] error code=auth_required http=401 detail="RTSP setup rejected"'
3544    )
3545
3546    assert stream._connect_error == CliError("auth_required", 401, "RTSP setup rejected")
3547
3548
3549@pytest.mark.asyncio
3550async def test_started_ack_status_line_parsing() -> None:
3551    """A started ack releases the START wait; a malformed one carries no details."""
3552    stream = AirPlayStream(_make_player())
3553
3554    stream._handle_status_line(
3555        "[STATUS] started requested_unix_ms=1750000000000 at_unix_ms=1750000000004"
3556    )
3557    assert stream._started.is_set()
3558    assert stream._start_ack == (1750000000000, 1750000000004)
3559
3560    stream._started.clear()
3561    stream._start_ack = None
3562    stream._handle_status_line("[STATUS] started requested_unix_ms=garbage at_unix_ms=1")
3563    assert stream._started.is_set()
3564    assert stream._start_ack is None
3565
3566
3567# --- Post-commit anchor verification ---
3568
3569
3570def test_anchor_corrected_status_line_rebases_the_position() -> None:
3571    """A correction carrying a content cut moves the reported-position base by it."""
3572    stream = AirPlayStream(_make_player())
3573    stream._start_position = 12.0
3574
3575    ended = stream._handle_status_line(
3576        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3577        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3578    )
3579
3580    assert ended is False
3581    assert stream._start_position == 12.5
3582
3583
3584def test_anchor_corrected_status_line_logs_a_warning(caplog: pytest.LogCaptureFixture) -> None:
3585    """The correction is logged loudly, including the display name and the delta."""
3586    stream = AirPlayStream(_make_player())
3587
3588    with caplog.at_level(logging.WARNING):
3589        stream._handle_status_line(
3590            "[STATUS] anchor_corrected requested_unix_ms=0 "
3591            "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3592        )
3593
3594    assert "Player A" in caplog.text
3595    assert "+500 ms" in caplog.text
3596
3597
3598def test_anchor_corrected_status_line_tolerates_malformed_line() -> None:
3599    """A malformed anchor_corrected line is dropped instead of raising or rebasing."""
3600    stream = AirPlayStream(_make_player())
3601    stream._start_position = 12.0
3602
3603    ended = stream._handle_status_line("[STATUS] anchor_corrected requested_unix_ms=garbage")
3604
3605    assert ended is False
3606    assert stream._start_position == 12.0
3607
3608
3609def test_content_cut_short_rebases_the_position_and_warns(
3610    caplog: pytest.LogCaptureFixture,
3611) -> None:
3612    """A cut that ended early gives back the ms the correction over-advanced the base by."""
3613    stream = AirPlayStream(_make_player())
3614    stream._start_position = 12.0
3615    stream._handle_status_line(
3616        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3617        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3618    )
3619    assert stream._start_position == 12.5
3620
3621    with caplog.at_level(logging.WARNING):
3622        ended = stream._handle_status_line(
3623            "[STATUS] content_cut requested_ms=500 cut_ms=180 cut_bytes=31752 drain_ms=210"
3624        )
3625
3626    assert ended is False
3627    assert stream._start_position == pytest.approx(12.18)
3628    assert "AirPlay content cut" in caplog.text
3629    assert "Player A" in caplog.text
3630    assert "320 ms short" in caplog.text
3631
3632
3633def test_content_cut_in_full_leaves_the_position_alone(caplog: pytest.LogCaptureFixture) -> None:
3634    """A cut that took what it asked for needs no correction and stays quiet."""
3635    stream = AirPlayStream(_make_player())
3636    stream._start_position = 12.0
3637    stream._handle_status_line(
3638        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3639        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3640    )
3641
3642    caplog.clear()
3643    with caplog.at_level(logging.WARNING):
3644        # a few ms below the request is byte quantization, not a short cut
3645        stream._handle_status_line(
3646            "[STATUS] content_cut requested_ms=500 cut_ms=498 cut_bytes=87887 drain_ms=505"
3647        )
3648
3649    assert stream._start_position == 12.5
3650    assert caplog.text == ""
3651
3652
3653def test_content_cut_after_a_new_anchor_is_not_reconciled() -> None:
3654    """A cut settling after a START must not be taken off that START's absolute base."""
3655    stream = AirPlayStream(_make_player())
3656    stream._start_position = 12.0
3657    stream._handle_status_line(
3658        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3659        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3660    )
3661    stream.rebase_position(30_000)
3662
3663    stream._handle_status_line(
3664        "[STATUS] content_cut requested_ms=500 cut_ms=0 cut_bytes=0 drain_ms=12"
3665    )
3666
3667    assert stream._start_position == 30.0
3668
3669
3670def test_content_cut_status_line_tolerates_malformed_line() -> None:
3671    """A malformed content_cut line is dropped instead of raising or rebasing."""
3672    stream = AirPlayStream(_make_player())
3673    stream._start_position = 12.0
3674    stream._handle_status_line(
3675        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3676        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3677    )
3678
3679    ended = stream._handle_status_line("[STATUS] content_cut requested_ms=500 cut_ms=garbage")
3680
3681    assert ended is False
3682    assert stream._start_position == 12.5
3683
3684
3685def test_clock_verified_status_line_is_debug_logged(caplog: pytest.LogCaptureFixture) -> None:
3686    """A clock_verified line needs no server action beyond a debug note of the margin."""
3687    stream = AirPlayStream(_make_player())
3688
3689    with caplog.at_level(logging.DEBUG):
3690        ended = stream._handle_status_line("[STATUS] clock_verified margin_ms=42")
3691
3692    assert ended is False
3693    assert "42" in caplog.text
3694
3695
3696@pytest.mark.asyncio
3697async def test_connect_error_status_line_tolerates_missing_fields() -> None:
3698    """A failure line without http/detail still yields the reported code."""
3699    stream = AirPlayStream(_make_player())
3700
3701    stream._handle_status_line("[STATUS] error code=connect_failed")
3702
3703    assert stream._connect_error == CliError("connect_failed", 0, "")
3704
3705
3706@pytest.mark.asyncio
3707async def test_auth_required_surfaces_password_required_error() -> None:
3708    """A device asking for a password produces an actionable, translated error."""
3709    stream = AirPlayStream(_make_player())
3710    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
3711    stream._process_ended.set()
3712
3713    with (
3714        patch.object(stream, "_cli_proc", MagicMock()),
3715        pytest.raises(PlayerCommandFailed) as err,
3716    ):
3717        await stream.wait_for_connection()
3718
3719    assert err.value.translation_key == "password_required"
3720
3721
3722@pytest.mark.asyncio
3723async def test_auth_failed_surfaces_authentication_failed_error() -> None:
3724    """A rejected password is reported as an authentication failure, not a timeout."""
3725    stream = AirPlayStream(_make_player())
3726    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
3727    stream._process_ended.set()
3728
3729    with (
3730        patch.object(stream, "_cli_proc", MagicMock()),
3731        pytest.raises(PlayerCommandFailed) as err,
3732    ):
3733        await stream.wait_for_connection()
3734
3735    assert err.value.translation_key == "authentication_failed"
3736
3737
3738@pytest.mark.asyncio
3739@pytest.mark.parametrize("code", ["auth_required", "auth_failed"])
3740async def test_refused_connection_is_not_reported_as_a_password_problem(code: str) -> None:
3741    """A device that turns the handshake away points at pairing, not at a password."""
3742    stream = AirPlayStream(_make_player())
3743    stream._handle_status_line(f'[STATUS] error code={code} http=403 detail="refused"')
3744    stream._process_ended.set()
3745
3746    with (
3747        patch.object(stream, "_cli_proc", MagicMock()),
3748        pytest.raises(PlayerCommandFailed) as err,
3749    ):
3750        await stream.wait_for_connection()
3751
3752    assert err.value.translation_key == "connection_refused"
3753
3754
3755@pytest.mark.asyncio
3756@pytest.mark.parametrize("code", ["auth_required", "auth_failed"])
3757async def test_refused_connection_never_marks_the_password_invalid(code: str) -> None:
3758    """
3759    A refusal must not leave a player demanding a password it may not even have.
3760
3761    tvOS 26 answers the pairing handshake with 403 for reasons unrelated to any
3762    secret, and the marker persists across restarts - so latching it there would
3763    strand the player in a setup flow no password can complete.
3764    """
3765    player = _make_player()
3766    stream = AirPlayStream(player)
3767
3768    stream._handle_status_line(f'[STATUS] error code={code} http=403 detail="refused"')
3769
3770    player.set_password_invalid.assert_not_called()
3771
3772
3773@pytest.mark.asyncio
3774async def test_generic_connect_failure_keeps_the_timeout_semantics() -> None:
3775    """A non-auth failure keeps raising the plain timeout its callers already handle."""
3776    stream = AirPlayStream(_make_player())
3777    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
3778    stream._process_ended.set()
3779
3780    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
3781        await stream.wait_for_connection()
3782
3783
3784@pytest.mark.asyncio
3785async def test_dead_process_fails_the_connect_wait_immediately() -> None:
3786    """
3787    A binary that reports no reason at all leaves the wait its plain timeout error.
3788
3789    It must still end the moment the process is gone instead of running out the
3790    full connect timeout.
3791    """
3792    stream = AirPlayStream(_make_player())
3793    stream._process_ended.set()  # process died without emitting a [STATUS] error line
3794
3795    started = asyncio.get_running_loop().time()
3796    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
3797        await stream.wait_for_connection()
3798
3799    assert asyncio.get_running_loop().time() - started < 1
3800
3801
3802@pytest.mark.asyncio
3803async def test_auth_failed_marks_the_stored_password_invalid() -> None:
3804    """A rejected password is persisted so the player keeps offering its setup action."""
3805    player = _make_player()
3806    stream = AirPlayStream(player)
3807
3808    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
3809
3810    player.set_password_invalid.assert_called_once_with(True)
3811
3812
3813@pytest.mark.asyncio
3814async def test_auth_required_also_marks_the_password_as_needed() -> None:
3815    """
3816    A device that demanded a password we could not supply flips into setup.
3817
3818    Devices can enforce a password without announcing it (stale TXT records), so
3819    the runtime signal must set the marker too - it is the only reliable one.
3820    """
3821    player = _make_player()
3822    stream = AirPlayStream(player)
3823
3824    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
3825
3826    player.set_password_invalid.assert_called_once_with(True)
3827
3828
3829@pytest.mark.asyncio
3830async def test_plain_connect_failures_leave_the_password_marker_alone() -> None:
3831    """A non-authentication failure says nothing about the stored password."""
3832    player = _make_player()
3833    stream = AirPlayStream(player)
3834
3835    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
3836
3837    player.set_password_invalid.assert_not_called()
3838
3839
3840@pytest.mark.asyncio
3841async def test_successful_connect_clears_the_password_marker() -> None:
3842    """Whatever the device accepted is a working password."""
3843    player = _make_player()
3844    stream = AirPlayStream(player)
3845
3846    stream._handle_status_line("[STATUS] connected")
3847
3848    assert stream.connected is True
3849    player.set_password_invalid.assert_called_once_with(False)
3850
3851
3852# --- Password preflight ---
3853
3854
3855@pytest.mark.asyncio
3856async def test_connect_refuses_password_device_without_password_or_credentials() -> None:
3857    """A password-protected AirPlay 2 device with nothing to authenticate never spawns a process."""
3858    player = _make_player()
3859    player.password_required = True
3860    player.get_setup_value = MagicMock(return_value=None)
3861    stream = AirPlayStream(player)
3862
3863    with pytest.raises(PlayerCommandFailed) as err:
3864        await stream.connect()
3865
3866    assert err.value.translation_key == "password_required"
3867    assert stream._cli_proc is None
3868
3869
3870@pytest.mark.asyncio
3871async def test_password_preflight_passes_with_password_or_credentials() -> None:
3872    """Either a configured password or stored credentials let the connect proceed."""
3873    player = _make_player()
3874    player.password_required = True
3875    player.get_setup_value = MagicMock(return_value=None)
3876    player.config.get_value = MagicMock(
3877        side_effect=lambda key, default=None: "s3cret" if key == CONF_PASSWORD else default
3878    )
3879    AirPlayStream(player)._check_password_preflight()
3880
3881    # credentials alone are enough: the binary's pair-verify leg may still succeed
3882    player.config.get_value = MagicMock(side_effect=lambda _key, default=None: default)
3883    player.get_setup_value = MagicMock(
3884        side_effect=lambda key, default=None: (
3885            "ab" * 96 if key == CONF_AIRPLAY_CREDENTIALS else default
3886        )
3887    )
3888    AirPlayStream(player)._check_password_preflight()
3889
3890
3891@pytest.mark.asyncio
3892async def test_password_preflight_skipped_for_raop() -> None:
3893    """The preflight only guards the native AirPlay 2 flow; RAOP carries its own password."""
3894    player = _make_player()
3895    player.protocol = StreamingProtocol.RAOP
3896    player.password_required = True
3897    player.get_setup_value = MagicMock(return_value=None)
3898
3899    AirPlayStream(player)._check_password_preflight()
3900
3901
3902@pytest.mark.asyncio
3903async def test_accepts_audio_ends_with_the_audio_eof() -> None:
3904    """
3905    A stream that was sent its audio EOF keeps running but takes no more audio.
3906
3907    The EOF closes the binary's stdin for good while the process itself plays out
3908    and exits, so a warm refill has to read this rather than the process state.
3909    """
3910    player = _make_player()
3911    stream = AirPlayStream(player)
3912    cli_proc = AsyncProcess(["cat"], stdin=True, stdout=True)
3913    await cli_proc.start()
3914    stream._cli_proc = cli_proc
3915    try:
3916        assert stream.running
3917        assert stream.accepts_audio
3918
3919        await stream.write_audio_eof()
3920
3921        assert stream.running
3922        assert not stream.accepts_audio
3923    finally:
3924        await cli_proc.close()
3925