music-assistant-server

1"""Unit tests for AirPlay stream session late-join logic."""
2
3import asyncio
4import logging
5import time
6from collections.abc import AsyncGenerator
7from typing import Any
8from unittest.mock import AsyncMock, MagicMock, patch
9
10import pytest
11from music_assistant_models.enums import PlaybackState
12from music_assistant_models.errors import PlayerCommandFailed
13
14from music_assistant.providers.airplay.constants import (
15    AIRPLAY_CLOCK_READY_LEAD_MS,
16    AIRPLAY_CLOCK_READY_TIMEOUT_MS,
17    AIRPLAY_COLD_GROUP_START_LEAD_MS,
18    AIRPLAY_GROUP_START_LEAD_MS,
19    AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS,
20    AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS,
21    AIRPLAY_LATE_JOIN_RING_MAX_BYTES,
22    AIRPLAY_LATE_JOIN_RING_MIN_SECONDS,
23    AIRPLAY_SPLICE_LEAD_MARGIN_MS,
24    AIRPLAY_START_LEAD_MS,
25    ClockReadiness,
26    StreamingProtocol,
27)
28from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
29
30PCM_SAMPLE_SIZE = 176400  # 44.1kHz / 16-bit / 2ch
31
32
33async def _ack_commanded_instant(start_unix_ms: int = 0, *_args: Any, **_kwargs: Any) -> int:
34    """Ack a START at exactly the instant it was commanded, as a feasible one is."""
35    return start_unix_ms
36
37
38def _stream_defaults(stream: MagicMock) -> MagicMock:
39    """
40    Apply the verified-start API defaults to a mocked stream.
41
42    Every START is acked at the commanded instant, with no warm-lead constraint
43    and no receiver clock projection, so the tests assert the commanded values
44    directly.
45    """
46    stream.start = AsyncMock(side_effect=_ack_commanded_instant)
47    # on the real stream this follows `running` until the audio EOF is written
48    stream.accepts_audio = bool(stream.running)
49    stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.UNREPORTED, 0))
50    stream.warm_lead_ms = 0
51    stream.flushed_head_unix_ms = 0
52    return stream
53
54
55def _make_session(
56    start_time: float,
57    seconds_streamed: float,
58) -> AirPlayStreamSession:
59    """
60    Create a stream session for testing.
61
62    :param start_time: The wall-clock time when the stream was started.
63    :param seconds_streamed: How many seconds of audio have been streamed.
64    """
65    prov = MagicMock()
66
67    pcm_format = MagicMock()
68    pcm_format.pcm_sample_size = PCM_SAMPLE_SIZE
69    pcm_format.sample_rate = 44100
70    pcm_format.bit_depth = 16
71    pcm_format.channels = 2
72
73    leader = MagicMock()
74    leader.player_id = "leader"
75    leader.protocol = StreamingProtocol.RAOP
76    # a joinable session has a reference member that is actually playing
77    leader.playback_state = PlaybackState.PLAYING
78    leader.stream = _stream_defaults(MagicMock())
79    leader.stream.running = True
80    leader.stream.connected = True
81    leader.stream.wait_audio_present = AsyncMock(return_value=True)
82    leader.stream.cumulative_shift_seconds = 0.0
83    leader.config.get_value = MagicMock(return_value=0)
84
85    session = AirPlayStreamSession(prov, [leader], pcm_format, MagicMock(elapsed_time=0))
86    session.start_time = start_time
87    session.seconds_streamed = seconds_streamed
88    session.start_unix_ms = 1  # dummy
89
90    return session
91
92
93def _make_late_joiner() -> MagicMock:
94    """Create a mock AirPlay player for late-join testing."""
95    player = MagicMock()
96    player.player_id = "late_joiner"
97    player.protocol = StreamingProtocol.RAOP
98    player.stream = None
99    player.config = MagicMock()
100    player.config.get_value = MagicMock(return_value=0)
101    return player
102
103
104def _setup_stream(player: MagicMock) -> Any:
105    """Return a side_effect callable that sets up the stream mock on the player."""
106
107    def _side_effect(*_args: Any, **_kwargs: Any) -> None:
108        player.stream = _stream_defaults(MagicMock())
109        player.stream.running = True
110        player.stream.connected = True
111        player.stream.wait_for_connection = AsyncMock()
112        player.stream.wait_audio_present = AsyncMock(return_value=True)
113        player.stream.flush = AsyncMock(return_value=True)
114        player.stream.cumulative_shift_seconds = 0.0
115
116    return _side_effect
117
118
119def _captured_start_at(player: MagicMock) -> float:
120    """Return the start time passed to the player's START command."""
121    start_unix_ms = player.stream.start.call_args[0][0]
122    assert isinstance(start_unix_ms, int)
123    return start_unix_ms / 1000
124
125
126@pytest.mark.asyncio
127async def test_initial_client_failure_stops_started_clients() -> None:
128    """A partial group startup failure cancels siblings before session teardown."""
129    session = _make_session(time.time(), 0)
130    first_player: Any = session.sync_clients[0]
131    second_player = MagicMock()
132    session.sync_clients.append(second_player)
133    first_started = asyncio.Event()
134    first_cancelled = asyncio.Event()
135
136    async def audio_source() -> AsyncGenerator[bytes]:
137        yield b""
138
139    async def start_client(player: Any, *_args: Any) -> None:
140        if player is first_player:
141            first_started.set()
142            try:
143                await asyncio.Event().wait()
144            except asyncio.CancelledError:
145                first_cancelled.set()
146                raise
147        await first_started.wait()
148        raise OSError("process failed")
149
150    with (
151        patch.object(
152            session,
153            "_start_client",
154            new_callable=AsyncMock,
155            side_effect=start_client,
156        ),
157        patch.object(session, "stop", new_callable=AsyncMock) as stop_session,
158        pytest.raises(PlayerCommandFailed, match="Playback failed to start"),
159    ):
160        await session.start(audio_source())
161
162    assert first_cancelled.is_set()
163    stop_session.assert_awaited_once()
164
165
166@pytest.mark.asyncio
167@pytest.mark.parametrize(
168    ("first_protocol", "second_protocol"),
169    [
170        (StreamingProtocol.RAOP, StreamingProtocol.RAOP),
171        (StreamingProtocol.AIRPLAY2, StreamingProtocol.AIRPLAY2),
172        (StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2),
173    ],
174)
175async def test_initial_group_waits_for_every_member_before_shared_start(
176    first_protocol: StreamingProtocol,
177    second_protocol: StreamingProtocol,
178) -> None:
179    """Every member connects before one shared START anchors the group."""
180    session = _make_session(0, 0)
181    first_player: Any = session.sync_clients[0]
182    first_player.player_id = "first"
183    first_player.protocol = first_protocol
184    first_player.config.get_value = MagicMock(return_value=0)
185    second_player = MagicMock(player_id="second")
186    second_player.protocol = second_protocol
187    second_player.config.get_value = MagicMock(return_value=0)
188    session.sync_clients.append(second_player)
189    session.media.elapsed_time = 12
190    operations: list[str] = []
191
192    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
193        stream = _stream_defaults(MagicMock(running=True))
194
195        async def wait_for_connection() -> None:
196            operations.append(f"connected:{player.player_id}")
197
198        async def start(start_unix_ms: int, position_ms: int) -> int:
199            assert position_ms == 12_000
200            operations.append(f"started:{player.player_id}:{start_unix_ms}")
201            return start_unix_ms
202
203        stream.wait_for_connection = AsyncMock(side_effect=wait_for_connection)
204        stream.wait_audio_present = AsyncMock(return_value=True)
205        stream.start = AsyncMock(side_effect=start)
206        player.stream = stream
207
208    with (
209        patch.object(session, "_start_client", side_effect=start_client),
210        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
211        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
212        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
213    ):
214        await session.start(MagicMock())
215
216    last_connected = max(i for i, op in enumerate(operations) if op.startswith("connected:"))
217    first_start = min(i for i, op in enumerate(operations) if op.startswith("started:"))
218    assert last_connected < first_start
219    # start = now (100_000 ms) + the cold group start lead, one shared instant
220    starts = {int(op.rsplit(":", 1)[1]) for op in operations if op.startswith("started:")}
221    assert starts == {100_000 + AIRPLAY_COLD_GROUP_START_LEAD_MS}
222
223
224@pytest.mark.asyncio
225async def test_group_start_never_anchors_before_every_receiver_clock_is_usable() -> None:
226    """A member whose clock lands past the group lead pushes the shared anchor out."""
227    now = 100.0
228    session = _make_session(0, 0)
229    first: Any = session.sync_clients[0]
230    first.protocol = StreamingProtocol.AIRPLAY2
231    first.config.get_value = MagicMock(return_value=0)
232    second = MagicMock(player_id="second", protocol=StreamingProtocol.AIRPLAY2)
233    second.config.get_value = MagicMock(return_value=0)
234    session.sync_clients.append(second)
235    # The slower member's clock lands past the cold group lead, so anchoring on
236    # that lead alone would leave it seated at a different instant to the first.
237    ready_at = int(now * 1000) + AIRPLAY_COLD_GROUP_START_LEAD_MS + 400
238
239    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
240        stream = _stream_defaults(MagicMock(running=True))
241        stream.wait_for_connection = AsyncMock()
242        stream.wait_audio_present = AsyncMock(return_value=True)
243        stream.wait_clock_ready = AsyncMock(
244            return_value=(
245                ClockReadiness.PROJECTED,
246                ready_at if player is second else int(now * 1000),
247            )
248        )
249        player.stream = stream
250
251    with (
252        patch.object(session, "_start_client", side_effect=start_client),
253        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
254        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
255        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
256    ):
257        await session.start(MagicMock())
258
259    expected = ready_at + AIRPLAY_CLOCK_READY_LEAD_MS
260    for player in (first, second):
261        assert player.stream.start.await_args.args[0] == expected
262
263
264@pytest.mark.asyncio
265async def test_solo_start_waits_for_the_receiver_clock_projection() -> None:
266    """A lone receiver on a cold clock renders silence, so its anchor waits for it too."""
267    now = 100.0
268    ready_at_unix_ms = int(now * 1000) + 5_000
269    session = _make_session(0, 0)
270    player: Any = session.sync_clients[0]
271    player.protocol = StreamingProtocol.AIRPLAY2
272    player.config.get_value = MagicMock(return_value=0)
273
274    async def start_client(_player: MagicMock, _use_shared_ptp: bool) -> None:
275        stream = _stream_defaults(MagicMock(running=True))
276        stream.wait_for_connection = AsyncMock()
277        stream.wait_audio_present = AsyncMock(return_value=True)
278        stream.wait_clock_ready = AsyncMock(
279            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
280        )
281        player.stream = stream
282
283    with (
284        patch.object(session, "_start_client", side_effect=start_client),
285        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
286        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
287        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
288    ):
289        await session.start(MagicMock())
290
291    player.stream.wait_clock_ready.assert_awaited_once()
292    # the projection is further out than the solo lead, so it carries the anchor
293    assert player.stream.start.await_args.args[0] == ready_at_unix_ms + AIRPLAY_CLOCK_READY_LEAD_MS
294
295
296@pytest.mark.asyncio
297async def test_solo_start_with_a_usable_clock_keeps_the_short_lead() -> None:
298    """A warm clock reports ready with a past instant, so waiting for it costs nothing."""
299    now = 100.0
300    session = _make_session(0, 0)
301    player: Any = session.sync_clients[0]
302    player.protocol = StreamingProtocol.AIRPLAY2
303    player.config.get_value = MagicMock(return_value=0)
304
305    async def start_client(_player: MagicMock, _use_shared_ptp: bool) -> None:
306        stream = _stream_defaults(MagicMock(running=True))
307        stream.wait_for_connection = AsyncMock()
308        stream.wait_audio_present = AsyncMock(return_value=True)
309        # already usable: the projection sits a second in the past
310        stream.wait_clock_ready = AsyncMock(
311            return_value=(ClockReadiness.PROJECTED, int(now * 1000) - 1_000)
312        )
313        player.stream = stream
314
315    with (
316        patch.object(session, "_start_client", side_effect=start_client),
317        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
318        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
319        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
320    ):
321        await session.start(MagicMock())
322
323    assert player.stream.start.await_args.args[0] == int(now * 1000) + AIRPLAY_START_LEAD_MS
324
325
326@pytest.mark.asyncio
327async def test_group_start_that_never_converges_anchors_where_members_landed() -> None:
328    """A group that keeps correcting is anchored at the members' own last instant."""
329    session = _make_session(0, 0)
330    first: Any = session.sync_clients[0]
331    first.config.get_value = MagicMock(return_value=0)
332    second = MagicMock(player_id="second")
333    second.protocol = StreamingProtocol.RAOP
334    second.config.get_value = MagicMock(return_value=0)
335    session.sync_clients.append(second)
336    commanded: list[int] = []
337
338    async def correcting_start(start_unix_ms: int, _position_ms: int) -> int:
339        # Correct every round, so the loop exhausts without ever converging.
340        commanded.append(start_unix_ms)
341        return start_unix_ms + 100
342
343    async def honoring_start(start_unix_ms: int, _position_ms: int) -> int:
344        return start_unix_ms
345
346    first.stream = _stream_defaults(MagicMock(running=True))
347    first.stream.start = AsyncMock(side_effect=correcting_start)
348    second.stream = _stream_defaults(MagicMock(running=True))
349    second.stream.start = AsyncMock(side_effect=honoring_start)
350
351    await session._start_members(0, 100_000)
352
353    # The retry after the final round is never commanded, so it must not become
354    # the session anchor: that would map every later joiner behind the group.
355    assert session.start_unix_ms == commanded[-1] + 100
356    assert session.start_unix_ms not in commanded
357
358
359@pytest.mark.asyncio
360async def test_corrected_solo_start_adopts_the_instant_without_reanchoring() -> None:
361    """A corrected solo start is anchored at the binary's instant with no re-START."""
362    session = _make_session(0, 0)
363    player: Any = session.sync_clients[0]
364    player.config.get_value = MagicMock(return_value=0)
365    stream = _stream_defaults(MagicMock(running=True))
366    # The binary corrects the commanded instant forward; a re-START would only
367    # re-base reported position on the raw one, so exactly one START may be
368    # commanded and its corrected ack becomes the anchor.
369    stream.start = AsyncMock(return_value=101_500)
370    player.stream = stream
371
372    await session._start_members(0, 100_000)
373
374    stream.start.assert_awaited_once_with(100_000, 0)
375    assert session.start_unix_ms == 101_500
376
377
378@pytest.mark.asyncio
379async def test_group_start_fails_when_a_member_never_acknowledges() -> None:
380    """An unacknowledged member start fails the session instead of recording its instant."""
381    session = _make_session(0, 0)
382    first_player: Any = session.sync_clients[0]
383    first_player.protocol = StreamingProtocol.RAOP
384    second_player = MagicMock(player_id="silent", protocol=StreamingProtocol.RAOP)
385    second_player.config.get_value = MagicMock(return_value=0)
386    session.sync_clients.append(second_player)
387    anchor_before = (session.start_unix_ms, session.start_time)
388
389    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
390        stream = _stream_defaults(MagicMock(running=True))
391        stream.wait_for_connection = AsyncMock()
392        stream.wait_audio_present = AsyncMock(return_value=True)
393        if player is second_player:
394            stream.start = AsyncMock(
395                side_effect=PlayerCommandFailed(
396                    "AirPlay player Player B did not acknowledge its start within 2.0s"
397                )
398            )
399        player.stream = stream
400
401    with (
402        patch.object(session, "_start_client", side_effect=start_client),
403        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
404        patch.object(session, "stop", new_callable=AsyncMock) as stop_session,
405        pytest.raises(PlayerCommandFailed, match="did not acknowledge its start"),
406    ):
407        await session.start(MagicMock())
408
409    # Nothing may be recorded from a round that had no verified answer: an
410    # anchor the group never played is what every later joiner aligns against.
411    assert (session.start_unix_ms, session.start_time) == anchor_before
412    stop_session.assert_awaited_once()
413
414
415@pytest.mark.asyncio
416@pytest.mark.parametrize(
417    "protocol",
418    [StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2],
419)
420async def test_initial_single_player_starts_after_connect(
421    protocol: StreamingProtocol,
422) -> None:
423    """A standalone player is anchored with a single START once connected."""
424    session = _make_session(0, 0)
425    player: Any = session.sync_clients[0]
426    player.player_id = "solo"
427    player.protocol = protocol
428    player.config.get_value = MagicMock(return_value=0)
429    stream = _stream_defaults(MagicMock(running=True))
430    stream.wait_for_connection = AsyncMock()
431    stream.wait_audio_present = AsyncMock(return_value=True)
432    stream.flush = AsyncMock(return_value=True)
433
434    async def start_client(_player: MagicMock, _use_shared_ptp: bool) -> None:
435        player.stream = stream
436
437    with (
438        patch.object(session, "_start_client", side_effect=start_client),
439        patch.object(session, "_resolve_shared_ptp", new_callable=AsyncMock, return_value=False),
440        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
441        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=200.0),
442    ):
443        await session.start(MagicMock())
444
445    # start = now (200_000 ms) + the solo start lead, position 0
446    stream.start.assert_awaited_once_with(200_000 + AIRPLAY_START_LEAD_MS, 0)
447
448
449@pytest.mark.asyncio
450async def test_initial_connection_failure_never_starts_partial_group() -> None:
451    """If any member fails to connect, no member receives START and the group is stopped."""
452    session = _make_session(0, 0)
453    first_player: Any = session.sync_clients[0]
454    first_player.protocol = StreamingProtocol.RAOP
455    second_player = MagicMock(player_id="second")
456    second_player.protocol = StreamingProtocol.RAOP
457    session.sync_clients.append(second_player)
458
459    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
460        stream = _stream_defaults(MagicMock(running=True))
461        if player is second_player:
462            stream.wait_for_connection = AsyncMock(side_effect=TimeoutError("connect timeout"))
463        else:
464            stream.wait_for_connection = AsyncMock()
465        player.stream = stream
466
467    with (
468        patch.object(session, "_start_client", side_effect=start_client),
469        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
470        patch.object(session, "stop", new_callable=AsyncMock) as stop_session,
471        pytest.raises(PlayerCommandFailed, match="Playback failed to start"),
472    ):
473        await session.start(MagicMock())
474
475    for player in session.sync_clients:
476        stream: Any = player.stream
477        stream.start.assert_not_awaited()
478    stop_session.assert_awaited_once()
479
480
481@pytest.mark.asyncio
482async def test_connection_failure_names_the_member_that_failed(
483    caplog: pytest.LogCaptureFixture,
484) -> None:
485    """The log has to say WHICH speaker failed, and what its binary reported."""
486    session = _make_session(0, 0)
487    prov_logger = logging.getLogger("test.airplay.session")
488    session.prov.logger = prov_logger
489    first_player: Any = session.sync_clients[0]
490    first_player.display_name = "Player A"
491    first_player.protocol = StreamingProtocol.RAOP
492    second_player = MagicMock(player_id="second", display_name="Player B")
493    second_player.protocol = StreamingProtocol.RAOP
494    session.sync_clients.append(second_player)
495
496    async def start_client(player: MagicMock, _use_shared_ptp: bool) -> None:
497        stream = _stream_defaults(MagicMock(running=True))
498        if player is second_player:
499            stream.wait_for_connection = AsyncMock(
500                side_effect=TimeoutError("cliairplay did not connect to Player B: no route")
501            )
502        else:
503            stream.wait_for_connection = AsyncMock()
504        player.stream = stream
505
506    with (
507        caplog.at_level(logging.WARNING),
508        patch.object(session, "_start_client", side_effect=start_client),
509        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
510        patch.object(session, "stop", new_callable=AsyncMock),
511        pytest.raises(PlayerCommandFailed),
512    ):
513        await session.start(MagicMock())
514
515    assert "Player B failed to connect to its device" in caplog.text
516    assert "no route" in caplog.text
517    assert "Player A failed" not in caplog.text
518
519
520@pytest.mark.asyncio
521async def test_unconfirmed_audio_feed_names_the_silent_members() -> None:
522    """A member whose binary never reported audio flowing is named in the failure."""
523    session = _make_session(0, 0)
524    first_player: Any = session.sync_clients[0]
525    first_player.display_name = "Player A"
526    first_player.protocol = StreamingProtocol.RAOP
527    first_player.stream.wait_audio_present = AsyncMock(return_value=True)
528    second_player = MagicMock(player_id="second", display_name="Player B")
529    second_player.protocol = StreamingProtocol.RAOP
530    second_player.stream = _stream_defaults(MagicMock(running=True))
531    second_player.stream.wait_audio_present = AsyncMock(return_value=False)
532    session.sync_clients.append(second_player)
533
534    with pytest.raises(PlayerCommandFailed, match="not confirmed by Player B"):
535        await session._wait_members_audio_present()
536
537
538@pytest.mark.asyncio
539async def test_initial_connection_cancellation_never_starts_group() -> None:
540    """Cancellation while connecting cleans up without anchoring playback."""
541    session = _make_session(0, 0)
542    player: Any = session.sync_clients[0]
543    player.player_id = "solo"
544    player.protocol = StreamingProtocol.RAOP
545    connection_waiting = asyncio.Event()
546    stream = _stream_defaults(MagicMock(running=True))
547
548    async def wait_for_connection() -> None:
549        connection_waiting.set()
550        await asyncio.Event().wait()
551
552    stream.wait_for_connection = AsyncMock(side_effect=wait_for_connection)
553
554    async def start_client(_player: MagicMock, _use_shared_ptp: bool) -> None:
555        player.stream = stream
556
557    with (
558        patch.object(session, "_start_client", side_effect=start_client),
559        patch.object(session, "stop", new_callable=AsyncMock) as stop_session,
560    ):
561        start_task = asyncio.create_task(session.start(MagicMock()))
562        await connection_waiting.wait()
563        start_task.cancel()
564        with pytest.raises(asyncio.CancelledError):
565            await start_task
566
567    stream.start.assert_not_awaited()
568    stop_session.assert_awaited_once()
569
570
571@pytest.mark.parametrize(
572    "protocols",
573    [
574        (StreamingProtocol.RAOP,),
575        (StreamingProtocol.AIRPLAY2,),
576        (StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2),
577    ],
578)
579def test_warm_replace_supports_every_streaming_protocol(
580    protocols: tuple[StreamingProtocol, ...],
581) -> None:
582    """Connected legacy RAOP, AirPlay 2 and mixed sessions can replace warm."""
583    session = _make_session(0, 0)
584    players: Any = []
585    for index, protocol in enumerate(protocols):
586        player = MagicMock()
587        player.player_id = f"player-{index}"
588        player.protocol = protocol
589        player.stream = _stream_defaults(MagicMock(running=True, connected=True))
590        players.append(player)
591    session.sync_clients = players
592
593    assert session.can_replace(players, session.pcm_format)
594
595
596@pytest.mark.asyncio
597@pytest.mark.parametrize(
598    "protocols",
599    [
600        (StreamingProtocol.AIRPLAY2,),
601        (StreamingProtocol.RAOP,),
602        (StreamingProtocol.AIRPLAY2, StreamingProtocol.AIRPLAY2),
603        (StreamingProtocol.RAOP, StreamingProtocol.RAOP),
604        (StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2),
605    ],
606)
607async def test_standby_supports_every_connected_streaming_protocol(
608    protocols: tuple[StreamingProtocol, ...],
609) -> None:
610    """Connected legacy RAOP, AirPlay 2 and mixed sessions can enter standby."""
611    session = _make_session(0, 0)
612    players: Any = []
613    for index, protocol in enumerate(protocols):
614        player = MagicMock()
615        player.player_id = f"player-{index}"
616        player.protocol = protocol
617        player.stream = _stream_defaults(MagicMock(running=True, connected=True))
618        player.stream.send_cli_command = AsyncMock(return_value=True)
619        players.append(player)
620    session.sync_clients = players
621
622    assert await session.standby()
623    # the park is carried by the session, so it survives the group shrinking
624    assert session.parked is True
625    for player in players:
626        player.stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
627        player.set_state_from_stream.assert_called_once_with(
628            state=PlaybackState.PAUSED, stream=player.stream
629        )
630
631
632@pytest.mark.asyncio
633@pytest.mark.parametrize(
634    "protocols",
635    [
636        (StreamingProtocol.RAOP,),
637        (StreamingProtocol.AIRPLAY2,),
638        (StreamingProtocol.RAOP, StreamingProtocol.AIRPLAY2),
639    ],
640)
641async def test_standby_resumes_warm_on_existing_streams(
642    protocols: tuple[StreamingProtocol, ...],
643) -> None:
644    """RAOP, AirPlay 2 and mixed sessions resume warm via flush-refill on parked streams."""
645    session = _make_session(0, 0)
646    players: Any = []
647    original_streams: dict[str, MagicMock] = {}
648    for index, protocol in enumerate(protocols):
649        player = MagicMock()
650        player.player_id = f"player-{index}"
651        player.protocol = protocol
652        player.config.get_value = MagicMock(return_value=0)
653        player.stream = _stream_defaults(MagicMock(running=True, connected=True))
654        player.stream.send_cli_command = AsyncMock(return_value=True)
655        player.stream.wait_audio_present = AsyncMock(return_value=True)
656        player.stream.flush = AsyncMock(return_value=True)
657        players.append(player)
658        original_streams[player.player_id] = player.stream
659    session.sync_clients = players
660
661    assert await session.standby()
662    assert session.can_replace(players, session.pcm_format)
663    media = MagicMock(elapsed_time=10)
664    with (
665        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
666        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
667        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
668    ):
669        assert await session.replace(MagicMock(), media)
670
671    # the re-anchor ends the park, so the members are playing again
672    assert session.parked is False
673    # start = now (100_000 ms) + solo/group start lead, position 10s
674    expected_start = 100_000 + (
675        AIRPLAY_START_LEAD_MS if len(protocols) == 1 else AIRPLAY_GROUP_START_LEAD_MS
676    )
677    for player in players:
678        stream = original_streams[player.player_id]
679        assert player.stream is stream
680        stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
681        stream.flush.assert_awaited_once_with()
682        stream.start.assert_awaited_once_with(expected_start, 10_000)
683
684
685@pytest.mark.asyncio
686async def test_warm_replace_flushes_all_before_starting_any() -> None:
687    """A group flushes every member and awaits all acks before any shared START."""
688    session = _make_session(0, 0)
689    players: list[Any] = []
690    release_delayed = asyncio.Event()
691    delayed_waiting = asyncio.Event()
692
693    for player_id in ("first", "delayed"):
694        player = MagicMock(player_id=player_id, protocol=StreamingProtocol.AIRPLAY2)
695        player.config.get_value = MagicMock(return_value=0)
696        stream = _stream_defaults(MagicMock(running=True, connected=True))
697
698        async def flush(*, current_id: str = player_id) -> bool:
699            if current_id == "delayed":
700                delayed_waiting.set()
701                await release_delayed.wait()
702            return True
703
704        stream.flush = AsyncMock(side_effect=flush)
705        stream.wait_audio_present = AsyncMock(return_value=True)
706        player.stream = stream
707        players.append(player)
708    session.sync_clients = players
709
710    with (
711        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
712        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
713        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
714    ):
715        replace_task = asyncio.create_task(session.replace(MagicMock(), MagicMock(elapsed_time=0)))
716        await delayed_waiting.wait()
717        # one member's flush is still pending: no member may have been started yet
718        for player in players:
719            player.stream.start.assert_not_awaited()
720        release_delayed.set()
721        assert await replace_task
722
723    for player in players:
724        player.stream.flush.assert_awaited_once_with()
725        # start = now (100_000 ms) + the group start lead (500 ms), position 0
726        player.stream.start.assert_awaited_once_with(100_000 + AIRPLAY_GROUP_START_LEAD_MS, 0)
727
728
729@pytest.mark.asyncio
730async def test_warm_replace_stops_old_audio_and_ffmpeg_before_flush() -> None:
731    """The old audio feed and ffmpeg are torn down before any FLUSH is sent."""
732    session = _make_session(0, 0)
733    operations: list[str] = []
734    player: Any = session.sync_clients[0]
735    player.config.get_value = MagicMock(return_value=0)
736    stream = player.stream
737    stream.running = True
738    stream.connected = True
739
740    async def flush() -> bool:
741        operations.append("flush")
742        return True
743
744    stream.flush = AsyncMock(side_effect=flush)
745    old_ffmpeg = MagicMock(closed=False)
746
747    async def kill_ffmpeg() -> None:
748        operations.append("ffmpeg-killed")
749
750    old_ffmpeg.kill = AsyncMock(side_effect=kill_ffmpeg)
751    session._player_ffmpeg[player.player_id] = old_ffmpeg
752
753    async def old_audio() -> None:
754        try:
755            await asyncio.Event().wait()
756        except asyncio.CancelledError:
757            operations.append("audio-cancelled")
758            raise
759
760    session._audio_source_task = asyncio.create_task(old_audio())
761    # let old_audio reach its await so the cancellation lands inside its try/except
762    await asyncio.sleep(0)
763
764    with (
765        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
766        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
767        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
768    ):
769        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0))
770
771    assert operations.index("audio-cancelled") < operations.index("flush")
772    assert operations.index("ffmpeg-killed") < operations.index("flush")
773    old_ffmpeg.kill.assert_awaited_once()
774
775
776@pytest.mark.asyncio
777async def test_warm_replace_flush_failure_falls_back_to_cold() -> None:
778    """A member that never acknowledges its flush makes the whole replace fall back."""
779    session = _make_session(0, 0)
780    players: list[Any] = []
781    for player_id, acked in (("ok", True), ("failed", False)):
782        player = MagicMock(player_id=player_id, protocol=StreamingProtocol.RAOP)
783        player.config.get_value = MagicMock(return_value=0)
784        stream = _stream_defaults(MagicMock(running=True, connected=True))
785        stream.flush = AsyncMock(return_value=acked)
786        player.stream = stream
787        players.append(player)
788    session.sync_clients = players
789
790    with (
791        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock) as start_ffmpeg,
792        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
793    ):
794        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0)) is False
795
796    # no member is started and no fresh ffmpeg is wired once a flush is unacknowledged
797    for player in players:
798        player.stream.start.assert_not_awaited()
799    start_ffmpeg.assert_not_awaited()
800
801
802@pytest.mark.asyncio
803async def test_warm_replace_start_failure_falls_back_to_cold() -> None:
804    """A failed shared START after flush returns False so the caller restarts cold."""
805    session = _make_session(0, 0)
806    player: Any = session.sync_clients[0]
807    player.config.get_value = MagicMock(return_value=0)
808    stream = player.stream
809    stream.running = True
810    stream.connected = True
811    stream.flush = AsyncMock(return_value=True)
812    stream.start = AsyncMock(side_effect=OSError("start failed"))
813
814    with (
815        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
816        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
817        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
818    ):
819        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0)) is False
820
821
822@pytest.mark.parametrize(
823    ("member_specs", "expected_requirement_ms"),
824    [
825        # the largest member requirement wins, wherever that member sits
826        (((4000, 0), (1500, 0)), 4000),
827        (((1500, 0), (4000, 0)), 4000),
828        # a negative sync_adjust widens that member's own requirement by exactly
829        # the amount it moves that member's commanded instant earlier
830        (((4000, 0), (3600, -600)), 4200),
831    ],
832)
833def test_warm_anchor_clears_the_receivers_queued_audio(
834    member_specs: tuple[tuple[int, int], ...],
835    expected_requirement_ms: int,
836) -> None:
837    """
838    A warm re-anchor lands beyond the audio the receivers still have queued.
839
840    A splice-timeline member honors the commanded instant only when it lands
841    past that member's own queued audio, so the shared anchor has to clear the
842    largest member requirement: anchoring short leaves that member behind the
843    group and every warm start pays a corrective round.
844    """
845    session = _make_session(0, 0)
846    members: list[Any] = [session.sync_clients[0]]
847    second = MagicMock(player_id="second", protocol=StreamingProtocol.AIRPLAY2)
848    second.stream = _stream_defaults(MagicMock(running=True, connected=True))
849    session.sync_clients.append(second)
850    members.append(second)
851    for player, (warm_lead_ms, adjust_ms) in zip(members, member_specs, strict=True):
852        player.stream.warm_lead_ms = warm_lead_ms
853        player.config.get_value = MagicMock(return_value=adjust_ms)
854
855    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0):
856        anchor = session._anchor_start_unix_ms(warm=True)
857
858    assert anchor == 100_000 + expected_requirement_ms + AIRPLAY_SPLICE_LEAD_MARGIN_MS
859
860
861def test_warm_anchor_clears_the_head_every_flushed_member_froze() -> None:
862    """
863    A warm re-anchor lands beyond the frozen head the flushed members reported.
864
865    A member renders from its own commanded instant (the anchor plus its
866    sync_adjust), so it is that instant, not the bare anchor, that has to clear
867    the head the member froze at flush, with margin for the command round-trip.
868    """
869    session = _make_session(0, 0)
870    first: Any = session.sync_clients[0]
871    first.stream.flushed_head_unix_ms = 102_000
872    first.config.get_value = MagicMock(return_value=0)
873    second = MagicMock(player_id="second", protocol=StreamingProtocol.AIRPLAY2)
874    second.stream = _stream_defaults(MagicMock(running=True, connected=True))
875    second.stream.flushed_head_unix_ms = 105_000
876    second.config.get_value = MagicMock(return_value=300)
877    session.sync_clients.append(second)
878
879    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0):
880        anchor = session._anchor_start_unix_ms(warm=True)
881
882    # the later head wins, and the offset its member adds comes back out of it
883    assert anchor == 105_000 - 300 + AIRPLAY_SPLICE_LEAD_MARGIN_MS
884
885
886@pytest.mark.asyncio
887async def test_start_player_ffmpeg_wires_persistent_cli_stdin() -> None:
888    """The per-seek ffmpeg is wired to the member's persistent cli stdin fd and tracked."""
889    session = _make_session(0, 0)
890    player: Any = session.sync_clients[0]
891    old_ffmpeg = MagicMock()
892    old_ffmpeg.close = AsyncMock()
893    session._player_ffmpeg[player.player_id] = old_ffmpeg
894
895    stream = _stream_defaults(MagicMock())
896    stream.pcm_format = session.pcm_format
897    cli_proc = MagicMock()
898    cli_proc.proc.stdin.transport.get_extra_info.return_value.fileno.return_value = 77
899    stream._cli_proc = cli_proc
900    player.stream = stream
901    new_ffmpeg = MagicMock()
902    new_ffmpeg.start = AsyncMock(return_value=None)
903
904    with (
905        patch(
906            "music_assistant.providers.airplay.stream_session.get_final_output_format",
907            return_value=MagicMock(),
908        ),
909        patch(
910            "music_assistant.providers.airplay.stream_session.get_media_session_id",
911            return_value="session-id",
912        ),
913        patch(
914            "music_assistant.providers.airplay.stream_session.FFMpeg", return_value=new_ffmpeg
915        ) as ffmpeg_factory,
916    ):
917        await session._start_player_ffmpeg(player, MagicMock())
918
919    # the old ffmpeg is closed, never killing the shared cli stdin
920    old_ffmpeg.close.assert_awaited_once()
921    # the fresh ffmpeg writes into the cli process stdin fd (77)
922    assert ffmpeg_factory.call_args.kwargs["audio_output"] == 77
923    new_ffmpeg.start.assert_awaited_once()
924    assert session._player_ffmpeg[player.player_id] is new_ffmpeg
925
926
927@pytest.mark.asyncio
928@pytest.mark.parametrize(
929    ("transitioning", "live_session_id", "ends_stream"),
930    [
931        # a seek/next: the queue rotates its session while it loads the new stream
932        (True, "session-2", False),
933        # the queue played out, or the flow broke off for the queue to restart once
934        # the player reports idle - which only the audio EOF can bring about
935        (False, "session-1", True),
936        # not this session's transition (the queue is starting the one we play)
937        (True, "session-1", True),
938        # a transition that already finished cannot be waiting on this session
939        (False, "session-2", True),
940    ],
941)
942async def test_source_end_only_keeps_stdin_open_for_a_pending_replacement(
943    transitioning: bool, live_session_id: str, ends_stream: bool
944) -> None:
945    """
946    The binary's stdin is closed unless the queue is mid-handover to a new stream.
947
948    Closing it ends the stream for good - it cannot be reopened - so a seek would
949    be left with a cold restart as its only option. Everywhere else the EOF is
950    what makes the binary end and the player report idle.
951    """
952    session = _make_session(0, 0)
953    player: Any = session.sync_clients[0]
954    player.stream.write_audio_eof = AsyncMock()
955    session.media = MagicMock(source_id="queue-1", queue_session_id="session-1")
956    queues: Any = session.mass.player_queues
957    queues.queue_data_or_none = MagicMock(
958        return_value=MagicMock(session_id=live_session_id, transitioning=transitioning)
959    )
960    ffmpeg = MagicMock(closed=False)
961    ffmpeg.write_eof = AsyncMock()
962    ffmpeg.wait_with_timeout = AsyncMock()
963    ffmpeg.kill = AsyncMock()
964    session._player_ffmpeg[player.player_id] = ffmpeg
965
966    no_chunks: list[bytes] = []
967
968    async def exhausted_source() -> AsyncGenerator[bytes]:
969        for chunk in no_chunks:
970            yield chunk
971
972    await session._audio_streamer(exhausted_source())
973
974    assert player.player_id not in session._player_ffmpeg
975    assert player.stream.write_audio_eof.await_count == (1 if ends_stream else 0)
976    if ends_stream:
977        # the audio ffmpeg still holds is handed over before the binary is told
978        ffmpeg.write_eof.assert_awaited_once()
979        ffmpeg.wait_with_timeout.assert_awaited_once()
980        ffmpeg.kill.assert_not_awaited()
981    else:
982        # the replacement flushes that audio away, and nothing may reach the
983        # binary between the old ffmpeg dying and that flush
984        ffmpeg.kill.assert_awaited_once()
985        ffmpeg.write_eof.assert_not_awaited()
986
987
988@pytest.mark.asyncio
989@pytest.mark.parametrize("stream_state", ["missing", "stopped", "disconnected", "audio_ended"])
990async def test_standby_requires_every_member_running_and_connected(stream_state: str) -> None:
991    """Standby is unavailable when any member lacks a reusable connected session."""
992    session = _make_session(0, 0)
993    ready_player = MagicMock(player_id="ready", protocol=StreamingProtocol.AIRPLAY2)
994    ready_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
995    ready_player.stream.send_cli_command = AsyncMock()
996    unavailable_player = MagicMock(player_id="unavailable", protocol=StreamingProtocol.RAOP)
997    unavailable_player.stream = None
998    if stream_state != "missing":
999        unavailable_player.stream = MagicMock(
1000            running=stream_state != "stopped",
1001            # a stream that was sent its audio EOF is still running, but its
1002            # stdin is closed for good so it can never be refilled
1003            accepts_audio=stream_state not in ("stopped", "audio_ended"),
1004            connected=stream_state != "disconnected",
1005        )
1006        unavailable_player.stream.send_cli_command = AsyncMock()
1007    players: Any = [ready_player, unavailable_player]
1008    session.sync_clients = players
1009
1010    assert await session.standby() is False
1011    assert session.can_replace(players, session.pcm_format) is False
1012    ready_player.stream.send_cli_command.assert_not_awaited()
1013    ready_player.set_state_from_stream.assert_not_called()
1014    if unavailable_player.stream:
1015        unavailable_player.stream.send_cli_command.assert_not_awaited()
1016        unavailable_player.set_state_from_stream.assert_not_called()
1017
1018
1019@pytest.mark.asyncio
1020async def test_standby_returns_false_when_command_is_not_delivered() -> None:
1021    """Standby fails without changing state for a member that misses the command."""
1022    session = _make_session(0, 0)
1023    logger = MagicMock()
1024    session.prov.logger = logger
1025    delivered_player = MagicMock(player_id="delivered", protocol=StreamingProtocol.AIRPLAY2)
1026    delivered_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1027    delivered_player.stream.send_cli_command = AsyncMock(return_value=True)
1028    dropped_player = MagicMock(player_id="dropped", protocol=StreamingProtocol.RAOP)
1029    dropped_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1030    dropped_player.stream.send_cli_command = AsyncMock(return_value=False)
1031    pending_player = MagicMock(player_id="pending", protocol=StreamingProtocol.AIRPLAY2)
1032    pending_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1033    pending_player.stream.send_cli_command = AsyncMock(return_value=True)
1034    session.sync_clients = [delivered_player, dropped_player, pending_player]
1035
1036    assert await session.standby() is False
1037    delivered_player.stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
1038    delivered_player.set_state_from_stream.assert_called_once_with(
1039        state=PlaybackState.PAUSED, stream=delivered_player.stream
1040    )
1041    dropped_player.stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
1042    dropped_player.set_state_from_stream.assert_not_called()
1043    pending_player.stream.send_cli_command.assert_not_awaited()
1044    pending_player.set_state_from_stream.assert_not_called()
1045    logger.warning.assert_called_once()
1046
1047
1048@pytest.mark.asyncio
1049async def test_standby_returns_false_when_command_raises() -> None:
1050    """Standby fails without changing state when command delivery raises."""
1051    session = _make_session(0, 0)
1052    logger = MagicMock()
1053    session.prov.logger = logger
1054    player = MagicMock(player_id="failed", protocol=StreamingProtocol.AIRPLAY2)
1055    player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1056    player.stream.send_cli_command = AsyncMock(side_effect=OSError("command pipe failed"))
1057    session.sync_clients = [player]
1058
1059    assert await session.standby() is False
1060    player.set_state_from_stream.assert_not_called()
1061    logger.warning.assert_called_once()
1062
1063
1064@pytest.mark.asyncio
1065async def test_late_join_empty_buffer() -> None:
1066    """Test that with an empty buffer, start_at = start_time + seconds_streamed."""
1067    now = time.time()
1068    start_time = now - 8
1069    seconds_streamed = 12.5
1070    session = _make_session(start_time, seconds_streamed)
1071    player = _make_late_joiner()
1072
1073    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1074        mock_start.side_effect = _setup_stream(player)
1075        await session.add_client(player)
1076
1077    assert mock_start.called, "_start_client was never called"
1078    expected = start_time + seconds_streamed
1079    assert abs(_captured_start_at(player) - expected) < 0.1
1080
1081
1082@pytest.mark.asyncio
1083async def test_late_join_caps_prime_at_whole_ring_when_position_predates_it() -> None:
1084    """A due position older than the ring caps the prime at the whole buffer."""
1085    now = 1_000_000.0
1086    # Synthetic anchor in the future forces the due position behind the oldest
1087    # buffered sample: only the whole ring can prime and the anchor is pulled
1088    # to the ring's first sample so content and anchor stay exactly aligned.
1089    seconds_streamed = 12.5
1090    buffer_seconds = 3
1091    start_time = now + 5.0
1092    session = _make_session(start_time, seconds_streamed)
1093    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * buffer_seconds)
1094    player = _make_late_joiner()
1095
1096    written_chunks: list[bytes] = []
1097
1098    async def capture_write(_player: Any, chunk: bytes) -> None:
1099        written_chunks.append(chunk)
1100
1101    with (
1102        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1103        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1104        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1105    ):
1106        mock_start.side_effect = _setup_stream(player)
1107        await session.add_client(player)
1108
1109    assert mock_start.called, "_start_client was never called"
1110    # the whole ring is primed, nothing is skipped, and the anchor maps to the
1111    # ring's first sample: start_at = start_time + (seconds_streamed - buffer)
1112    assert session._client_skip_bytes[player.player_id] == 0
1113    assert written_chunks, "expected the whole ring to be primed"
1114    assert len(written_chunks[0]) / PCM_SAMPLE_SIZE == pytest.approx(buffer_seconds, abs=0.01)
1115    expected = start_time + (seconds_streamed - buffer_seconds)
1116    assert _captured_start_at(player) == pytest.approx(expected, abs=0.02)
1117
1118
1119@pytest.mark.asyncio
1120async def test_late_join_adds_to_sync_clients() -> None:
1121    """Test that the late joiner is added to sync_clients."""
1122    now = time.time()
1123    session = _make_session(now - 10, 12.5)
1124    player = _make_late_joiner()
1125
1126    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1127        mock_start.side_effect = _setup_stream(player)
1128        await session.add_client(player)
1129
1130    assert player in session.sync_clients
1131
1132
1133@pytest.mark.asyncio
1134async def test_late_join_start_failure_stops_client() -> None:
1135    """A late joiner whose START fails is torn down before joining the session."""
1136    session = _make_session(time.time() - 10, 12.5)
1137    player = _make_late_joiner()
1138
1139    def setup_failing_start(*_args: Any, **_kwargs: Any) -> None:
1140        _setup_stream(player)()
1141        player.stream.start = AsyncMock(side_effect=OSError("start failed"))
1142
1143    with (
1144        patch.object(session, "_start_client", side_effect=setup_failing_start),
1145        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1146    ):
1147        await session.add_client(player)
1148
1149    player.stream.start.assert_awaited_once()
1150    # START precedes the prime feed and the sync_clients append, so a failed
1151    # joiner is stopped without ever having joined the session.
1152    assert player not in session.sync_clients
1153    stop_client.assert_awaited_once_with(player, reason="late joiner start/prime failed")
1154
1155
1156@pytest.mark.asyncio
1157async def test_late_join_unacknowledged_start_stops_client() -> None:
1158    """A joiner whose START is never acked is torn down, never mapped onto that instant."""
1159    session = _make_session(time.time() - 10, 12.5)
1160    player = _make_late_joiner()
1161
1162    def setup_unacknowledged_start(*_args: Any, **_kwargs: Any) -> None:
1163        _setup_stream(player)()
1164        player.stream.start = AsyncMock(
1165            side_effect=PlayerCommandFailed(
1166                "AirPlay player Player B did not acknowledge its start within 5.0s"
1167            )
1168        )
1169
1170    with (
1171        patch.object(session, "_start_client", side_effect=setup_unacknowledged_start),
1172        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1173    ):
1174        await session.add_client(player)
1175
1176    assert player not in session.sync_clients
1177    assert player.player_id not in session._client_skip_bytes
1178    player.stream.rebase_position.assert_not_called()
1179    stop_client.assert_awaited_once_with(player, reason="late joiner start/prime failed")
1180
1181
1182@pytest.mark.asyncio
1183async def test_late_join_refuses_a_parked_session() -> None:
1184    """A parked (standby) session has no live timeline, so it cannot absorb a joiner."""
1185    session = _make_session(time.time() - 10, 12.5)
1186    reference: Any = session.sync_clients[0]
1187    reference.stream.send_cli_command = AsyncMock(return_value=True)
1188    reference.set_state_from_stream = MagicMock(
1189        side_effect=lambda **kwargs: setattr(reference, "playback_state", kwargs["state"])
1190    )
1191    assert await session.standby()
1192    assert reference.playback_state == PlaybackState.PAUSED
1193    player = _make_late_joiner()
1194
1195    with (
1196        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1197        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1198    ):
1199        await session.add_client(player)
1200
1201    # The parked session zeroed seconds_streamed while start_time stayed put, so
1202    # anchoring here maps the joiner onto a timeline nothing is playing.
1203    mock_start.assert_not_called()
1204    stop_client.assert_not_awaited()
1205    assert player not in session.sync_clients
1206
1207
1208@pytest.mark.asyncio
1209async def test_late_join_no_running_session() -> None:
1210    """Test that add_client is a no-op when no session is running."""
1211    now = time.time()
1212    session = _make_session(now - 10, 12.5)
1213    # Make the leader's stream not running
1214    leader = session.sync_clients[0]
1215    leader.stream = _stream_defaults(MagicMock())
1216    leader.stream.running = False
1217    player = _make_late_joiner()
1218
1219    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1220        await session.add_client(player)
1221        mock_start.assert_not_called()
1222        assert player not in session.sync_clients
1223
1224
1225@pytest.mark.asyncio
1226async def test_late_join_primes_from_ring_tail_at_headroom() -> None:
1227    """A due position inside the ring primes from the tail and anchors at now + headroom."""
1228    # Freeze time so both the test and the code under test agree on `now`.
1229    now = 1_000_000.0
1230    start_time = now - 0.5
1231    seconds_streamed = 5.0
1232    session = _make_session(start_time, seconds_streamed)
1233    # Fill ring buffer with 5 seconds of non-silent PCM.
1234    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1235    player = _make_late_joiner()
1236
1237    written_chunks: list[bytes] = []
1238
1239    async def capture_write(_player: Any, chunk: bytes) -> None:
1240        written_chunks.append(chunk)
1241
1242    with (
1243        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1244        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1245        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1246    ):
1247        mock_start.side_effect = _setup_stream(player)
1248        await session.add_client(player)
1249
1250    # start_at is now + min_headroom (the late-join floor); fed_pos_due = 2.0s
1251    assert mock_start.called, "_start_client was never called"
1252    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1253    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01), (
1254        f"start_at should be at now + min_headroom, "
1255        f"got offset {_captured_start_at(player) - now:.4f}s"
1256    )
1257
1258    # The last 3.0s of the ring is primed (positions 2.0s..5.0s), nothing skipped.
1259    assert session._client_skip_bytes[player.player_id] == 0
1260    assert written_chunks, "No data was written to the player"
1261    remaining_seconds = len(written_chunks[0]) / PCM_SAMPLE_SIZE
1262    expected_primed = seconds_streamed - (expected_headroom + (now - start_time))
1263    assert remaining_seconds == pytest.approx(expected_primed, abs=0.01), (
1264        f"expected {expected_primed:.2f}s primed, got {remaining_seconds:.4f}s"
1265    )
1266
1267
1268@pytest.mark.asyncio
1269async def test_late_join_skips_live_feed_when_anchor_ahead_of_write_head() -> None:
1270    """When the due position is ahead of the write head, skip that many live bytes."""
1271    # Freeze time so both the test and the code under test agree on `now`.
1272    now = 1_000_000.0
1273    # Diagnosed clamp case: now - start_time = 8.84s, seconds_streamed = 10.0s,
1274    # min_headroom = 2.5s (the late-join floor, no readiness projection) and no
1275    # group shift, so the anchor is due at 11.34s of feed, past the 10.0s write
1276    # head.
1277    start_time = now - 8.84
1278    seconds_streamed = 10.0
1279    session = _make_session(start_time, seconds_streamed)
1280    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 10)
1281    player = _make_late_joiner()
1282
1283    written_chunks: list[bytes] = []
1284
1285    async def capture_write(_player: Any, chunk: bytes) -> None:
1286        written_chunks.append(chunk)
1287
1288    with (
1289        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1290        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1291        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1292    ):
1293        mock_start.side_effect = _setup_stream(player)
1294        await session.add_client(player)
1295
1296    # anchor is now + min_headroom and nothing is primed (position past the head)
1297    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1298    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1299    assert written_chunks == []
1300    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1301    assert skip_seconds == pytest.approx(1.34, abs=0.01)
1302
1303
1304@pytest.mark.asyncio
1305async def test_late_join_primes_from_ring_under_group_shift() -> None:
1306    """A reference member that re-anchored later pulls the due position back into the ring."""
1307    # Freeze time so both the test and the code under test agree on `now`.
1308    now = 1_000_000.0
1309    # Same base as the clamp case, but the reference member accumulated a
1310    # +3.039s starvation shift (134020 frames @44100) so the group's effective
1311    # anchor is later: fed_pos_due = 8.30s, back inside the ring. The joiner is
1312    # primed with ~1.7s from the ring tail and skips nothing.
1313    start_time = now - 8.84
1314    seconds_streamed = 10.0
1315    session = _make_session(start_time, seconds_streamed)
1316    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 10)
1317    reference: Any = session.sync_clients[0]
1318    reference.stream.cumulative_shift_seconds = 134020 / 44100
1319    player = _make_late_joiner()
1320
1321    written_chunks: list[bytes] = []
1322
1323    async def capture_write(_player: Any, chunk: bytes) -> None:
1324        written_chunks.append(chunk)
1325
1326    with (
1327        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1328        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1329        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1330    ):
1331        mock_start.side_effect = _setup_stream(player)
1332        await session.add_client(player)
1333
1334    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1335    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1336    assert session._client_skip_bytes[player.player_id] == 0
1337    assert written_chunks, "expected a prime write from the ring tail"
1338    primed_seconds = len(written_chunks[0]) / PCM_SAMPLE_SIZE
1339    assert primed_seconds == pytest.approx(1.699, abs=0.01)
1340
1341
1342@pytest.mark.asyncio
1343async def test_late_join_anchors_on_the_reported_clock_readiness(
1344    caplog: pytest.LogCaptureFixture,
1345) -> None:
1346    """A projected readiness instant anchors the join, just past the receiver's clock."""
1347    # Freeze time so both the test and the code under test agree on `now`.
1348    now = 1_000_000.0
1349    session = _make_session(now - 5.0, 5.0)
1350    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1351    session.prov.logger = logging.getLogger("test.airplay.session")
1352    player = _make_late_joiner()
1353    # A cold receiver: its clock is projected usable 3.0s out, well past the floor.
1354    ready_at_unix_ms = int((now + 3.0) * 1000)
1355
1356    def setup_with_projection(*_args: Any, **_kwargs: Any) -> None:
1357        _setup_stream(player)()
1358        player.stream.wait_clock_ready = AsyncMock(
1359            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
1360        )
1361
1362    with (
1363        caplog.at_level(logging.DEBUG),
1364        patch.object(session, "_start_client", side_effect=setup_with_projection),
1365        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1366        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1367    ):
1368        await session.add_client(player)
1369
1370    expected_lead = AIRPLAY_CLOCK_READY_LEAD_MS / 1000
1371    assert _captured_start_at(player) - now == pytest.approx(3.0 + expected_lead, abs=0.01)
1372    player.stream.wait_clock_ready.assert_awaited_once_with(
1373        timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
1374    )
1375    assert "receiver clock usable in 3.00s; anchoring no earlier than that" in caplog.text
1376
1377
1378@pytest.mark.asyncio
1379async def test_late_join_floor_wins_over_a_clock_that_is_already_ready(
1380    caplog: pytest.LogCaptureFixture,
1381) -> None:
1382    """A receiver whose clock is already locked still gets the join floor as its anchor."""
1383    # Freeze time so both the test and the code under test agree on `now`.
1384    now = 1_000_000.0
1385    session = _make_session(now - 5.0, 5.0)
1386    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1387    session.prov.logger = logging.getLogger("test.airplay.session")
1388    player = _make_late_joiner()
1389    # A warm receiver reports a readiness instant that has already passed.
1390    ready_at_unix_ms = int((now - 1.0) * 1000)
1391
1392    def setup_with_projection(*_args: Any, **_kwargs: Any) -> None:
1393        _setup_stream(player)()
1394        player.stream.wait_clock_ready = AsyncMock(
1395            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
1396        )
1397
1398    with (
1399        caplog.at_level(logging.DEBUG),
1400        patch.object(session, "_start_client", side_effect=setup_with_projection),
1401        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1402        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1403    ):
1404        await session.add_client(player)
1405
1406    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1407    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1408    assert "receiver clock became usable 1.00s ago; anchoring on the join floor" in caplog.text
1409
1410
1411@pytest.mark.asyncio
1412async def test_late_join_falls_back_to_the_floor_without_a_clock_projection() -> None:
1413    """No projection (NTP timing or a silent receiver) anchors on the floor."""
1414    # Freeze time so both the test and the code under test agree on `now`.
1415    now = 1_000_000.0
1416    session = _make_session(now - 5.0, 5.0)
1417    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1418    player = _make_late_joiner()
1419
1420    with (
1421        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1422        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1423        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1424    ):
1425        mock_start.side_effect = _setup_stream(player)
1426        await session.add_client(player)
1427
1428    # every fallback shape surfaces as "no projection" to the session
1429    player.stream.wait_clock_ready.assert_awaited_once_with(
1430        timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
1431    )
1432    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1433    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1434    assert player in session.sync_clients
1435
1436
1437@pytest.mark.asyncio
1438@pytest.mark.parametrize(
1439    "readiness",
1440    [ClockReadiness.UNREPORTED, ClockReadiness.NOT_APPLICABLE],
1441    ids=["unreported", "ntp"],
1442)
1443async def test_late_join_without_a_projection_still_joins(readiness: ClockReadiness) -> None:
1444    """A device with no clock to wait for is a fallback, not a reason to refuse it."""
1445    now = 1_000_000.0
1446    session = _make_session(now - 5.0, 5.0)
1447    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1448    player = _make_late_joiner()
1449
1450    def setup_without_projection(*_args: Any, **_kwargs: Any) -> None:
1451        _setup_stream(player)()
1452        player.stream.wait_clock_ready = AsyncMock(return_value=(readiness, 0))
1453
1454    with (
1455        patch.object(session, "_start_client", side_effect=setup_without_projection),
1456        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1457        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1458    ):
1459        await session.add_client(player)
1460
1461    assert player in session.sync_clients
1462
1463
1464@pytest.mark.asyncio
1465async def test_late_joiner_with_a_stalled_clock_is_not_added() -> None:
1466    """A receiver that never answered our clock renders silence, so keep it out of the group."""
1467    now = 1_000_000.0
1468    session = _make_session(now - 5.0, 5.0)
1469    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1470    player = _make_late_joiner()
1471
1472    def setup_stalled(*_args: Any, **_kwargs: Any) -> None:
1473        _setup_stream(player)()
1474        player.stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.STALLED, 0))
1475
1476    with (
1477        patch.object(session, "_start_client", side_effect=setup_stalled),
1478        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1479        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1480        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1481    ):
1482        await session.add_client(player)
1483
1484    player.stream.start.assert_not_awaited()
1485    assert player not in session.sync_clients
1486    stop_client.assert_awaited_once_with(player, reason="receiver clock stalled")
1487
1488
1489@pytest.mark.asyncio
1490async def test_late_join_feed_keeps_flowing_while_waiting_for_clock_readiness() -> None:
1491    """The group keeps being fed while a joiner's receiver clock projection is pending."""
1492    session = _make_session(time.time() - 5, 5.0)
1493    player = _make_late_joiner()
1494    readiness_pending = asyncio.Event()
1495    readiness_released = asyncio.Event()
1496
1497    def setup_pending_projection(*_args: Any, **_kwargs: Any) -> None:
1498        _setup_stream(player)()
1499
1500        async def wait_clock_ready(*_args: Any, **_kwargs: Any) -> tuple[ClockReadiness, int]:
1501            readiness_pending.set()
1502            await readiness_released.wait()
1503            return (ClockReadiness.UNREPORTED, 0)
1504
1505        player.stream.wait_clock_ready = AsyncMock(side_effect=wait_clock_ready)
1506
1507    with (
1508        patch.object(session, "_start_client", side_effect=setup_pending_projection),
1509        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1510    ):
1511        join = asyncio.create_task(session.add_client(player))
1512        await asyncio.wait_for(readiness_pending.wait(), timeout=5)
1513        assert await asyncio.wait_for(
1514            session._write_chunk_to_all_players(b"\x02" * PCM_SAMPLE_SIZE), timeout=5
1515        )
1516        readiness_released.set()
1517        await asyncio.wait_for(join, timeout=5)
1518
1519    assert session.seconds_streamed == pytest.approx(6.0)
1520    assert player in session.sync_clients
1521
1522
1523@pytest.mark.asyncio
1524async def test_late_join_feed_keeps_flowing_while_start_ack_is_outstanding() -> None:
1525    """The group keeps being fed while a join's START ack is outstanding."""
1526    # Freeze time so both the test and the code under test agree on `now`.
1527    now = 1_000_000.0
1528    start_time = now - 5.0
1529    session = _make_session(start_time, 5.0)
1530    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1531    player = _make_late_joiner()
1532    ack_outstanding = asyncio.Event()
1533    ack_released = asyncio.Event()
1534
1535    def setup_deferred_ack(*_args: Any, **_kwargs: Any) -> None:
1536        _setup_stream(player)()
1537
1538        async def start(start_unix_ms: int, *_args: Any, **_kwargs: Any) -> int:
1539            # the binary holds its ack until the receiver clock is verified
1540            ack_outstanding.set()
1541            await ack_released.wait()
1542            return start_unix_ms
1543
1544        player.stream.start = AsyncMock(side_effect=start)
1545
1546    writes: list[tuple[str, int]] = []
1547
1548    async def capture_write(target: Any, chunk: bytes) -> None:
1549        writes.append((target.player_id, len(chunk)))
1550
1551    with (
1552        patch.object(session, "_start_client", side_effect=setup_deferred_ack),
1553        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1554        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1555    ):
1556        join = asyncio.create_task(session.add_client(player))
1557        await asyncio.wait_for(ack_outstanding.wait(), timeout=5)
1558        assert await asyncio.wait_for(
1559            session._write_chunk_to_all_players(b"\x02" * PCM_SAMPLE_SIZE), timeout=5
1560        )
1561        ack_released.set()
1562        await asyncio.wait_for(join, timeout=5)
1563
1564    # that second of feed reached the leader and moved the write head to 6.0s
1565    assert ("leader", PCM_SAMPLE_SIZE) in writes
1566    assert session.seconds_streamed == pytest.approx(6.0)
1567    # the anchor (now + the 2.5s floor) is due at 7.5s of feed, so the joiner
1568    # skips only the 1.5s still to come, not the 2.5s due at the commanded
1569    # mapping: the content is mapped against the head the feed actually reached
1570    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1571    assert skip_seconds == pytest.approx(1.5, abs=0.01)
1572    assert player in session.sync_clients
1573
1574
1575@pytest.mark.asyncio
1576async def test_late_join_cancelled_while_ack_outstanding_stops_the_client() -> None:
1577    """A join cancelled while its START ack is outstanding never half-joins the session."""
1578    session = _make_session(time.time() - 5, 5.0)
1579    player = _make_late_joiner()
1580    ack_outstanding = asyncio.Event()
1581
1582    def setup_pending_ack(*_args: Any, **_kwargs: Any) -> None:
1583        _setup_stream(player)()
1584
1585        async def start(*_args: Any, **_kwargs: Any) -> None:
1586            ack_outstanding.set()
1587            await asyncio.Event().wait()
1588
1589        player.stream.start = AsyncMock(side_effect=start)
1590
1591    with (
1592        patch.object(session, "_start_client", side_effect=setup_pending_ack),
1593        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1594    ):
1595        join = asyncio.create_task(session.add_client(player))
1596        await asyncio.wait_for(ack_outstanding.wait(), timeout=5)
1597        join.cancel()
1598        with pytest.raises(asyncio.CancelledError):
1599            await join
1600
1601    assert player not in session.sync_clients
1602    stop_client.assert_awaited_once_with(player, reason="late joiner start cancelled")
1603
1604
1605@pytest.mark.asyncio
1606async def test_late_join_maps_content_from_the_acked_instant() -> None:
1607    """A binary that acks later than commanded gets its content mapped to the acked instant."""
1608    # Freeze time so both the test and the code under test agree on `now`.
1609    now = 1_000_000.0
1610    start_time = now - 5.0
1611    session = _make_session(start_time, 5.0)
1612    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1613    player = _make_late_joiner()
1614    # A sync_adjust rides on top of the commanded instant, so it has to be taken
1615    # back out of the ack before the content is mapped onto it.
1616    adjust_ms = 200
1617    player.config.get_value = MagicMock(return_value=adjust_ms)
1618    deferral_ms = 1500
1619
1620    def setup_deferred_ack(*_args: Any, **_kwargs: Any) -> None:
1621        _setup_stream(player)()
1622
1623        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1624            assert join is True
1625            return start_unix_ms + deferral_ms
1626
1627        player.stream.start = AsyncMock(side_effect=start)
1628
1629    written_chunks: list[bytes] = []
1630
1631    async def capture_write(_player: Any, chunk: bytes) -> None:
1632        written_chunks.append(chunk)
1633
1634    with (
1635        patch.object(session, "_start_client", side_effect=setup_deferred_ack),
1636        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1637        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1638    ):
1639        await session.add_client(player)
1640
1641    commanded_ms = int((now + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000) * 1000) + adjust_ms
1642    assert player.stream.start.await_args.args[0] == commanded_ms
1643    # The ack lands 1.5s later than commanded, so the due position is 9.0s of
1644    # feed instead of 7.5s: the joiner skips 4.0s of the live feed.
1645    assert written_chunks == []
1646    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1647    assert skip_seconds == pytest.approx(4.0, abs=0.01)
1648    # Progress is reported against the sample that lands on the acked instant,
1649    # not the one that would have landed on the commanded instant.
1650    player.stream.rebase_position.assert_called_once_with(9000)
1651
1652
1653@pytest.mark.asyncio
1654async def test_write_chunk_drains_skip_counter_across_chunks() -> None:
1655    """A per-client skip is consumed across chunks, slicing the partial one."""
1656    session = _make_session(0, 0)
1657    player: Any = session.sync_clients[0]
1658    ffmpeg = MagicMock(closed=False)
1659    ffmpeg.write = AsyncMock()
1660    session._player_ffmpeg[player.player_id] = ffmpeg
1661    # skip 1.5s of a 1s-per-chunk feed
1662    session._client_skip_bytes[player.player_id] = PCM_SAMPLE_SIZE * 3 // 2
1663
1664    chunk_one = b"\x01" * PCM_SAMPLE_SIZE
1665    chunk_two = b"\x02" * PCM_SAMPLE_SIZE
1666    chunk_three = b"\x03" * PCM_SAMPLE_SIZE
1667    for chunk in (chunk_one, chunk_two, chunk_three):
1668        await session._write_chunk_to_player(player, chunk)
1669
1670    written = [call_args.args[0] for call_args in ffmpeg.write.await_args_list]
1671    # first chunk fully consumed, second chunk sliced in half, third whole
1672    assert written == [b"\x02" * (PCM_SAMPLE_SIZE // 2), chunk_three]
1673    assert session._client_skip_bytes[player.player_id] == 0
1674
1675
1676def test_effective_start_time_adds_reference_member_shift() -> None:
1677    """The effective anchor adds the first sync client's accumulated shift."""
1678    session = _make_session(100.0, 0)
1679    reference: Any = session.sync_clients[0]
1680    reference.stream.cumulative_shift_seconds = 1.539
1681    assert session.effective_start_time == pytest.approx(101.539)
1682
1683    # the reference transfers to whichever client is first
1684    other = MagicMock()
1685    other.stream.cumulative_shift_seconds = 0.5
1686    session.sync_clients.insert(0, other)
1687    assert session.effective_start_time == pytest.approx(100.5)
1688
1689    # a missing stream falls back to the raw anchor
1690    other.stream = None
1691    assert session.effective_start_time == pytest.approx(100.0)
1692
1693
1694@pytest.mark.asyncio
1695async def test_stop_client_clears_skip_and_shift_state() -> None:
1696    """Tearing a client down drops its skip counter and resets its playout shift."""
1697    session = _make_session(0, 0)
1698    player = _make_late_joiner()
1699    player.stream = _stream_defaults(MagicMock())
1700    player.stream.session = session
1701    player.stream.stop = AsyncMock()
1702    session._client_skip_bytes[player.player_id] = 12_345
1703
1704    await session.stop_client(player)
1705
1706    assert player.player_id not in session._client_skip_bytes
1707    player.stream.reset_reanchor_shift.assert_called_once_with()
1708    player.stream.stop.assert_awaited_once_with(force=True)
1709
1710
1711@pytest.mark.asyncio
1712async def test_replace_clears_skip_and_shift_state() -> None:
1713    """A warm replace re-anchors everyone, so skip counters and shifts reset."""
1714    session = _make_session(0, 0)
1715    player: Any = session.sync_clients[0]
1716    player.config.get_value = MagicMock(return_value=0)
1717    stream = player.stream
1718    stream.running = True
1719    stream.connected = True
1720    stream.flush = AsyncMock(return_value=True)
1721    stream.wait_audio_present = AsyncMock(return_value=True)
1722    session._client_skip_bytes[player.player_id] = 999
1723
1724    with (
1725        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
1726        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
1727        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
1728    ):
1729        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0))
1730
1731    assert session._client_skip_bytes == {}
1732    stream.reset_reanchor_shift.assert_called()
1733
1734
1735@pytest.mark.asyncio
1736async def test_cleanup_after_removal_skips_idle_when_player_has_new_session_stream() -> None:
1737    """Cleanup must not idle a player that was already re-added to another session."""
1738    now = time.time()
1739    session = _make_session(now - 10, 12.5)
1740    player = _make_late_joiner()
1741    other_session = object()
1742    player.set_state_from_stream = MagicMock()
1743    player.stream = _stream_defaults(MagicMock())
1744    player.stream.session = other_session
1745    session.sync_clients.clear()
1746
1747    with (
1748        patch.object(session, "stop_client", new_callable=AsyncMock),
1749        patch.object(session, "stop", new_callable=AsyncMock),
1750    ):
1751        await session._cleanup_after_removal(player)
1752
1753    player.set_state_from_stream.assert_not_called()
1754
1755
1756@pytest.mark.asyncio
1757async def test_late_join_pads_with_silence_when_the_ring_ran_out_under_a_committed_anchor() -> None:
1758    """A due position lost to the ring after the START is covered with silence, not a moved anchor."""
1759    # Freeze time so both the test and the code under test agree on `now`.
1760    now = 1_000_000.0
1761    start_time = now - 100.0
1762    session = _make_session(start_time, 110.0)
1763    session._pcm_total_fed = int(110.0 * PCM_SAMPLE_SIZE)
1764    # An 8s ring against a 10s write-head lead: wide enough when the anchor is
1765    # planned, too narrow by the time the binary owns the instant.
1766    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 8)
1767    session._pcm_buffer_max = PCM_SAMPLE_SIZE * 8
1768    # Freeze the adaptive sizing so this test covers the cap, not the growth.
1769    session._peak_lead_seconds = 1e6
1770    logger = MagicMock()
1771    session.prov.logger = logger
1772    player = _make_late_joiner()
1773
1774    written: list[tuple[str, bytes]] = []
1775
1776    async def capture_write(target: Any, chunk: bytes) -> None:
1777        written.append((target.player_id, chunk))
1778
1779    def setup_with_feed(*_args: Any, **_kwargs: Any) -> None:
1780        _setup_stream(player)()
1781
1782        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1783            assert join is True
1784            # The group keeps being fed while the START ack is outstanding: this
1785            # is what pushes the joiner's due position off the back of the ring.
1786            await session._write_chunk_to_all_players(b"\x02" * int(2.0 * PCM_SAMPLE_SIZE))
1787            return start_unix_ms
1788
1789        player.stream.start = AsyncMock(side_effect=start)
1790
1791    with (
1792        patch.object(session, "_start_client", side_effect=setup_with_feed),
1793        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1794        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1795    ):
1796        await session.add_client(player)
1797
1798    # The anchor is the one that was commanded: an acked instant is never moved.
1799    headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1800    assert _captured_start_at(player) == pytest.approx(now + headroom, abs=0.001)
1801    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1802    # due is 102.5s of feed and the write head reached 112.0s, so 9.5s is owed
1803    # while only 8s survives in the ring: the missing 1.5s opens as silence.
1804    assert len(prime) / PCM_SAMPLE_SIZE == pytest.approx(9.5, abs=0.001)
1805    pad = int(1.5 * PCM_SAMPLE_SIZE)
1806    assert prime[:pad] == bytes(pad), "the missing head must be silence"
1807    assert set(prime[pad:]) == {1, 2}, "the rest must be the buffered feed"
1808    assert pad % session._pcm_frame_size == 0
1809    assert session._client_skip_bytes[player.player_id] == 0
1810    # Position still reports where the GROUP is at that instant, because the
1811    # real content lands exactly where it would have without the shortfall.
1812    player.stream.rebase_position.assert_not_called()
1813    logger.warning.assert_called_once()
1814    assert "silence" in logger.warning.call_args.args[0]
1815
1816
1817@pytest.mark.asyncio
1818async def test_late_join_ring_shortfall_keeps_a_misaligned_ring_head_frame_aligned() -> None:
1819    """A silence pad over a mid-frame ring head still lands the feed on frame boundaries."""
1820    now = 1_000_000.0
1821    session = _make_session(now - 100.0, 110.0)
1822    # Both the write head and the ring head sit mid-frame.
1823    session._pcm_total_fed = int(110.0 * PCM_SAMPLE_SIZE) + 3
1824    session._pcm_buffer = bytearray(b"\x01" * (PCM_SAMPLE_SIZE * 8 + 2))
1825    session._pcm_buffer_max = len(session._pcm_buffer)
1826    session._peak_lead_seconds = 1e6
1827    player = _make_late_joiner()
1828
1829    written: list[tuple[str, bytes]] = []
1830
1831    async def capture_write(target: Any, chunk: bytes) -> None:
1832        written.append((target.player_id, chunk))
1833
1834    def setup_with_feed(*_args: Any, **_kwargs: Any) -> None:
1835        _setup_stream(player)()
1836
1837        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1838            assert join is True
1839            await session._write_chunk_to_all_players(b"\x02" * (int(2.0 * PCM_SAMPLE_SIZE) + 1))
1840            return start_unix_ms
1841
1842        player.stream.start = AsyncMock(side_effect=start)
1843
1844    with (
1845        patch.object(session, "_start_client", side_effect=setup_with_feed),
1846        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1847        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1848    ):
1849        await session.add_client(player)
1850
1851    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1852    frame_size = session._pcm_frame_size
1853    # The prime ends exactly at the write head, so its start - and therefore the
1854    # whole padded prime - has to sit on an absolute frame boundary.
1855    assert (session._pcm_total_fed - len(prime)) % frame_size == 0
1856    pad_len = len(prime) - len(prime.lstrip(b"\x00"))
1857    assert pad_len % frame_size == 0
1858
1859
1860def test_ring_grows_to_the_observed_write_head_lead() -> None:
1861    """The ring tracks the largest lead a session shows, above a floor and under a byte cap."""
1862    now = 1_000_000.0
1863    session = _make_session(now - 100.0, 100.0)
1864    assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1865
1866    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1867        # A 4s lead stays under the floor, which keeps the ring where it was.
1868        session.seconds_streamed = 104.0
1869        session._observe_write_head_lead()
1870        assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1871
1872        # A 15s lead carries the ring past the floor, with the margin on top.
1873        session.seconds_streamed = 115.0
1874        session._observe_write_head_lead()
1875        assert session._peak_lead_seconds == pytest.approx(15.0, abs=0.001)
1876        expected = int((15.0 + AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS) * PCM_SAMPLE_SIZE)
1877        assert session._pcm_buffer_max == expected
1878
1879        # A lead that falls back never shrinks the ring: the history a joiner
1880        # still needs is already in it.
1881        session.seconds_streamed = 108.0
1882        session._observe_write_head_lead()
1883        assert session._pcm_buffer_max == expected
1884
1885        # Growth is bounded in bytes, so a hi-res rate cannot multiply it out.
1886        session.seconds_streamed = 100.0 + 3600.0
1887        session._observe_write_head_lead()
1888        assert session._pcm_buffer_max == AIRPLAY_LATE_JOIN_RING_MAX_BYTES
1889
1890
1891def test_write_head_lead_is_not_measured_before_the_anchor_arrives() -> None:
1892    """Audio fed inside the start lead is not counted as pipeline depth."""
1893    now = 1_000_000.0
1894    # Anchored 2.5s into the future: nothing is audible yet.
1895    session = _make_session(now + 2.5, 8.0)
1896    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1897        session._observe_write_head_lead()
1898    assert session._peak_lead_seconds == 0.0
1899    assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1900
1901    unanchored = _make_session(0.0, 8.0)
1902    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1903        unanchored._observe_write_head_lead()
1904    assert unanchored._peak_lead_seconds == 0.0
1905
1906
1907@pytest.mark.asyncio
1908async def test_late_join_silence_pad_is_bounded_and_reports_the_residual() -> None:
1909    """An implausible ack is padded only up to the ring bound, and says so."""
1910    now = 1_000_000.0
1911    session = _make_session(now - 400.0, 420.0)
1912    session._pcm_total_fed = int(420.0 * PCM_SAMPLE_SIZE)
1913    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 2)
1914    session._pcm_buffer_max = PCM_SAMPLE_SIZE * 2
1915    session._peak_lead_seconds = 1e6
1916    logger = MagicMock()
1917    session.prov.logger = logger
1918    player = _make_late_joiner()
1919
1920    written: list[tuple[str, bytes]] = []
1921
1922    async def capture_write(target: Any, chunk: bytes) -> None:
1923        written.append((target.player_id, chunk))
1924
1925    def setup_with_stale_ack(*_args: Any, **_kwargs: Any) -> None:
1926        _setup_stream(player)()
1927        # The binary reports an instant far behind the commanded one, mapping
1928        # the joiner back near the start of the session. 320s of head is owed.
1929        player.stream.start = AsyncMock(return_value=int((now - 300.0) * 1000))
1930
1931    with (
1932        patch.object(session, "_start_client", side_effect=setup_with_stale_ack),
1933        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1934        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1935    ):
1936        await session.add_client(player)
1937
1938    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1939    # Bounded at one ring of silence plus the ring itself - never the 320s owed.
1940    assert len(prime) / PCM_SAMPLE_SIZE == pytest.approx(4.0, abs=0.01)
1941    assert (session._pcm_total_fed - len(prime)) % session._pcm_frame_size == 0
1942    # The joiner cannot be placed exactly, so the log must not claim sync.
1943    logger.warning.assert_called_once()
1944    tail = logger.warning.call_args.args[-1]
1945    assert "ahead of the group" in tail
1946    assert "in sync" not in tail
1947
1948
1949@pytest.mark.asyncio
1950async def test_start_client_releases_a_foreign_mute_latch() -> None:
1951    """A client joining the session gets its foreign mute latch released on start."""
1952    session = _make_session(start_time=0.0, seconds_streamed=0.0)
1953    player = _make_late_joiner()
1954
1955    with (
1956        patch(
1957            "music_assistant.providers.airplay.stream_session.AirPlayStream",
1958            return_value=MagicMock(connect=AsyncMock()),
1959        ),
1960        patch.object(session, "_start_player_ffmpeg", AsyncMock()),
1961    ):
1962        await session._start_client(player, use_shared_ptp=False)
1963
1964    player.release_foreign_mute_latch.assert_called_once_with()
1965