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