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
2095def test_native_control_failure_before_publication_stays_reportable(
2096    caplog: pytest.LogCaptureFixture,
2097) -> None:
2098    """
2099    A failure reported before the stream owns the player does not spend the one-shot.
2100
2101    A bridge stream is published only once its process has connected, so a
2102    verdict dropped during that window must not stop the same failure from
2103    being reported afterwards.
2104    """
2105    player = _make_player()
2106    stream = AirPlayStream(player)
2107
2108    with caplog.at_level(logging.WARNING):
2109        stream._handle_status_line("[ERROR] AirPlay 2 control channel failed")
2110        assert "stopped answering native AirPlay 2 control keepalives" not in caplog.text
2111
2112        player.stream = stream
2113        stream._handle_status_line("[ERROR] AirPlay 2 control channel failed")
2114
2115    assert "stopped answering native AirPlay 2 control keepalives" in caplog.text
2116    player.provider.mass.config.set_raw_player_config_value.assert_not_called()
2117
2118
2119def test_unrelated_cli_error_is_not_a_control_channel_verdict() -> None:
2120    """A different runtime failure does not diagnose the native control route."""
2121    player = _make_player()
2122    stream = AirPlayStream(player)
2123    player.stream = stream
2124
2125    stream._handle_status_line("[ERROR] AirPlay 2 audio send failed")
2126
2127    assert stream._native_control_failure_warned is False
2128
2129
2130@pytest.mark.asyncio
2131async def test_clock_ready_projection_resolves_the_wait() -> None:
2132    """A probing receiver reports when its clock becomes usable, from its first probe."""
2133    stream = AirPlayStream(_make_player())
2134
2135    assert (
2136        stream._handle_status_line(
2137            "[STATUS] clock_ready mode=ptp state=probing streak_ms=0 exchanges=1 "
2138            f"ready_in_ms=2300 ready_at_unix_ms={START_UNIX_MS}"
2139        )
2140        is False
2141    )
2142
2143    assert await stream.wait_clock_ready(timeout=0.01) == (
2144        ClockReadiness.PROJECTED,
2145        START_UNIX_MS,
2146    )
2147
2148
2149@pytest.mark.asyncio
2150async def test_clock_ready_cold_line_keeps_waiting_for_a_projection() -> None:
2151    """A receiver that has not probed yet carries no projection, so the wait goes on."""
2152    stream = AirPlayStream(_make_player())
2153
2154    stream._handle_status_line(
2155        "[STATUS] clock_ready mode=ptp state=cold streak_ms=0 exchanges=0 "
2156        "ready_in_ms=0 ready_at_unix_ms=0"
2157    )
2158
2159    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.UNREPORTED, 0)
2160    assert not stream._clock_ready.is_set()
2161
2162    stream._handle_status_line(
2163        "[STATUS] clock_ready mode=ptp state=ready streak_ms=2400 exchanges=9 "
2164        f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
2165    )
2166
2167    assert await stream.wait_clock_ready(timeout=0.01) == (
2168        ClockReadiness.PROJECTED,
2169        START_UNIX_MS,
2170    )
2171
2172
2173@pytest.mark.asyncio
2174async def test_clock_ready_ntp_resolves_without_a_projection() -> None:
2175    """NTP timing has no receiver clock to wait for, so the wait ends with nothing."""
2176    stream = AirPlayStream(_make_player())
2177
2178    stream._handle_status_line(
2179        "[STATUS] clock_ready mode=ntp state=ready streak_ms=0 exchanges=0 "
2180        f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
2181    )
2182
2183    assert stream._clock_ready.is_set()
2184    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.NOT_APPLICABLE, 0)
2185
2186
2187@pytest.mark.asyncio
2188async def test_wait_clock_ready_times_out_for_a_binary_that_never_reports() -> None:
2189    """A binary that does not report readiness is told apart from one that answered."""
2190    stream = AirPlayStream(_make_player())
2191
2192    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.UNREPORTED, 0)
2193
2194
2195@pytest.mark.asyncio
2196async def test_clock_ready_stalled_state_warns_once(caplog: pytest.LogCaptureFixture) -> None:
2197    """
2198    A stalled receiver is reported loudly, once per stream session.
2199
2200    The warning names the player and the PTP ports to check; repeats of the
2201    same verdict within the session stay quiet.
2202    """
2203    grouped_player = _make_player()
2204    grouped_player.synced_to = "apleader"
2205    stream = AirPlayStream(grouped_player)
2206    grouped_player.stream = stream
2207
2208    with caplog.at_level(logging.DEBUG):
2209        ended = stream._handle_status_line(
2210            "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
2211            "ready_in_ms=0 ready_at_unix_ms=0"
2212        )
2213        stream._handle_status_line(
2214            "[STATUS] clock_ready mode=ptp state=stalled streak_ms=0 exchanges=0 "
2215            "ready_in_ms=0 ready_at_unix_ms=0"
2216        )
2217
2218    warnings = [record for record in caplog.records if record.levelno == logging.WARNING]
2219    assert ended is False
2220    assert len(warnings) == 1
2221    assert "Player A" in warnings[0].getMessage()
2222    assert "319/320" in warnings[0].getMessage()
2223    assert stream._clock_ready.is_set()
2224    assert await stream.wait_clock_ready(timeout=0.01) == (ClockReadiness.STALLED, 0)
2225
2226
2227def test_clock_ready_stall_warning_is_ptp_only(caplog: pytest.LogCaptureFixture) -> None:
2228    """An NTP-timed session has no clock of ours to answer, so it never reads as a stall."""
2229    stream = AirPlayStream(_make_player())
2230
2231    with caplog.at_level(logging.DEBUG):
2232        stream._handle_status_line(
2233            "[STATUS] clock_ready mode=ntp state=stalled streak_ms=0 exchanges=0 "
2234            f"ready_in_ms=0 ready_at_unix_ms={START_UNIX_MS}"
2235        )
2236
2237    assert [record for record in caplog.records if record.levelno >= logging.WARNING] == []
2238    assert stream._clock_ready_at_unix_ms == 0
2239
2240
2241@pytest.mark.parametrize("state", ["cold", "probing", "ready"])
2242def test_clock_ready_handshake_states_do_not_warn(
2243    state: str, caplog: pytest.LogCaptureFixture
2244) -> None:
2245    """Any state but a stall is a normal step of the clock handshake and stays quiet."""
2246    stream = AirPlayStream(_make_player())
2247
2248    with caplog.at_level(logging.DEBUG):
2249        ended = stream._handle_status_line(
2250            f"[STATUS] clock_ready mode=ptp state={state} streak_ms=900 exchanges=4 "
2251            f"ready_in_ms=940 ready_at_unix_ms={START_UNIX_MS}"
2252        )
2253
2254    assert ended is False
2255    assert [record for record in caplog.records if record.levelno >= logging.WARNING] == []
2256
2257
2258def test_elapsed_includes_start_position() -> None:
2259    """Reported progress is the current anchor's media base plus the binary delta."""
2260    player = _make_player()
2261    stream = AirPlayStream(player)
2262    stream._start_position = 12.0
2263
2264    stream._update_elapsed(1.5)
2265
2266    player.set_state_from_stream.assert_called_once_with(
2267        state=PlaybackState.PLAYING,
2268        elapsed_time=13.5,
2269        stream=stream,
2270    )
2271
2272
2273def test_reanchor_status_sets_the_cumulative_total() -> None:
2274    """Each [STATUS] REANCHOR line carries the authoritative total, so it SETS the shift."""
2275    stream = AirPlayStream(_make_player())
2276    assert stream.cumulative_shift_seconds == 0.0
2277
2278    assert (
2279        stream._handle_status_line(
2280            "[STATUS] REANCHOR shifted_frames=67870 total_shifted_frames=67870 sample_rate=44100"
2281        )
2282        is False
2283    )
2284    assert stream.cumulative_shift_seconds == pytest.approx(67870 / 44100)
2285
2286    # the second event reports the running total, never a delta to add
2287    stream._handle_status_line(
2288        "[STATUS] REANCHOR shifted_frames=67870 total_shifted_frames=135740 sample_rate=44100"
2289    )
2290    assert stream.cumulative_shift_seconds == pytest.approx(135740 / 44100)
2291
2292
2293def test_human_readable_reanchor_warn_is_not_counted() -> None:
2294    """The binary's warn line accompanies the status line; counting it would double it."""
2295    stream = AirPlayStream(_make_player())
2296
2297    ended = stream._handle_status_line(
2298        "[AP2] Re-anchored after PCM starvation: shifted_frames=67870 lead_frames=77175 count=1"
2299    )
2300
2301    assert ended is False
2302    assert stream.cumulative_shift_seconds == 0.0
2303
2304
2305def test_reanchor_status_prefers_line_sample_rate() -> None:
2306    """The sample rate carried on the [STATUS] REANCHOR line wins over the stream format."""
2307    player = _make_player()
2308    stream = AirPlayStream(player)
2309    stream.pcm_format = AudioFormat(
2310        content_type=ContentType.PCM_S16LE, sample_rate=48000, bit_depth=16
2311    )
2312
2313    stream._handle_status_line(
2314        "[STATUS] REANCHOR shifted_frames=44100 total_shifted_frames=44100 sample_rate=44100"
2315    )
2316
2317    # converts at 44100 (from the line), not 48000 (the stream format) -> exactly 1.0s
2318    assert stream.cumulative_shift_seconds == pytest.approx(1.0)
2319
2320
2321def test_reanchor_status_ignores_line_without_total() -> None:
2322    """A [STATUS] REANCHOR line missing the total leaves the shift unchanged."""
2323    stream = AirPlayStream(_make_player())
2324    stream.cumulative_shift_seconds = 1.5
2325
2326    stream._handle_status_line("[STATUS] REANCHOR shifted_frames=67870 sample_rate=44100")
2327
2328    assert stream.cumulative_shift_seconds == 1.5
2329
2330
2331@pytest.mark.asyncio
2332async def test_start_resets_reanchor_shift() -> None:
2333    """A START re-anchors from scratch, clearing the accumulated shift."""
2334    stream = AirPlayStream(_make_player())
2335    stream._cli_proc = _make_cli_proc()
2336    stream._connected.set()
2337    stream.cumulative_shift_seconds = 3.078
2338
2339    with patch.object(
2340        stream,
2341        "_write_cli_command",
2342        new_callable=AsyncMock,
2343        side_effect=_acking_write_cli_command(stream),
2344    ):
2345        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
2346
2347    assert stream.cumulative_shift_seconds == 0.0
2348
2349
2350@pytest.mark.asyncio
2351async def test_connect_resets_accumulated_shift() -> None:
2352    """A fresh cliairplay process starts from a zero playout shift."""
2353    player = _make_player()
2354    stream = AirPlayStream(player)
2355    stream.cumulative_shift_seconds = 5.0
2356    process = MagicMock(closed=False)
2357    process.start = AsyncMock(return_value=None)
2358
2359    def consume_task(awaitable: Any) -> MagicMock:
2360        awaitable.close()
2361        task = MagicMock()
2362        task.done.return_value = True
2363        return task
2364
2365    player.provider.mass.create_task.side_effect = consume_task
2366    with (
2367        patch.object(stream, "_build_cli_args", new_callable=AsyncMock, return_value=["binary"]),
2368        patch("music_assistant.providers.airplay.stream.AsyncProcess", return_value=process),
2369        patch.object(stream.commands_pipe, "create", new_callable=AsyncMock),
2370    ):
2371        await stream.connect()
2372
2373    assert stream.cumulative_shift_seconds == 0.0
2374
2375
2376@pytest.mark.asyncio
2377async def test_initial_metadata_skips_artwork() -> None:
2378    """The pre-connect metadata push cannot delay setup on artwork rendering."""
2379    player = _make_player()
2380    stream = AirPlayStream(player)
2381    stream._cli_proc = _make_cli_proc()
2382    metadata = MagicMock(
2383        title="Track",
2384        artist="Artist",
2385        album="Album",
2386        duration=180,
2387        image_url="image",
2388    )
2389
2390    with (
2391        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2392        patch.object(
2393            stream,
2394            "_render_and_send_artwork",
2395            new_callable=AsyncMock,
2396        ) as send_artwork,
2397    ):
2398        await stream.send_metadata(0, metadata, send_artwork=False)
2399
2400    # the metadata push is followed by an explicit progress anchor, even at
2401    # zero (receivers like the WiiM Amp gate their rendering on it)
2402    assert send_command.await_count == 2
2403    assert "TITLE=Track" in send_command.await_args_list[0].args[0]
2404    assert send_command.await_args_list[1].args[0].endswith("PROGRESS=0")
2405    assert stream._last_progress_sent == 0
2406    send_artwork.assert_not_awaited()
2407
2408
2409@pytest.mark.asyncio
2410async def test_wait_for_connection_pushes_metadata_immediately() -> None:
2411    """
2412    Track metadata is pushed the instant the binary can receive it.
2413
2414    Receivers that gate audio rendering on receiving timeline-anchored metadata
2415    (e.g. Sonos over native AirPlay 2) must not be left silent while a deferred
2416    push is pending, so the metadata callback runs synchronously on connect
2417    while only the volume resend stays on the delayed path.
2418    """
2419    player = _make_player()
2420    player.volume_muted = False
2421    stream = AirPlayStream(player)
2422    stream._connected.set()  # connection already established
2423    player.provider.mass.call_later = MagicMock()
2424    operation_order: list[str] = []
2425
2426    async def wait_for_reader(_timeout: float) -> bool:
2427        operation_order.append("reader")
2428        return True
2429
2430    async def send_current_metadata(**_kwargs: Any) -> None:
2431        operation_order.append("metadata")
2432
2433    with (
2434        patch.object(stream, "_cli_proc", MagicMock()),  # non-None so the method proceeds
2435        patch.object(stream.commands_pipe, "wait_for_reader", side_effect=wait_for_reader),
2436        patch.object(stream, "_send_current_metadata", side_effect=send_current_metadata),
2437        patch.object(stream, "send_cli_command", return_value=None),  # avoid a real coroutine
2438    ):
2439        await stream.wait_for_connection()
2440
2441    # Nothing is written before the binary has a reader on the command pipe.
2442    assert operation_order == ["reader", "metadata"]
2443    # Metadata pushed synchronously on connect...
2444    player.on_player_media_updated.assert_called_once_with()
2445    # ...and never routed through the delayed call_later path.
2446    deferred_callables = [call.args[1] for call in player.provider.mass.call_later.call_args_list]
2447    assert player.on_player_media_updated not in deferred_callables
2448    # The volume resend is still deferred (existing behavior preserved).
2449    assert player.provider.mass.call_later.call_count == 1
2450    assert player.provider.mass.call_later.call_args_list[0].args[0] == 2
2451
2452
2453@pytest.mark.asyncio
2454async def test_deferred_volume_resend_reads_the_state_when_it_fires() -> None:
2455    """The repeated volume push must carry the volume at fire time, not at connect time."""
2456    player = _make_player()
2457    player.volume_muted = False
2458    player.volume_level = 40
2459    stream = AirPlayStream(player)
2460    stream._connected.set()  # connection already established
2461    player.provider.mass.call_later = MagicMock()
2462
2463    with (
2464        patch.object(stream, "_cli_proc", MagicMock()),  # non-None so the method proceeds
2465        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=True)),
2466        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2467        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2468    ):
2469        await stream.wait_for_connection()
2470        send_command.assert_awaited_with("VOLUME=40")
2471        deferred = player.provider.mass.call_later.call_args_list[0].args[1]
2472
2473        # a volume change between the connect and the resend firing must survive
2474        player.volume_level = 75
2475        await deferred()
2476        send_command.assert_awaited_with("VOLUME=75")
2477
2478        # ...and so must a mute
2479        player.volume_muted = True
2480        await deferred()
2481
2482    send_command.assert_awaited_with("VOLUME=0")
2483
2484
2485@pytest.mark.asyncio
2486async def test_wait_for_connection_sends_volume_when_player_owns_it() -> None:
2487    """The initial volume push is sent when this output owns its own volume."""
2488    player = _make_player()
2489    player.owns_volume = True
2490    player.volume_muted = False
2491    stream = AirPlayStream(player)
2492    stream._connected.set()
2493
2494    with (
2495        patch.object(stream, "_cli_proc", MagicMock()),
2496        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=True)),
2497        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2498        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2499    ):
2500        await stream.wait_for_connection()
2501
2502    send_command.assert_awaited_with(f"VOLUME={player.volume_level}")
2503
2504
2505@pytest.mark.asyncio
2506async def test_wait_for_connection_skips_volume_when_another_control_owns_it() -> None:
2507    """No unsolicited volume push when another control owns this output's volume."""
2508    player = _make_player()
2509    player.owns_volume = False
2510    player.volume_muted = False
2511    stream = AirPlayStream(player)
2512    stream._connected.set()
2513
2514    with (
2515        patch.object(stream, "_cli_proc", MagicMock()),
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
2522    send_command.assert_not_awaited()
2523
2524
2525@pytest.mark.asyncio
2526async def test_wait_for_connection_sends_volume_when_muted_without_ownership() -> None:
2527    """A latched mute is still pushed even when another control owns the volume."""
2528    player = _make_player()
2529    player.owns_volume = False
2530    player.volume_muted = True
2531    stream = AirPlayStream(player)
2532    stream._connected.set()
2533
2534    with (
2535        patch.object(stream, "_cli_proc", MagicMock()),
2536        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=True)),
2537        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2538        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2539    ):
2540        await stream.wait_for_connection()
2541
2542    send_command.assert_awaited_with("VOLUME=0")
2543
2544
2545@pytest.mark.asyncio
2546async def test_wait_for_connection_fails_on_an_unread_command_pipe() -> None:
2547    """A binary that never attaches to the command pipe can never be anchored: fail the connect."""
2548    player = _make_player()
2549    player.logger = MagicMock()
2550    player.volume_muted = False
2551    stream = AirPlayStream(player)
2552    stream._connected.set()
2553    stream._cli_proc = _make_cli_proc()
2554
2555    with (
2556        patch.object(
2557            stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=False)
2558        ) as wait_for_reader,
2559        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2560        patch.object(stream, "send_cli_command", return_value=None),
2561        pytest.raises(PlayerCommandFailed, match="command pipe") as err,
2562    ):
2563        await stream.wait_for_connection()
2564
2565    wait_for_reader.assert_awaited_once()
2566    assert player.display_name in str(err.value)
2567
2568
2569@pytest.mark.asyncio
2570async def test_wait_for_connection_stays_quiet_about_a_stopped_stream() -> None:
2571    """A stream torn down while connecting took its command pipe along, which is no fault."""
2572    player = _make_player()
2573    player.logger = MagicMock()
2574    player.volume_muted = False
2575    stream = AirPlayStream(player)
2576    stream._connected.set()
2577    stream._cli_proc = _make_cli_proc()
2578    stream._stopping = True
2579
2580    with (
2581        patch.object(stream.commands_pipe, "wait_for_reader", new=AsyncMock(return_value=False)),
2582        patch.object(stream, "_send_current_metadata", new_callable=AsyncMock),
2583        patch.object(stream, "send_cli_command", return_value=None),
2584    ):
2585        await stream.wait_for_connection()
2586
2587    player.logger.warning.assert_not_called()
2588
2589
2590@pytest.mark.asyncio
2591async def test_prepare_artwork_returns_cache_path() -> None:
2592    """Artwork preparation returns the shared cache path without a per-player copy."""
2593    player = _make_player()
2594    stream = AirPlayStream(player)
2595    image_url = "https://example.com/artwork.png"
2596    cached_path = "/cache/thumbnails/artwork_flat.jpg"
2597
2598    with patch(
2599        "music_assistant.providers.airplay.stream.get_image_thumb_path",
2600        new=AsyncMock(return_value=cached_path),
2601    ) as get_thumb_path:
2602        result = await stream._prepare_artwork(image_url, 1)
2603
2604    assert result == cached_path
2605    assert not hasattr(stream, "_artwork_paths")
2606    get_thumb_path.assert_awaited_once_with(
2607        stream.mass,
2608        image_url,
2609        AIRPLAY_ARTWORK_SIZE,
2610        "",
2611        image_format="JPEG",
2612        flatten_transparency=True,
2613    )
2614
2615
2616@pytest.mark.asyncio
2617async def test_stop_cleans_up_when_stop_command_fails() -> None:
2618    """A command-pipe failure cannot skip process and stream cleanup."""
2619    player = _make_player()
2620    stream = AirPlayStream(player)
2621    process = MagicMock()
2622    process.closed = False
2623    process.kill = AsyncMock()
2624    stream._cli_proc = process
2625
2626    with (
2627        patch.object(
2628            stream.commands_pipe,
2629            "write",
2630            new_callable=AsyncMock,
2631            side_effect=OSError("command pipe failed"),
2632        ),
2633        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
2634        pytest.raises(OSError, match="command pipe failed"),
2635    ):
2636        await stream.stop(force=True)
2637
2638    assert stream._stopped is True
2639    assert stream._cleanup_complete is True
2640    remove_pipe.assert_awaited_once()
2641    process.kill.assert_awaited_once()
2642    player.set_state_from_stream.assert_called_once_with(
2643        state=PlaybackState.IDLE,
2644        elapsed_time=0,
2645        stream=stream,
2646    )
2647
2648
2649@pytest.mark.asyncio
2650async def test_stop_awaits_cancelled_stdout_reader() -> None:
2651    """Stream teardown waits for the stdout reader to release process resources."""
2652    player = _make_player()
2653    stream = AirPlayStream(player)
2654    process = MagicMock()
2655    process.closed = False
2656    process.kill = AsyncMock()
2657    stream._cli_proc = process
2658    reader_started = asyncio.Event()
2659
2660    async def _stdout_reader() -> None:
2661        reader_started.set()
2662        await asyncio.Event().wait()
2663
2664    reader_task = asyncio.create_task(_stdout_reader())
2665    stream._stdout_reader_task = reader_task
2666    await reader_started.wait()
2667
2668    with (
2669        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock),
2670        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2671    ):
2672        await stream.stop(force=True)
2673
2674    assert reader_task.cancelled()
2675    process.kill.assert_awaited_once()
2676
2677
2678@pytest.mark.asyncio
2679async def test_force_stop_does_not_wait_for_artwork_render() -> None:
2680    """Force-stop tears down immediately while remote artwork rendering finishes."""
2681    player = _make_player()
2682    stream = AirPlayStream(player)
2683    process = MagicMock()
2684    process.closed = False
2685    process.kill = AsyncMock()
2686    stream._cli_proc = process
2687    metadata = MagicMock(
2688        title="Track",
2689        artist="Artist",
2690        album="Album",
2691        duration=180,
2692        image_url="slow-image",
2693    )
2694    artwork_started = asyncio.Event()
2695    release_artwork = asyncio.Event()
2696
2697    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
2698        artwork_started.set()
2699        await release_artwork.wait()
2700        return "late.jpg"
2701
2702    with (
2703        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock),
2704        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2705        patch.object(
2706            stream,
2707            "_prepare_artwork",
2708            new_callable=AsyncMock,
2709            side_effect=_prepare_artwork,
2710        ),
2711    ):
2712        metadata_task = asyncio.create_task(stream.send_metadata(0, metadata))
2713        await artwork_started.wait()
2714        await asyncio.wait_for(stream.stop(force=True), timeout=0.5)
2715        release_artwork.set()
2716        await metadata_task
2717
2718    assert stream._cleanup_complete is True
2719    process.kill.assert_awaited_once()
2720
2721
2722@pytest.mark.asyncio
2723async def test_process_eof_cleans_up_command_pipe() -> None:
2724    """A naturally ended CLI stream removes its command pipe."""
2725    player = _make_player()
2726    stream = AirPlayStream(player)
2727    process = MagicMock()
2728    stream._cli_proc = process
2729
2730    async def _stderr_lines() -> AsyncGenerator[str]:
2731        yield "[STATUS] eof"
2732
2733    with (
2734        patch.object(process, "iter_stderr", return_value=_stderr_lines()),
2735        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock) as remove_pipe,
2736    ):
2737        await stream._stderr_reader()
2738
2739    assert stream._stopped is True
2740    remove_pipe.assert_awaited_once()
2741    player.schedule_group_rejoin.assert_not_called()
2742
2743
2744async def _run_unexpected_process_death(
2745    player: MagicMock, *, still_owned: bool = True
2746) -> AirPlayStream:
2747    """
2748    Drive the stderr reader through an unexpected process exit.
2749
2750    :param still_owned: Whether the dying stream is still the player's current
2751        one. False models a stream a newer session already superseded.
2752    """
2753    stream = AirPlayStream(player)
2754    if still_owned:
2755        player.stream = stream
2756    process = MagicMock()
2757    stream._cli_proc = process
2758
2759    async def _stderr_lines() -> AsyncGenerator[str]:
2760        yield "some final log line"
2761
2762    with (
2763        patch.object(process, "iter_stderr", return_value=_stderr_lines()),
2764        patch.object(stream.commands_pipe, "remove", new_callable=AsyncMock),
2765    ):
2766        await stream._stderr_reader()
2767    return stream
2768
2769
2770@pytest.mark.asyncio
2771async def test_unexpected_death_of_synced_child_schedules_rejoin() -> None:
2772    """A grouped member whose process dies unexpectedly gets a re-join scheduled."""
2773    player = _make_player()
2774    player.synced_to = "leader"
2775    player.group_members = []
2776    # the leader's other members are captured as fallback candidates in case
2777    # leadership transfers while the re-join backoff runs
2778    leader = MagicMock()
2779    leader.group_members = ["leader", player.player_id, "sibling"]
2780    player.provider.mass.players.get_player.return_value = leader
2781
2782    stream = await _run_unexpected_process_death(player)
2783
2784    player.schedule_group_rejoin.assert_called_once_with(["leader", "sibling"])
2785    player.set_state_from_stream.assert_called_once_with(
2786        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2787    )
2788
2789
2790@pytest.mark.asyncio
2791async def test_unexpected_death_of_leader_schedules_rejoin_to_members() -> None:
2792    """A dying leader re-joins towards its surviving members (leadership transfers)."""
2793    player = _make_player()
2794    player.synced_to = None
2795    player.group_members = [player.player_id, "child1", "child2"]
2796
2797    await _run_unexpected_process_death(player)
2798
2799    player.schedule_group_rejoin.assert_called_once_with(["child1", "child2"])
2800    # the controller sets the leader's final state (transfer or dissolve)
2801    player.set_state_from_stream.assert_not_called()
2802
2803
2804@pytest.mark.asyncio
2805async def test_unexpected_death_of_solo_player_schedules_no_rejoin() -> None:
2806    """An ungrouped player's process death only marks the player idle."""
2807    player = _make_player()
2808    player.synced_to = None
2809    player.group_members = []
2810
2811    stream = await _run_unexpected_process_death(player)
2812
2813    player.schedule_group_rejoin.assert_not_called()
2814    player.set_state_from_stream.assert_called_once_with(
2815        state=PlaybackState.IDLE, elapsed_time=0, stream=stream
2816    )
2817
2818
2819@pytest.mark.asyncio
2820async def test_unexpected_death_of_static_group_member_drops_member_only() -> None:
2821    """A static group member's death drops just that member, never the whole group."""
2822    player = _make_player()
2823    player.synced_to = "leader"
2824    player.group_members = []
2825    # the player is a static member of an actively playing group player, for
2826    # which cmd_ungroup would release (stop) the WHOLE group
2827    player.state.active_group = "syncgroup1"
2828    group_player = MagicMock()
2829    group_player.static_group_members = ["leader", player.player_id]
2830    leader = MagicMock()
2831    leader.group_members = ["leader", player.player_id]
2832    player.provider.mass.players.get_player.side_effect = lambda player_id: {
2833        "syncgroup1": group_player,
2834        "leader": leader,
2835    }.get(player_id)
2836
2837    await _run_unexpected_process_death(player)
2838
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"])
2845
2846
2847@pytest.mark.asyncio
2848async def test_unexpected_death_of_superseded_stream_leaves_the_group_alone() -> None:
2849    """
2850    A superseded process dying does not ungroup the player or schedule a re-join.
2851
2852    Its death is the newer session taking the receiver over, so acting on it
2853    would tear down the healthy session that replaced it.
2854    """
2855    player = _make_player()
2856    player.synced_to = "leader"
2857    player.group_members = []
2858    player.stream = AirPlayStream(player)
2859
2860    stream = await _run_unexpected_process_death(player, still_owned=False)
2861
2862    players_controller = player.provider.mass.players
2863    players_controller.cmd_ungroup.assert_not_called()
2864    players_controller.cmd_set_members.assert_not_called()
2865    player.schedule_group_rejoin.assert_not_called()
2866    player.set_state_from_stream.assert_not_called()
2867    assert stream._stopped is True
2868
2869
2870@pytest.mark.asyncio
2871async def test_process_eof_during_render_does_not_send_artwork() -> None:
2872    """A cache lookup finishing after EOF cannot send stale artwork."""
2873    player = _make_player()
2874    stream = AirPlayStream(player)
2875    render_started = asyncio.Event()
2876    release_render = asyncio.Event()
2877
2878    async def _get_image_thumb_path(*_args: Any, **_kwargs: Any) -> str:
2879        render_started.set()
2880        await release_render.wait()
2881        return "/cache/thumbnails/artwork.jpg"
2882
2883    with (
2884        patch(
2885            "music_assistant.providers.airplay.stream.get_image_thumb_path",
2886            new_callable=AsyncMock,
2887            side_effect=_get_image_thumb_path,
2888        ),
2889        patch.object(stream, "send_cli_command", new_callable=AsyncMock) as send_command,
2890    ):
2891        render_task = asyncio.create_task(stream._render_and_send_artwork("image", 1))
2892        await render_started.wait()
2893        stream._stopped = True
2894        release_render.set()
2895        await render_task
2896
2897    send_command.assert_not_awaited()
2898
2899
2900@pytest.mark.asyncio
2901async def test_concurrent_metadata_updates_only_send_latest_artwork() -> None:
2902    """An older slow artwork render cannot overwrite a newer track update."""
2903    player = _make_player()
2904    stream = AirPlayStream(player)
2905    process = MagicMock()
2906    process.closed = False
2907    stream._cli_proc = process
2908    first_render_started = asyncio.Event()
2909    release_first_render = asyncio.Event()
2910
2911    old_metadata = MagicMock(
2912        title="Old track",
2913        artist="Artist",
2914        album="Album",
2915        duration=180,
2916        image_url="old-image",
2917    )
2918    new_metadata = MagicMock(
2919        title="New track",
2920        artist="Artist",
2921        album="Album",
2922        duration=180,
2923        image_url="new-image",
2924    )
2925
2926    async def _prepare_artwork(image_url: str, _generation: int) -> str:
2927        if image_url == "old-image":
2928            first_render_started.set()
2929            await release_first_render.wait()
2930            return "old.jpg"
2931        return "new.jpg"
2932
2933    with (
2934        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2935        patch.object(
2936            stream,
2937            "_prepare_artwork",
2938            new_callable=AsyncMock,
2939            side_effect=_prepare_artwork,
2940        ),
2941        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
2942    ):
2943        old_task = asyncio.create_task(stream.send_metadata(0, old_metadata))
2944        await first_render_started.wait()
2945        new_task = asyncio.create_task(stream.send_metadata(0, new_metadata))
2946        # the new update supersedes the old render once the old push's render
2947        # budget lapses and the metadata lock is released
2948        await new_task
2949        assert stream._metadata_generation == 2
2950        release_first_render.set()
2951        await old_task
2952
2953    commands = [call.args[0].decode() for call in write_command.await_args_list]
2954    assert any("TITLE=New track" in command for command in commands)
2955    assert not any("ARTWORK=old.jpg" in command for command in commands)
2956    last_bundle = [command for command in commands if command.endswith("ACTION=SENDMETA\n")][-1]
2957    assert "ARTWORKFILE=new.jpg\n" in last_bundle
2958
2959
2960@pytest.mark.asyncio
2961async def test_artwork_url_form_change_does_not_resend_artwork() -> None:
2962    """Alternating URL forms of the same imageproxy image send artwork only once."""
2963    player = _make_player()
2964    stream = AirPlayStream(player)
2965    stream._cli_proc = _make_cli_proc()
2966
2967    def make_metadata(image_url: str) -> MagicMock:
2968        return MagicMock(
2969            corrected_elapsed_time=0,
2970            queue_item_id="item-1",
2971            title="Track",
2972            artist="Artist",
2973            album="Album",
2974            duration=180,
2975            image_url=image_url,
2976        )
2977
2978    # the queue session builds the image URL on the stream server base, the
2979    # player state on the webserver base - same image id behind both forms
2980    image_id = "ab" * 32
2981    other_image_id = "cd" * 32
2982    session_media = make_metadata(
2983        f"http://192.168.1.5:8097/imageproxy/{image_id}?size=512&fmt=jpeg"
2984    )
2985    state_media = make_metadata(f"http://192.168.1.5:8095/imageproxy/{image_id}?size=512&fmt=png")
2986    other_image_media = make_metadata(
2987        f"http://192.168.1.5:8095/imageproxy/{other_image_id}?size=512&fmt=png"
2988    )
2989
2990    with (
2991        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
2992        patch.object(
2993            stream,
2994            "_prepare_artwork",
2995            new_callable=AsyncMock,
2996            return_value="/cache/thumb.jpg",
2997        ) as prepare_artwork,
2998    ):
2999        await stream.send_metadata(None, session_media)
3000        generation_after_first_send = stream._metadata_generation
3001        # the post-START push (session media) and the media-updated push
3002        # (player state) alternate on every seek
3003        await stream.send_metadata(None, state_media)
3004        await stream.send_metadata(None, session_media)
3005        assert stream._metadata_generation == generation_after_first_send
3006        await stream.send_metadata(None, other_image_media)
3007
3008    commands = [call.args[0].decode() for call in write_command.await_args_list]
3009    bundled = [command for command in commands if "ARTWORKFILE=" in command]
3010    resends = [command for command in commands if command.startswith("ARTWORK=")]
3011    assert len(bundled) == 1
3012    assert resends == ["ARTWORK=/cache/thumb.jpg\n"]
3013    assert prepare_artwork.await_count == 2
3014    assert stream._metadata_artwork_checksum == other_image_id
3015
3016
3017@pytest.mark.asyncio
3018async def test_metadata_revert_resends_text_after_superseded_artwork() -> None:
3019    """Reverting while artwork renders restores the previously displayed track text."""
3020    player = _make_player()
3021    stream = AirPlayStream(player)
3022    process = MagicMock()
3023    process.closed = False
3024    stream._cli_proc = process
3025    first_metadata = MagicMock(
3026        title="First track",
3027        artist="Artist",
3028        album="Album",
3029        duration=180,
3030        image_url=None,
3031    )
3032    second_metadata = MagicMock(
3033        title="Second track",
3034        artist="Artist",
3035        album="Album",
3036        duration=180,
3037        image_url="second-image",
3038    )
3039    first_checksum = "First track|Artist|Album|180|None"
3040    stream._metadata_text_checksum = first_checksum
3041    stream._pending_metadata_checksum = first_checksum
3042    artwork_started = asyncio.Event()
3043    release_artwork = asyncio.Event()
3044
3045    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3046        artwork_started.set()
3047        await release_artwork.wait()
3048        return "second.jpg"
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            side_effect=_prepare_artwork,
3057        ),
3058    ):
3059        second_task = asyncio.create_task(stream.send_metadata(0, second_metadata))
3060        await artwork_started.wait()
3061        revert_task = asyncio.create_task(stream.send_metadata(0, first_metadata))
3062        await asyncio.sleep(0)
3063        release_artwork.set()
3064        await asyncio.gather(second_task, revert_task)
3065
3066    metadata_commands = [
3067        call.args[0].decode()
3068        for call in write_command.await_args_list
3069        if "ACTION=SENDMETA" in call.args[0].decode()
3070    ]
3071    assert "TITLE=Second track" in metadata_commands[0]
3072    assert "TITLE=First track" in metadata_commands[-1]
3073
3074
3075@pytest.mark.asyncio
3076async def test_repeated_metadata_retries_superseded_artwork() -> None:
3077    """A B-to-C-to-B update sequence still applies B artwork after supersession."""
3078    player = _make_player()
3079    stream = AirPlayStream(player)
3080    process = MagicMock()
3081    process.closed = False
3082    stream._cli_proc = process
3083    initial_text_checksum = "item-initial|Initial|Artist|Album"
3084    initial_checksum = f"{initial_text_checksum}|initial-image"
3085    stream._metadata_artwork_checksum = "initial-image"
3086    stream._metadata_text_checksum = initial_text_checksum
3087    stream._pending_metadata_checksum = initial_checksum
3088    metadata_b = MagicMock(
3089        queue_item_id="item-b",
3090        title="Track B",
3091        artist="Artist",
3092        album="Album",
3093        duration=180,
3094        image_url="b-image",
3095    )
3096    metadata_c = MagicMock(
3097        queue_item_id="item-c",
3098        title="Track C",
3099        artist="Artist",
3100        album="Album",
3101        duration=180,
3102        image_url="c-image",
3103    )
3104    first_artwork_started = asyncio.Event()
3105    release_first_artwork = asyncio.Event()
3106    c_artwork_started = asyncio.Event()
3107    release_c_artwork = asyncio.Event()
3108    b_render_count = 0
3109
3110    async def _prepare_artwork(image_url: str, _generation: int) -> str:
3111        nonlocal b_render_count
3112        if image_url == "b-image":
3113            b_render_count += 1
3114            if b_render_count == 1:
3115                first_artwork_started.set()
3116                await release_first_artwork.wait()
3117                return "b-stale.jpg"
3118            return "b-final.jpg"
3119        c_artwork_started.set()
3120        await release_c_artwork.wait()
3121        return "c.jpg"
3122
3123    with (
3124        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3125        patch.object(
3126            stream,
3127            "_prepare_artwork",
3128            new_callable=AsyncMock,
3129            side_effect=_prepare_artwork,
3130        ) as prepare_artwork,
3131        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.05),
3132    ):
3133        first_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
3134        await first_artwork_started.wait()
3135        c_task = asyncio.create_task(stream.send_metadata(0, metadata_c))
3136        await c_artwork_started.wait()
3137        final_b_task = asyncio.create_task(stream.send_metadata(0, metadata_b))
3138        await final_b_task
3139        release_first_artwork.set()
3140        release_c_artwork.set()
3141        await asyncio.gather(first_b_task, c_task)
3142
3143    rendered_images = [args.args[0] for args in prepare_artwork.await_args_list]
3144    commands = [args.args[0].decode() for args in write_command.await_args_list]
3145    assert rendered_images == ["b-image", "c-image", "b-image"]
3146    assert "ARTWORK=b-stale.jpg\n" not in commands
3147    assert "ARTWORK=c.jpg\n" not in commands
3148    # the final B render completed within the budget, so it rides the bundle
3149    last_bundle = [command for command in commands if command.endswith("ACTION=SENDMETA\n")][-1]
3150    assert "ARTWORKFILE=b-final.jpg\n" in last_bundle
3151    assert stream._metadata_artwork_checksum == "b-image"
3152
3153
3154@pytest.mark.asyncio
3155async def test_send_metadata_passes_cached_artwork_path_to_binary() -> None:
3156    """The staged artwork carries the absolute cache path returned by preparation."""
3157    player = _make_player()
3158    stream = AirPlayStream(player)
3159    metadata = MagicMock(
3160        duration=180,
3161        title="Track",
3162        artist="Artist",
3163        album="Album",
3164        image_url="https://example.com/artwork.png",
3165    )
3166    cached_path = "/cache/thumbnails/artwork_flat.jpg"
3167    send_command = AsyncMock()
3168
3169    with (
3170        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value=cached_path)),
3171        patch.object(stream, "send_cli_command", new=send_command),
3172    ):
3173        await stream.send_metadata(None, metadata)
3174
3175    assert f"ARTWORKFILE={cached_path}\n" in send_command.await_args_list[-1].args[0]
3176
3177
3178@pytest.mark.asyncio
3179async def test_track_change_bundles_ready_artwork_into_a_single_push() -> None:
3180    """A track change whose artwork renders within budget lands as ONE bundled write."""
3181    stream = AirPlayStream(_make_player())
3182    stream._cli_proc = _make_cli_proc()
3183    metadata = MagicMock(
3184        queue_item_id="item-1",
3185        duration=180,
3186        title="Track",
3187        artist="Artist",
3188        album="Album",
3189        image_url="image",
3190    )
3191
3192    with (
3193        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3194        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
3195    ):
3196        await stream.send_metadata(0, metadata)
3197
3198    assert write_command.await_count == 2
3199    lines = write_command.await_args_list[0].args[0].decode().splitlines()
3200    assert "TITLE=Track" in lines
3201    assert "ITEMID=item-1" in lines
3202    # the artwork is staged before the SENDMETA applies the whole bundle
3203    assert lines[-2:] == ["ARTWORKFILE=/cache/art.jpg", "ACTION=SENDMETA"]
3204    # the bundle is one write; the explicit progress anchor follows separately
3205    assert write_command.await_args_list[1].args[0].decode().endswith("PROGRESS=0\n")
3206    assert stream._last_progress_sent == 0
3207    assert stream._metadata_artwork_checksum == "image"
3208
3209
3210@pytest.mark.asyncio
3211async def test_track_change_artwork_missing_the_budget_follows_as_artwork_command() -> None:
3212    """A render missing the bundling budget still delivers via ARTWORK once it completes."""
3213    stream = AirPlayStream(_make_player())
3214    stream._cli_proc = _make_cli_proc()
3215    metadata = MagicMock(
3216        queue_item_id="item-1",
3217        duration=180,
3218        title="Track",
3219        artist="Artist",
3220        album="Album",
3221        image_url="image",
3222    )
3223    release_render = asyncio.Event()
3224
3225    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3226        await release_render.wait()
3227        return "/cache/late.jpg"
3228
3229    with (
3230        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3231        patch.object(
3232            stream, "_prepare_artwork", new_callable=AsyncMock, side_effect=_prepare_artwork
3233        ),
3234        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.01),
3235    ):
3236        push = asyncio.create_task(stream.send_metadata(0, metadata))
3237        async with asyncio.timeout(2):
3238            while write_command.await_count == 0:
3239                await asyncio.sleep(0)
3240        # the identity bundle went out without artwork once the budget lapsed
3241        assert stream._metadata_artwork_checksum == ""
3242        release_render.set()
3243        await push
3244
3245    commands = [args.args[0].decode() for args in write_command.await_args_list]
3246    assert "ARTWORKFILE" not in commands[0]
3247    assert commands[0].endswith("ACTION=SENDMETA\n")
3248    assert [command for command in commands if command.startswith("ARTWORK=")] == [
3249        "ARTWORK=/cache/late.jpg\n"
3250    ]
3251    assert stream._metadata_artwork_checksum == "image"
3252
3253
3254@pytest.mark.asyncio
3255async def test_pending_start_interrupts_the_artwork_wait() -> None:
3256    """A pending START releases the bounded artwork wait instead of queueing behind it."""
3257    player = _make_player()
3258    stream = AirPlayStream(player)
3259    stream._cli_proc = _make_cli_proc()
3260    stream._connected.set()
3261    metadata = MagicMock(
3262        queue_item_id="item-1",
3263        duration=180,
3264        title="Track",
3265        artist="Artist",
3266        album="Album",
3267        image_url="image",
3268    )
3269    release_render = asyncio.Event()
3270    render_started = asyncio.Event()
3271
3272    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3273        render_started.set()
3274        await release_render.wait()
3275        return "/cache/late.jpg"
3276
3277    with (
3278        patch.object(
3279            stream,
3280            "_write_cli_command",
3281            new_callable=AsyncMock,
3282            side_effect=_acking_write_cli_command(stream),
3283        ) as write_command,
3284        patch.object(
3285            stream, "_prepare_artwork", new_callable=AsyncMock, side_effect=_prepare_artwork
3286        ),
3287    ):
3288        push = asyncio.create_task(stream.send_metadata(0, metadata))
3289        await render_started.wait()
3290        # the metadata push sits in its render budget holding the lock; the
3291        # START must release that wait instead of losing its anchor lead to it
3292        assert await stream.start(START_UNIX_MS, 0) == START_UNIX_MS
3293        release_render.set()
3294        await push
3295
3296    commands = [args.args[0] for args in write_command.await_args_list]
3297    assert commands[0].endswith("ACTION=SENDMETA\n")
3298    assert "ARTWORKFILE" not in commands[0]
3299    # the START may only queue behind the push's quick pipe writes (the
3300    # progress anchor), never behind the artwork render itself
3301    start_index = next(
3302        index
3303        for index, command in enumerate(commands)
3304        if command.startswith(f"START_UNIX_MS={START_UNIX_MS}")
3305    )
3306    assert start_index <= 2
3307    assert not any("late.jpg" in command for command in commands[:start_index])
3308
3309
3310@pytest.mark.asyncio
3311async def test_track_change_starting_mid_track_sends_a_progress_correction() -> None:
3312    """A track change landing mid-position corrects the timeline after the bundle."""
3313    stream = AirPlayStream(_make_player())
3314    stream._cli_proc = _make_cli_proc()
3315    metadata = MagicMock(
3316        queue_item_id="item-1",
3317        duration=180,
3318        title="Track",
3319        artist="Artist",
3320        album="Album",
3321        image_url="image",
3322    )
3323
3324    with (
3325        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3326        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
3327    ):
3328        await stream.send_metadata(120, metadata)
3329
3330    commands = [args.args[0].decode() for args in write_command.await_args_list]
3331    assert len(commands) == 2
3332    assert commands[0].endswith("ACTION=SENDMETA\n")
3333    # the push reset the device position to zero, so the mid-track start is
3334    # corrected right after
3335    assert commands[1].endswith("PROGRESS=120\n")
3336    assert stream._last_progress_sent == 120
3337
3338
3339@pytest.mark.asyncio
3340async def test_track_change_at_position_zero_still_sends_a_progress_anchor() -> None:
3341    """
3342    A track starting at zero still gets an explicit PROGRESS anchor.
3343
3344    Some receivers gate their rendering on an explicit timeline anchor: a WiiM
3345    Amp mutes a flushed-and-restarted session a couple of minutes in when no
3346    PROGRESS ever follows the metadata push, and un-mutes the instant one
3347    arrives. Relying on SENDMETA's implicit reset to zero is not enough.
3348    """
3349    stream = AirPlayStream(_make_player())
3350    stream._cli_proc = _make_cli_proc()
3351    metadata = MagicMock(
3352        queue_item_id="item-1",
3353        duration=180,
3354        title="Track",
3355        artist="Artist",
3356        album="Album",
3357        image_url="image",
3358    )
3359
3360    with (
3361        patch.object(stream.commands_pipe, "write", new_callable=AsyncMock) as write_command,
3362        patch.object(stream, "_prepare_artwork", new=AsyncMock(return_value="/cache/art.jpg")),
3363    ):
3364        await stream.send_metadata(0, metadata)
3365
3366    commands = [args.args[0].decode() for args in write_command.await_args_list]
3367    assert len(commands) == 2
3368    assert commands[0].endswith("ACTION=SENDMETA\n")
3369    assert commands[1].endswith("PROGRESS=0\n")
3370    assert stream._last_progress_sent == 0
3371
3372
3373@pytest.mark.asyncio
3374async def test_failed_artwork_delivery_is_retried() -> None:
3375    """A dropped ARTWORK command remains pending for the next metadata update."""
3376    stream = AirPlayStream(_make_player())
3377    metadata = MagicMock(
3378        queue_item_id="item-1",
3379        duration=180,
3380        title="Track",
3381        artist="Artist",
3382        album="Album",
3383        image_url="image",
3384    )
3385    artwork_path = "/cache/thumbnails/artwork.jpg"
3386    release_render = asyncio.Event()
3387
3388    async def _prepare_artwork(_image_url: str, _generation: int) -> str:
3389        # the first render misses the bundling budget, so the artwork goes
3390        # out through the stand-alone ARTWORK command
3391        if not release_render.is_set():
3392            await release_render.wait()
3393        return artwork_path
3394
3395    with (
3396        patch.object(
3397            stream,
3398            "_prepare_artwork",
3399            new_callable=AsyncMock,
3400            side_effect=_prepare_artwork,
3401        ) as prepare_artwork,
3402        patch.object(
3403            stream,
3404            "send_cli_command",
3405            new_callable=AsyncMock,
3406            side_effect=[True, False, True],
3407        ) as send_command,
3408        patch("music_assistant.providers.airplay.stream.AIRPLAY_ARTWORK_RENDER_TIMEOUT", 0.01),
3409    ):
3410        first_push = asyncio.create_task(stream.send_metadata(None, metadata))
3411        async with asyncio.timeout(2):
3412            while send_command.await_count == 0:
3413                await asyncio.sleep(0)
3414        release_render.set()
3415        await first_push
3416        assert stream._metadata_artwork_checksum == ""
3417        await stream.send_metadata(None, metadata)
3418
3419    assert prepare_artwork.await_count == 2
3420    assert [args.args[0] for args in send_command.await_args_list].count(
3421        f"ARTWORK={artwork_path}"
3422    ) == 2
3423    assert stream._metadata_artwork_checksum == "image"
3424
3425
3426@pytest.mark.asyncio
3427async def test_text_refinement_keeps_delivered_artwork_settled() -> None:
3428    """A text-only metadata update after delivery does not re-render unchanged art."""
3429    stream = AirPlayStream(_make_player())
3430    metadata = MagicMock(
3431        duration=180,
3432        title="Track",
3433        artist="Artist",
3434        album="Album",
3435        image_url="image",
3436    )
3437    refined = MagicMock(
3438        queue_item_id=metadata.queue_item_id,
3439        duration=180,
3440        title="Track (Remastered)",
3441        artist="Artist",
3442        album="Album",
3443        image_url="image",
3444    )
3445    artwork_path = "/cache/thumbnails/artwork.jpg"
3446
3447    with (
3448        patch.object(
3449            stream,
3450            "_prepare_artwork",
3451            new_callable=AsyncMock,
3452            return_value=artwork_path,
3453        ) as prepare_artwork,
3454        patch.object(
3455            stream,
3456            "send_cli_command",
3457            new_callable=AsyncMock,
3458            return_value=True,
3459        ) as send_command,
3460    ):
3461        await stream.send_metadata(None, metadata)
3462        assert stream._metadata_artwork_checksum == "image"
3463        # the refinement bumps the metadata generation (pending identity
3464        # changed), which before the identity settle re-armed the artwork
3465        await stream.send_metadata(None, refined)
3466
3467    prepare_artwork.assert_awaited_once()
3468    commands = [args.args[0] for args in send_command.await_args_list]
3469    assert sum(f"ARTWORKFILE={artwork_path}\n" in command for command in commands) == 1
3470    assert "ARTWORKFILE" not in commands[-1]
3471    assert "TITLE=Track (Remastered)" in commands[-1]
3472
3473
3474# --- Structured connect failures reported by the binary ---
3475
3476
3477@pytest.mark.asyncio
3478async def test_connect_error_status_line_is_parsed() -> None:
3479    """The machine-readable failure line is captured with all of its fields."""
3480    stream = AirPlayStream(_make_player())
3481
3482    stream._handle_status_line(
3483        '[STATUS] error code=auth_required http=401 detail="RTSP setup rejected"'
3484    )
3485
3486    assert stream._connect_error == CliError("auth_required", 401, "RTSP setup rejected")
3487
3488
3489@pytest.mark.asyncio
3490async def test_started_ack_status_line_parsing() -> None:
3491    """A started ack releases the START wait; a malformed one carries no details."""
3492    stream = AirPlayStream(_make_player())
3493
3494    stream._handle_status_line(
3495        "[STATUS] started requested_unix_ms=1750000000000 at_unix_ms=1750000000004"
3496    )
3497    assert stream._started.is_set()
3498    assert stream._start_ack == (1750000000000, 1750000000004)
3499
3500    stream._started.clear()
3501    stream._start_ack = None
3502    stream._handle_status_line("[STATUS] started requested_unix_ms=garbage at_unix_ms=1")
3503    assert stream._started.is_set()
3504    assert stream._start_ack is None
3505
3506
3507# --- Post-commit anchor verification ---
3508
3509
3510def test_anchor_corrected_status_line_rebases_the_position() -> None:
3511    """A correction carrying a content cut moves the reported-position base by it."""
3512    stream = AirPlayStream(_make_player())
3513    stream._start_position = 12.0
3514
3515    ended = stream._handle_status_line(
3516        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3517        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3518    )
3519
3520    assert ended is False
3521    assert stream._start_position == 12.5
3522
3523
3524def test_anchor_corrected_status_line_logs_a_warning(caplog: pytest.LogCaptureFixture) -> None:
3525    """The correction is logged loudly, including the display name and the delta."""
3526    stream = AirPlayStream(_make_player())
3527
3528    with caplog.at_level(logging.WARNING):
3529        stream._handle_status_line(
3530            "[STATUS] anchor_corrected requested_unix_ms=0 "
3531            "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3532        )
3533
3534    assert "Player A" in caplog.text
3535    assert "+500 ms" in caplog.text
3536
3537
3538def test_anchor_corrected_status_line_tolerates_malformed_line() -> None:
3539    """A malformed anchor_corrected line is dropped instead of raising or rebasing."""
3540    stream = AirPlayStream(_make_player())
3541    stream._start_position = 12.0
3542
3543    ended = stream._handle_status_line("[STATUS] anchor_corrected requested_unix_ms=garbage")
3544
3545    assert ended is False
3546    assert stream._start_position == 12.0
3547
3548
3549def test_content_cut_short_rebases_the_position_and_warns(
3550    caplog: pytest.LogCaptureFixture,
3551) -> None:
3552    """A cut that ended early gives back the ms the correction over-advanced the base by."""
3553    stream = AirPlayStream(_make_player())
3554    stream._start_position = 12.0
3555    stream._handle_status_line(
3556        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3557        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3558    )
3559    assert stream._start_position == 12.5
3560
3561    with caplog.at_level(logging.WARNING):
3562        ended = stream._handle_status_line(
3563            "[STATUS] content_cut requested_ms=500 cut_ms=180 cut_bytes=31752 drain_ms=210"
3564        )
3565
3566    assert ended is False
3567    assert stream._start_position == pytest.approx(12.18)
3568    assert "AirPlay content cut" in caplog.text
3569    assert "Player A" in caplog.text
3570    assert "320 ms short" in caplog.text
3571
3572
3573def test_content_cut_in_full_leaves_the_position_alone(caplog: pytest.LogCaptureFixture) -> None:
3574    """A cut that took what it asked for needs no correction and stays quiet."""
3575    stream = AirPlayStream(_make_player())
3576    stream._start_position = 12.0
3577    stream._handle_status_line(
3578        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3579        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3580    )
3581
3582    caplog.clear()
3583    with caplog.at_level(logging.WARNING):
3584        # a few ms below the request is byte quantization, not a short cut
3585        stream._handle_status_line(
3586            "[STATUS] content_cut requested_ms=500 cut_ms=498 cut_bytes=87887 drain_ms=505"
3587        )
3588
3589    assert stream._start_position == 12.5
3590    assert caplog.text == ""
3591
3592
3593def test_content_cut_after_a_new_anchor_is_not_reconciled() -> None:
3594    """A cut settling after a START must not be taken off that START's absolute base."""
3595    stream = AirPlayStream(_make_player())
3596    stream._start_position = 12.0
3597    stream._handle_status_line(
3598        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3599        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3600    )
3601    stream.rebase_position(30_000)
3602
3603    stream._handle_status_line(
3604        "[STATUS] content_cut requested_ms=500 cut_ms=0 cut_bytes=0 drain_ms=12"
3605    )
3606
3607    assert stream._start_position == 30.0
3608
3609
3610def test_content_cut_status_line_tolerates_malformed_line() -> None:
3611    """A malformed content_cut line is dropped instead of raising or rebasing."""
3612    stream = AirPlayStream(_make_player())
3613    stream._start_position = 12.0
3614    stream._handle_status_line(
3615        "[STATUS] anchor_corrected requested_unix_ms=1750000000000 "
3616        "from_unix_ms=1750000000400 at_unix_ms=1750000000900 content_cut_ms=500"
3617    )
3618
3619    ended = stream._handle_status_line("[STATUS] content_cut requested_ms=500 cut_ms=garbage")
3620
3621    assert ended is False
3622    assert stream._start_position == 12.5
3623
3624
3625def test_clock_verified_status_line_is_debug_logged(caplog: pytest.LogCaptureFixture) -> None:
3626    """A clock_verified line needs no server action beyond a debug note of the margin."""
3627    stream = AirPlayStream(_make_player())
3628
3629    with caplog.at_level(logging.DEBUG):
3630        ended = stream._handle_status_line("[STATUS] clock_verified margin_ms=42")
3631
3632    assert ended is False
3633    assert "42" in caplog.text
3634
3635
3636@pytest.mark.asyncio
3637async def test_connect_error_status_line_tolerates_missing_fields() -> None:
3638    """A failure line without http/detail still yields the reported code."""
3639    stream = AirPlayStream(_make_player())
3640
3641    stream._handle_status_line("[STATUS] error code=connect_failed")
3642
3643    assert stream._connect_error == CliError("connect_failed", 0, "")
3644
3645
3646@pytest.mark.asyncio
3647async def test_auth_required_surfaces_password_required_error() -> None:
3648    """A device asking for a password produces an actionable, translated error."""
3649    stream = AirPlayStream(_make_player())
3650    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
3651    stream._process_ended.set()
3652
3653    with (
3654        patch.object(stream, "_cli_proc", MagicMock()),
3655        pytest.raises(PlayerCommandFailed) as err,
3656    ):
3657        await stream.wait_for_connection()
3658
3659    assert err.value.translation_key == "password_required"
3660
3661
3662@pytest.mark.asyncio
3663async def test_auth_failed_surfaces_authentication_failed_error() -> None:
3664    """A rejected password is reported as an authentication failure, not a timeout."""
3665    stream = AirPlayStream(_make_player())
3666    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
3667    stream._process_ended.set()
3668
3669    with (
3670        patch.object(stream, "_cli_proc", MagicMock()),
3671        pytest.raises(PlayerCommandFailed) as err,
3672    ):
3673        await stream.wait_for_connection()
3674
3675    assert err.value.translation_key == "authentication_failed"
3676
3677
3678@pytest.mark.asyncio
3679@pytest.mark.parametrize("code", ["auth_required", "auth_failed"])
3680async def test_refused_connection_is_not_reported_as_a_password_problem(code: str) -> None:
3681    """A device that turns the handshake away points at pairing, not at a password."""
3682    stream = AirPlayStream(_make_player())
3683    stream._handle_status_line(f'[STATUS] error code={code} http=403 detail="refused"')
3684    stream._process_ended.set()
3685
3686    with (
3687        patch.object(stream, "_cli_proc", MagicMock()),
3688        pytest.raises(PlayerCommandFailed) as err,
3689    ):
3690        await stream.wait_for_connection()
3691
3692    assert err.value.translation_key == "connection_refused"
3693
3694
3695@pytest.mark.asyncio
3696@pytest.mark.parametrize("code", ["auth_required", "auth_failed"])
3697async def test_refused_connection_never_marks_the_password_invalid(code: str) -> None:
3698    """
3699    A refusal must not leave a player demanding a password it may not even have.
3700
3701    tvOS 26 answers the pairing handshake with 403 for reasons unrelated to any
3702    secret, and the marker persists across restarts - so latching it there would
3703    strand the player in a setup flow no password can complete.
3704    """
3705    player = _make_player()
3706    stream = AirPlayStream(player)
3707
3708    stream._handle_status_line(f'[STATUS] error code={code} http=403 detail="refused"')
3709
3710    player.set_password_invalid.assert_not_called()
3711
3712
3713@pytest.mark.asyncio
3714async def test_generic_connect_failure_keeps_the_timeout_semantics() -> None:
3715    """A non-auth failure keeps raising the plain timeout its callers already handle."""
3716    stream = AirPlayStream(_make_player())
3717    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
3718    stream._process_ended.set()
3719
3720    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
3721        await stream.wait_for_connection()
3722
3723
3724@pytest.mark.asyncio
3725async def test_dead_process_fails_the_connect_wait_immediately() -> None:
3726    """
3727    A binary that reports no reason at all leaves the wait its plain timeout error.
3728
3729    It must still end the moment the process is gone instead of running out the
3730    full connect timeout.
3731    """
3732    stream = AirPlayStream(_make_player())
3733    stream._process_ended.set()  # process died without emitting a [STATUS] error line
3734
3735    started = asyncio.get_running_loop().time()
3736    with patch.object(stream, "_cli_proc", MagicMock()), pytest.raises(TimeoutError):
3737        await stream.wait_for_connection()
3738
3739    assert asyncio.get_running_loop().time() - started < 1
3740
3741
3742@pytest.mark.asyncio
3743async def test_auth_failed_marks_the_stored_password_invalid() -> None:
3744    """A rejected password is persisted so the player keeps offering its setup action."""
3745    player = _make_player()
3746    stream = AirPlayStream(player)
3747
3748    stream._handle_status_line('[STATUS] error code=auth_failed http=401 detail="bad password"')
3749
3750    player.set_password_invalid.assert_called_once_with(True)
3751
3752
3753@pytest.mark.asyncio
3754async def test_auth_required_also_marks_the_password_as_needed() -> None:
3755    """
3756    A device that demanded a password we could not supply flips into setup.
3757
3758    Devices can enforce a password without announcing it (stale TXT records), so
3759    the runtime signal must set the marker too - it is the only reliable one.
3760    """
3761    player = _make_player()
3762    stream = AirPlayStream(player)
3763
3764    stream._handle_status_line('[STATUS] error code=auth_required http=401 detail="no password"')
3765
3766    player.set_password_invalid.assert_called_once_with(True)
3767
3768
3769@pytest.mark.asyncio
3770async def test_plain_connect_failures_leave_the_password_marker_alone() -> None:
3771    """A non-authentication failure says nothing about the stored password."""
3772    player = _make_player()
3773    stream = AirPlayStream(player)
3774
3775    stream._handle_status_line('[STATUS] error code=connect_failed http=0 detail="no route"')
3776
3777    player.set_password_invalid.assert_not_called()
3778
3779
3780@pytest.mark.asyncio
3781async def test_successful_connect_clears_the_password_marker() -> None:
3782    """Whatever the device accepted is a working password."""
3783    player = _make_player()
3784    stream = AirPlayStream(player)
3785
3786    stream._handle_status_line("[STATUS] connected")
3787
3788    assert stream.connected is True
3789    player.set_password_invalid.assert_called_once_with(False)
3790
3791
3792# --- Password preflight ---
3793
3794
3795@pytest.mark.asyncio
3796async def test_connect_refuses_password_device_without_password_or_credentials() -> None:
3797    """A password-protected AirPlay 2 device with nothing to authenticate never spawns a process."""
3798    player = _make_player()
3799    player.password_required = True
3800    player.get_setup_value = MagicMock(return_value=None)
3801    stream = AirPlayStream(player)
3802
3803    with pytest.raises(PlayerCommandFailed) as err:
3804        await stream.connect()
3805
3806    assert err.value.translation_key == "password_required"
3807    assert stream._cli_proc is None
3808
3809
3810@pytest.mark.asyncio
3811async def test_password_preflight_passes_with_password_or_credentials() -> None:
3812    """Either a configured password or stored credentials let the connect proceed."""
3813    player = _make_player()
3814    player.password_required = True
3815    player.get_setup_value = MagicMock(return_value=None)
3816    player.config.get_value = MagicMock(
3817        side_effect=lambda key, default=None: "s3cret" if key == CONF_PASSWORD else default
3818    )
3819    AirPlayStream(player)._check_password_preflight()
3820
3821    # credentials alone are enough: the binary's pair-verify leg may still succeed
3822    player.config.get_value = MagicMock(side_effect=lambda _key, default=None: default)
3823    player.get_setup_value = MagicMock(
3824        side_effect=lambda key, default=None: (
3825            "ab" * 96 if key == CONF_AIRPLAY_CREDENTIALS else default
3826        )
3827    )
3828    AirPlayStream(player)._check_password_preflight()
3829
3830
3831@pytest.mark.asyncio
3832async def test_password_preflight_skipped_for_raop() -> None:
3833    """The preflight only guards the native AirPlay 2 flow; RAOP carries its own password."""
3834    player = _make_player()
3835    player.protocol = StreamingProtocol.RAOP
3836    player.password_required = True
3837    player.get_setup_value = MagicMock(return_value=None)
3838
3839    AirPlayStream(player)._check_password_preflight()
3840
3841
3842@pytest.mark.asyncio
3843async def test_accepts_audio_ends_with_the_audio_eof() -> None:
3844    """
3845    A stream that was sent its audio EOF keeps running but takes no more audio.
3846
3847    The EOF closes the binary's stdin for good while the process itself plays out
3848    and exits, so a warm refill has to read this rather than the process state.
3849    """
3850    player = _make_player()
3851    stream = AirPlayStream(player)
3852    cli_proc = AsyncProcess(["cat"], stdin=True, stdout=True)
3853    await cli_proc.start()
3854    stream._cli_proc = cli_proc
3855    try:
3856        assert stream.running
3857        assert stream.accepts_audio
3858
3859        await stream.write_audio_eof()
3860
3861        assert stream.running
3862        assert not stream.accepts_audio
3863    finally:
3864        await cli_proc.close()
3865