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 AudioError, 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
928async def test_audio_confirmation_waits_for_the_source_to_feed() -> None:
929    """
930    A binary is only judged silent once it has actually been handed audio.
931
932    A seek can land seconds ahead of what the source has produced, and giving up
933    on the member there would restart the session into the very same wait.
934    """
935    session = _make_session(0, 0)
936    player: Any = session.sync_clients[0]
937    player.stream.wait_audio_present = AsyncMock(return_value=True)
938    feeding = asyncio.Event()
939
940    async def _source() -> None:
941        await feeding.wait()
942
943    session._audio_source_task = asyncio.create_task(_source())
944
945    waiter = asyncio.create_task(session._wait_members_audio_present())
946    for _ in range(5):
947        await asyncio.sleep(0)
948    assert not waiter.done()
949    player.stream.wait_audio_present.assert_not_awaited()
950
951    session._feed_settled.set()
952    await waiter
953
954    player.stream.wait_audio_present.assert_awaited_once()
955    feeding.set()
956    await session._audio_source_task
957
958
959@pytest.mark.asyncio
960async def test_a_source_that_dies_settles_the_feed_question() -> None:
961    """A source that fails never leaves a start waiting for audio it will not get."""
962    session = _make_session(0, 0)
963    player: Any = session.sync_clients[0]
964    player.stream.write_audio_eof = AsyncMock()
965    session.media = MagicMock(source_id=None, queue_session_id=None)
966    no_chunks: list[bytes] = []
967
968    async def failing_source() -> AsyncGenerator[bytes]:
969        for chunk in no_chunks:
970            yield chunk
971        raise AudioError("source died")
972
973    await session._audio_streamer(failing_source())
974
975    assert session._feed_settled.is_set()
976
977
978@pytest.mark.asyncio
979async def test_warm_replace_rearms_the_feed_question() -> None:
980    """The new source answers for its own feed; what the old one delivered does not count."""
981    session = _make_session(0, 0)
982    player: Any = session.sync_clients[0]
983    player.config.get_value = MagicMock(return_value=0)
984    player.stream.flush = AsyncMock(return_value=True)
985    session._feed_settled.set()
986
987    with (
988        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
989        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
990        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
991    ):
992        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0))
993
994    assert not session._feed_settled.is_set()
995
996
997@pytest.mark.asyncio
998async def test_audio_confirmation_is_released_by_a_source_that_never_feeds() -> None:
999    """A source that ends without handing anything over never holds up the start."""
1000    session = _make_session(0, 0)
1001    player: Any = session.sync_clients[0]
1002    player.display_name = "Kantoor"
1003    player.stream.wait_audio_present = AsyncMock(return_value=False)
1004    session._audio_source_task = asyncio.create_task(asyncio.sleep(0))
1005    await session._audio_source_task
1006
1007    with pytest.raises(PlayerCommandFailed, match="audio feed was not confirmed"):
1008        await session._wait_members_audio_present()
1009
1010
1011@pytest.mark.asyncio
1012@pytest.mark.parametrize(
1013    ("transitioning", "live_session_id", "ends_stream"),
1014    [
1015        # a seek/next: the queue rotates its session while it loads the new stream
1016        (True, "session-2", False),
1017        # the queue played out, or the flow broke off for the queue to restart once
1018        # the player reports idle - which only the audio EOF can bring about
1019        (False, "session-1", True),
1020        # not this session's transition (the queue is starting the one we play)
1021        (True, "session-1", True),
1022        # a transition that already finished cannot be waiting on this session
1023        (False, "session-2", True),
1024    ],
1025)
1026async def test_source_end_only_keeps_stdin_open_for_a_pending_replacement(
1027    transitioning: bool, live_session_id: str, ends_stream: bool
1028) -> None:
1029    """
1030    The binary's stdin is closed unless the queue is mid-handover to a new stream.
1031
1032    Closing it ends the stream for good - it cannot be reopened - so a seek would
1033    be left with a cold restart as its only option. Everywhere else the EOF is
1034    what makes the binary end and the player report idle.
1035    """
1036    session = _make_session(0, 0)
1037    player: Any = session.sync_clients[0]
1038    player.stream.write_audio_eof = AsyncMock()
1039    session.media = MagicMock(source_id="queue-1", queue_session_id="session-1")
1040    queues: Any = session.mass.player_queues
1041    queues.queue_data_or_none = MagicMock(
1042        return_value=MagicMock(session_id=live_session_id, transitioning=transitioning)
1043    )
1044    ffmpeg = MagicMock(closed=False)
1045    ffmpeg.write_eof = AsyncMock()
1046    ffmpeg.wait_with_timeout = AsyncMock()
1047    ffmpeg.kill = AsyncMock()
1048    session._player_ffmpeg[player.player_id] = ffmpeg
1049
1050    no_chunks: list[bytes] = []
1051
1052    async def exhausted_source() -> AsyncGenerator[bytes]:
1053        for chunk in no_chunks:
1054            yield chunk
1055
1056    await session._audio_streamer(exhausted_source())
1057
1058    assert player.player_id not in session._player_ffmpeg
1059    assert player.stream.write_audio_eof.await_count == (1 if ends_stream else 0)
1060    if ends_stream:
1061        # the audio ffmpeg still holds is handed over before the binary is told
1062        ffmpeg.write_eof.assert_awaited_once()
1063        ffmpeg.wait_with_timeout.assert_awaited_once()
1064        ffmpeg.kill.assert_not_awaited()
1065    else:
1066        # the replacement flushes that audio away, and nothing may reach the
1067        # binary between the old ffmpeg dying and that flush
1068        ffmpeg.kill.assert_awaited_once()
1069        ffmpeg.write_eof.assert_not_awaited()
1070
1071
1072@pytest.mark.asyncio
1073@pytest.mark.parametrize("stream_state", ["missing", "stopped", "disconnected", "audio_ended"])
1074async def test_standby_requires_every_member_running_and_connected(stream_state: str) -> None:
1075    """Standby is unavailable when any member lacks a reusable connected session."""
1076    session = _make_session(0, 0)
1077    ready_player = MagicMock(player_id="ready", protocol=StreamingProtocol.AIRPLAY2)
1078    ready_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1079    ready_player.stream.send_cli_command = AsyncMock()
1080    unavailable_player = MagicMock(player_id="unavailable", protocol=StreamingProtocol.RAOP)
1081    unavailable_player.stream = None
1082    if stream_state != "missing":
1083        unavailable_player.stream = MagicMock(
1084            running=stream_state != "stopped",
1085            # a stream that was sent its audio EOF is still running, but its
1086            # stdin is closed for good so it can never be refilled
1087            accepts_audio=stream_state not in ("stopped", "audio_ended"),
1088            connected=stream_state != "disconnected",
1089        )
1090        unavailable_player.stream.send_cli_command = AsyncMock()
1091    players: Any = [ready_player, unavailable_player]
1092    session.sync_clients = players
1093
1094    assert await session.standby() is False
1095    assert session.can_replace(players, session.pcm_format) is False
1096    ready_player.stream.send_cli_command.assert_not_awaited()
1097    ready_player.set_state_from_stream.assert_not_called()
1098    if unavailable_player.stream:
1099        unavailable_player.stream.send_cli_command.assert_not_awaited()
1100        unavailable_player.set_state_from_stream.assert_not_called()
1101
1102
1103@pytest.mark.asyncio
1104async def test_standby_returns_false_when_command_is_not_delivered() -> None:
1105    """Standby fails without changing state for a member that misses the command."""
1106    session = _make_session(0, 0)
1107    logger = MagicMock()
1108    session.prov.logger = logger
1109    delivered_player = MagicMock(player_id="delivered", protocol=StreamingProtocol.AIRPLAY2)
1110    delivered_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1111    delivered_player.stream.send_cli_command = AsyncMock(return_value=True)
1112    dropped_player = MagicMock(player_id="dropped", protocol=StreamingProtocol.RAOP)
1113    dropped_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1114    dropped_player.stream.send_cli_command = AsyncMock(return_value=False)
1115    pending_player = MagicMock(player_id="pending", protocol=StreamingProtocol.AIRPLAY2)
1116    pending_player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1117    pending_player.stream.send_cli_command = AsyncMock(return_value=True)
1118    session.sync_clients = [delivered_player, dropped_player, pending_player]
1119
1120    assert await session.standby() is False
1121    delivered_player.stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
1122    delivered_player.set_state_from_stream.assert_called_once_with(
1123        state=PlaybackState.PAUSED, stream=delivered_player.stream
1124    )
1125    dropped_player.stream.send_cli_command.assert_awaited_once_with("ACTION=STANDBY")
1126    dropped_player.set_state_from_stream.assert_not_called()
1127    pending_player.stream.send_cli_command.assert_not_awaited()
1128    pending_player.set_state_from_stream.assert_not_called()
1129    logger.warning.assert_called_once()
1130
1131
1132@pytest.mark.asyncio
1133async def test_standby_returns_false_when_command_raises() -> None:
1134    """Standby fails without changing state when command delivery raises."""
1135    session = _make_session(0, 0)
1136    logger = MagicMock()
1137    session.prov.logger = logger
1138    player = MagicMock(player_id="failed", protocol=StreamingProtocol.AIRPLAY2)
1139    player.stream = _stream_defaults(MagicMock(running=True, connected=True))
1140    player.stream.send_cli_command = AsyncMock(side_effect=OSError("command pipe failed"))
1141    session.sync_clients = [player]
1142
1143    assert await session.standby() is False
1144    player.set_state_from_stream.assert_not_called()
1145    logger.warning.assert_called_once()
1146
1147
1148@pytest.mark.asyncio
1149async def test_late_join_empty_buffer() -> None:
1150    """Test that with an empty buffer, start_at = start_time + seconds_streamed."""
1151    now = time.time()
1152    start_time = now - 8
1153    seconds_streamed = 12.5
1154    session = _make_session(start_time, seconds_streamed)
1155    player = _make_late_joiner()
1156
1157    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1158        mock_start.side_effect = _setup_stream(player)
1159        await session.add_client(player)
1160
1161    assert mock_start.called, "_start_client was never called"
1162    expected = start_time + seconds_streamed
1163    assert abs(_captured_start_at(player) - expected) < 0.1
1164
1165
1166@pytest.mark.asyncio
1167async def test_late_join_caps_prime_at_whole_ring_when_position_predates_it() -> None:
1168    """A due position older than the ring caps the prime at the whole buffer."""
1169    now = 1_000_000.0
1170    # Synthetic anchor in the future forces the due position behind the oldest
1171    # buffered sample: only the whole ring can prime and the anchor is pulled
1172    # to the ring's first sample so content and anchor stay exactly aligned.
1173    seconds_streamed = 12.5
1174    buffer_seconds = 3
1175    start_time = now + 5.0
1176    session = _make_session(start_time, seconds_streamed)
1177    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * buffer_seconds)
1178    player = _make_late_joiner()
1179
1180    written_chunks: list[bytes] = []
1181
1182    async def capture_write(_player: Any, chunk: bytes) -> None:
1183        written_chunks.append(chunk)
1184
1185    with (
1186        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1187        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1188        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1189    ):
1190        mock_start.side_effect = _setup_stream(player)
1191        await session.add_client(player)
1192
1193    assert mock_start.called, "_start_client was never called"
1194    # the whole ring is primed, nothing is skipped, and the anchor maps to the
1195    # ring's first sample: start_at = start_time + (seconds_streamed - buffer)
1196    assert session._client_skip_bytes[player.player_id] == 0
1197    assert written_chunks, "expected the whole ring to be primed"
1198    assert len(written_chunks[0]) / PCM_SAMPLE_SIZE == pytest.approx(buffer_seconds, abs=0.01)
1199    expected = start_time + (seconds_streamed - buffer_seconds)
1200    assert _captured_start_at(player) == pytest.approx(expected, abs=0.02)
1201
1202
1203@pytest.mark.asyncio
1204async def test_late_join_adds_to_sync_clients() -> None:
1205    """Test that the late joiner is added to sync_clients."""
1206    now = time.time()
1207    session = _make_session(now - 10, 12.5)
1208    player = _make_late_joiner()
1209
1210    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1211        mock_start.side_effect = _setup_stream(player)
1212        await session.add_client(player)
1213
1214    assert player in session.sync_clients
1215
1216
1217@pytest.mark.asyncio
1218async def test_late_join_start_failure_stops_client() -> None:
1219    """A late joiner whose START fails is torn down before joining the session."""
1220    session = _make_session(time.time() - 10, 12.5)
1221    player = _make_late_joiner()
1222
1223    def setup_failing_start(*_args: Any, **_kwargs: Any) -> None:
1224        _setup_stream(player)()
1225        player.stream.start = AsyncMock(side_effect=OSError("start failed"))
1226
1227    with (
1228        patch.object(session, "_start_client", side_effect=setup_failing_start),
1229        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1230    ):
1231        await session.add_client(player)
1232
1233    player.stream.start.assert_awaited_once()
1234    # START precedes the prime feed and the sync_clients append, so a failed
1235    # joiner is stopped without ever having joined the session.
1236    assert player not in session.sync_clients
1237    stop_client.assert_awaited_once_with(player, reason="late joiner start/prime failed")
1238
1239
1240@pytest.mark.asyncio
1241async def test_late_join_unacknowledged_start_stops_client() -> None:
1242    """A joiner whose START is never acked is torn down, never mapped onto that instant."""
1243    session = _make_session(time.time() - 10, 12.5)
1244    player = _make_late_joiner()
1245
1246    def setup_unacknowledged_start(*_args: Any, **_kwargs: Any) -> None:
1247        _setup_stream(player)()
1248        player.stream.start = AsyncMock(
1249            side_effect=PlayerCommandFailed(
1250                "AirPlay player Player B did not acknowledge its start within 5.0s"
1251            )
1252        )
1253
1254    with (
1255        patch.object(session, "_start_client", side_effect=setup_unacknowledged_start),
1256        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1257    ):
1258        await session.add_client(player)
1259
1260    assert player not in session.sync_clients
1261    assert player.player_id not in session._client_skip_bytes
1262    player.stream.rebase_position.assert_not_called()
1263    stop_client.assert_awaited_once_with(player, reason="late joiner start/prime failed")
1264
1265
1266@pytest.mark.asyncio
1267async def test_late_join_refuses_a_parked_session() -> None:
1268    """A parked (standby) session has no live timeline, so it cannot absorb a joiner."""
1269    session = _make_session(time.time() - 10, 12.5)
1270    reference: Any = session.sync_clients[0]
1271    reference.stream.send_cli_command = AsyncMock(return_value=True)
1272    reference.set_state_from_stream = MagicMock(
1273        side_effect=lambda **kwargs: setattr(reference, "playback_state", kwargs["state"])
1274    )
1275    assert await session.standby()
1276    assert reference.playback_state == PlaybackState.PAUSED
1277    player = _make_late_joiner()
1278
1279    with (
1280        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1281        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1282    ):
1283        await session.add_client(player)
1284
1285    # The parked session zeroed seconds_streamed while start_time stayed put, so
1286    # anchoring here maps the joiner onto a timeline nothing is playing.
1287    mock_start.assert_not_called()
1288    stop_client.assert_not_awaited()
1289    assert player not in session.sync_clients
1290
1291
1292@pytest.mark.asyncio
1293async def test_late_join_no_running_session() -> None:
1294    """Test that add_client is a no-op when no session is running."""
1295    now = time.time()
1296    session = _make_session(now - 10, 12.5)
1297    # Make the leader's stream not running
1298    leader = session.sync_clients[0]
1299    leader.stream = _stream_defaults(MagicMock())
1300    leader.stream.running = False
1301    player = _make_late_joiner()
1302
1303    with patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start:
1304        await session.add_client(player)
1305        mock_start.assert_not_called()
1306        assert player not in session.sync_clients
1307
1308
1309@pytest.mark.asyncio
1310async def test_late_join_primes_from_ring_tail_at_headroom() -> None:
1311    """A due position inside the ring primes from the tail and anchors at now + headroom."""
1312    # Freeze time so both the test and the code under test agree on `now`.
1313    now = 1_000_000.0
1314    start_time = now - 0.5
1315    seconds_streamed = 5.0
1316    session = _make_session(start_time, seconds_streamed)
1317    # Fill ring buffer with 5 seconds of non-silent PCM.
1318    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
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    # start_at is now + min_headroom (the late-join floor); fed_pos_due = 2.0s
1335    assert mock_start.called, "_start_client was never called"
1336    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1337    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01), (
1338        f"start_at should be at now + min_headroom, "
1339        f"got offset {_captured_start_at(player) - now:.4f}s"
1340    )
1341
1342    # The last 3.0s of the ring is primed (positions 2.0s..5.0s), nothing skipped.
1343    assert session._client_skip_bytes[player.player_id] == 0
1344    assert written_chunks, "No data was written to the player"
1345    remaining_seconds = len(written_chunks[0]) / PCM_SAMPLE_SIZE
1346    expected_primed = seconds_streamed - (expected_headroom + (now - start_time))
1347    assert remaining_seconds == pytest.approx(expected_primed, abs=0.01), (
1348        f"expected {expected_primed:.2f}s primed, got {remaining_seconds:.4f}s"
1349    )
1350
1351
1352@pytest.mark.asyncio
1353async def test_late_join_skips_live_feed_when_anchor_ahead_of_write_head() -> None:
1354    """When the due position is ahead of the write head, skip that many live bytes."""
1355    # Freeze time so both the test and the code under test agree on `now`.
1356    now = 1_000_000.0
1357    # Diagnosed clamp case: now - start_time = 8.84s, seconds_streamed = 10.0s,
1358    # min_headroom = 2.5s (the late-join floor, no readiness projection) and no
1359    # group shift, so the anchor is due at 11.34s of feed, past the 10.0s write
1360    # head.
1361    start_time = now - 8.84
1362    seconds_streamed = 10.0
1363    session = _make_session(start_time, seconds_streamed)
1364    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 10)
1365    player = _make_late_joiner()
1366
1367    written_chunks: list[bytes] = []
1368
1369    async def capture_write(_player: Any, chunk: bytes) -> None:
1370        written_chunks.append(chunk)
1371
1372    with (
1373        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1374        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1375        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1376    ):
1377        mock_start.side_effect = _setup_stream(player)
1378        await session.add_client(player)
1379
1380    # anchor is now + min_headroom and nothing is primed (position past the head)
1381    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1382    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1383    assert written_chunks == []
1384    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1385    assert skip_seconds == pytest.approx(1.34, abs=0.01)
1386
1387
1388@pytest.mark.asyncio
1389async def test_late_join_primes_from_ring_under_group_shift() -> None:
1390    """A reference member that re-anchored later pulls the due position back into the ring."""
1391    # Freeze time so both the test and the code under test agree on `now`.
1392    now = 1_000_000.0
1393    # Same base as the clamp case, but the reference member accumulated a
1394    # +3.039s starvation shift (134020 frames @44100) so the group's effective
1395    # anchor is later: fed_pos_due = 8.30s, back inside the ring. The joiner is
1396    # primed with ~1.7s from the ring tail and skips nothing.
1397    start_time = now - 8.84
1398    seconds_streamed = 10.0
1399    session = _make_session(start_time, seconds_streamed)
1400    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 10)
1401    reference: Any = session.sync_clients[0]
1402    reference.stream.cumulative_shift_seconds = 134020 / 44100
1403    player = _make_late_joiner()
1404
1405    written_chunks: list[bytes] = []
1406
1407    async def capture_write(_player: Any, chunk: bytes) -> None:
1408        written_chunks.append(chunk)
1409
1410    with (
1411        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1412        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1413        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1414    ):
1415        mock_start.side_effect = _setup_stream(player)
1416        await session.add_client(player)
1417
1418    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1419    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1420    assert session._client_skip_bytes[player.player_id] == 0
1421    assert written_chunks, "expected a prime write from the ring tail"
1422    primed_seconds = len(written_chunks[0]) / PCM_SAMPLE_SIZE
1423    assert primed_seconds == pytest.approx(1.699, abs=0.01)
1424
1425
1426@pytest.mark.asyncio
1427async def test_late_join_anchors_on_the_reported_clock_readiness(
1428    caplog: pytest.LogCaptureFixture,
1429) -> None:
1430    """A projected readiness instant anchors the join, just past the receiver's clock."""
1431    # Freeze time so both the test and the code under test agree on `now`.
1432    now = 1_000_000.0
1433    session = _make_session(now - 5.0, 5.0)
1434    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1435    session.prov.logger = logging.getLogger("test.airplay.session")
1436    player = _make_late_joiner()
1437    # A cold receiver: its clock is projected usable 3.0s out, well past the floor.
1438    ready_at_unix_ms = int((now + 3.0) * 1000)
1439
1440    def setup_with_projection(*_args: Any, **_kwargs: Any) -> None:
1441        _setup_stream(player)()
1442        player.stream.wait_clock_ready = AsyncMock(
1443            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
1444        )
1445
1446    with (
1447        caplog.at_level(logging.DEBUG),
1448        patch.object(session, "_start_client", side_effect=setup_with_projection),
1449        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1450        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1451    ):
1452        await session.add_client(player)
1453
1454    expected_lead = AIRPLAY_CLOCK_READY_LEAD_MS / 1000
1455    assert _captured_start_at(player) - now == pytest.approx(3.0 + expected_lead, abs=0.01)
1456    player.stream.wait_clock_ready.assert_awaited_once_with(
1457        timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
1458    )
1459    assert "receiver clock usable in 3.00s; anchoring no earlier than that" in caplog.text
1460
1461
1462@pytest.mark.asyncio
1463async def test_late_join_floor_wins_over_a_clock_that_is_already_ready(
1464    caplog: pytest.LogCaptureFixture,
1465) -> None:
1466    """A receiver whose clock is already locked still gets the join floor as its anchor."""
1467    # Freeze time so both the test and the code under test agree on `now`.
1468    now = 1_000_000.0
1469    session = _make_session(now - 5.0, 5.0)
1470    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1471    session.prov.logger = logging.getLogger("test.airplay.session")
1472    player = _make_late_joiner()
1473    # A warm receiver reports a readiness instant that has already passed.
1474    ready_at_unix_ms = int((now - 1.0) * 1000)
1475
1476    def setup_with_projection(*_args: Any, **_kwargs: Any) -> None:
1477        _setup_stream(player)()
1478        player.stream.wait_clock_ready = AsyncMock(
1479            return_value=(ClockReadiness.PROJECTED, ready_at_unix_ms)
1480        )
1481
1482    with (
1483        caplog.at_level(logging.DEBUG),
1484        patch.object(session, "_start_client", side_effect=setup_with_projection),
1485        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1486        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1487    ):
1488        await session.add_client(player)
1489
1490    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1491    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1492    assert "receiver clock became usable 1.00s ago; anchoring on the join floor" in caplog.text
1493
1494
1495@pytest.mark.asyncio
1496async def test_late_join_falls_back_to_the_floor_without_a_clock_projection() -> None:
1497    """No projection (NTP timing or a silent receiver) anchors on the floor."""
1498    # Freeze time so both the test and the code under test agree on `now`.
1499    now = 1_000_000.0
1500    session = _make_session(now - 5.0, 5.0)
1501    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1502    player = _make_late_joiner()
1503
1504    with (
1505        patch.object(session, "_start_client", new_callable=AsyncMock) as mock_start,
1506        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1507        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1508    ):
1509        mock_start.side_effect = _setup_stream(player)
1510        await session.add_client(player)
1511
1512    # every fallback shape surfaces as "no projection" to the session
1513    player.stream.wait_clock_ready.assert_awaited_once_with(
1514        timeout=AIRPLAY_CLOCK_READY_TIMEOUT_MS / 1000
1515    )
1516    expected_headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1517    assert _captured_start_at(player) - now == pytest.approx(expected_headroom, abs=0.01)
1518    assert player in session.sync_clients
1519
1520
1521@pytest.mark.asyncio
1522@pytest.mark.parametrize(
1523    "readiness",
1524    [ClockReadiness.UNREPORTED, ClockReadiness.NOT_APPLICABLE],
1525    ids=["unreported", "ntp"],
1526)
1527async def test_late_join_without_a_projection_still_joins(readiness: ClockReadiness) -> None:
1528    """A device with no clock to wait for is a fallback, not a reason to refuse it."""
1529    now = 1_000_000.0
1530    session = _make_session(now - 5.0, 5.0)
1531    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1532    player = _make_late_joiner()
1533
1534    def setup_without_projection(*_args: Any, **_kwargs: Any) -> None:
1535        _setup_stream(player)()
1536        player.stream.wait_clock_ready = AsyncMock(return_value=(readiness, 0))
1537
1538    with (
1539        patch.object(session, "_start_client", side_effect=setup_without_projection),
1540        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1541        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1542    ):
1543        await session.add_client(player)
1544
1545    assert player in session.sync_clients
1546
1547
1548@pytest.mark.asyncio
1549async def test_late_joiner_with_a_stalled_clock_is_not_added() -> None:
1550    """A receiver that never answered our clock renders silence, so keep it out of the group."""
1551    now = 1_000_000.0
1552    session = _make_session(now - 5.0, 5.0)
1553    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1554    player = _make_late_joiner()
1555
1556    def setup_stalled(*_args: Any, **_kwargs: Any) -> None:
1557        _setup_stream(player)()
1558        player.stream.wait_clock_ready = AsyncMock(return_value=(ClockReadiness.STALLED, 0))
1559
1560    with (
1561        patch.object(session, "_start_client", side_effect=setup_stalled),
1562        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1563        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1564        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1565    ):
1566        await session.add_client(player)
1567
1568    player.stream.start.assert_not_awaited()
1569    assert player not in session.sync_clients
1570    stop_client.assert_awaited_once_with(player, reason="receiver clock stalled")
1571
1572
1573@pytest.mark.asyncio
1574async def test_late_join_feed_keeps_flowing_while_waiting_for_clock_readiness() -> None:
1575    """The group keeps being fed while a joiner's receiver clock projection is pending."""
1576    session = _make_session(time.time() - 5, 5.0)
1577    player = _make_late_joiner()
1578    readiness_pending = asyncio.Event()
1579    readiness_released = asyncio.Event()
1580
1581    def setup_pending_projection(*_args: Any, **_kwargs: Any) -> None:
1582        _setup_stream(player)()
1583
1584        async def wait_clock_ready(*_args: Any, **_kwargs: Any) -> tuple[ClockReadiness, int]:
1585            readiness_pending.set()
1586            await readiness_released.wait()
1587            return (ClockReadiness.UNREPORTED, 0)
1588
1589        player.stream.wait_clock_ready = AsyncMock(side_effect=wait_clock_ready)
1590
1591    with (
1592        patch.object(session, "_start_client", side_effect=setup_pending_projection),
1593        patch.object(session, "_write_chunk_to_player", new_callable=AsyncMock),
1594    ):
1595        join = asyncio.create_task(session.add_client(player))
1596        await asyncio.wait_for(readiness_pending.wait(), timeout=5)
1597        assert await asyncio.wait_for(
1598            session._write_chunk_to_all_players(b"\x02" * PCM_SAMPLE_SIZE), timeout=5
1599        )
1600        readiness_released.set()
1601        await asyncio.wait_for(join, timeout=5)
1602
1603    assert session.seconds_streamed == pytest.approx(6.0)
1604    assert player in session.sync_clients
1605
1606
1607@pytest.mark.asyncio
1608async def test_late_join_feed_keeps_flowing_while_start_ack_is_outstanding() -> None:
1609    """The group keeps being fed while a join's START ack is outstanding."""
1610    # Freeze time so both the test and the code under test agree on `now`.
1611    now = 1_000_000.0
1612    start_time = now - 5.0
1613    session = _make_session(start_time, 5.0)
1614    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1615    player = _make_late_joiner()
1616    ack_outstanding = asyncio.Event()
1617    ack_released = asyncio.Event()
1618
1619    def setup_deferred_ack(*_args: Any, **_kwargs: Any) -> None:
1620        _setup_stream(player)()
1621
1622        async def start(start_unix_ms: int, *_args: Any, **_kwargs: Any) -> int:
1623            # the binary holds its ack until the receiver clock is verified
1624            ack_outstanding.set()
1625            await ack_released.wait()
1626            return start_unix_ms
1627
1628        player.stream.start = AsyncMock(side_effect=start)
1629
1630    writes: list[tuple[str, int]] = []
1631
1632    async def capture_write(target: Any, chunk: bytes) -> None:
1633        writes.append((target.player_id, len(chunk)))
1634
1635    with (
1636        patch.object(session, "_start_client", side_effect=setup_deferred_ack),
1637        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1638        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1639    ):
1640        join = asyncio.create_task(session.add_client(player))
1641        await asyncio.wait_for(ack_outstanding.wait(), timeout=5)
1642        assert await asyncio.wait_for(
1643            session._write_chunk_to_all_players(b"\x02" * PCM_SAMPLE_SIZE), timeout=5
1644        )
1645        ack_released.set()
1646        await asyncio.wait_for(join, timeout=5)
1647
1648    # that second of feed reached the leader and moved the write head to 6.0s
1649    assert ("leader", PCM_SAMPLE_SIZE) in writes
1650    assert session.seconds_streamed == pytest.approx(6.0)
1651    # the anchor (now + the 2.5s floor) is due at 7.5s of feed, so the joiner
1652    # skips only the 1.5s still to come, not the 2.5s due at the commanded
1653    # mapping: the content is mapped against the head the feed actually reached
1654    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1655    assert skip_seconds == pytest.approx(1.5, abs=0.01)
1656    assert player in session.sync_clients
1657
1658
1659@pytest.mark.asyncio
1660async def test_late_join_cancelled_while_ack_outstanding_stops_the_client() -> None:
1661    """A join cancelled while its START ack is outstanding never half-joins the session."""
1662    session = _make_session(time.time() - 5, 5.0)
1663    player = _make_late_joiner()
1664    ack_outstanding = asyncio.Event()
1665
1666    def setup_pending_ack(*_args: Any, **_kwargs: Any) -> None:
1667        _setup_stream(player)()
1668
1669        async def start(*_args: Any, **_kwargs: Any) -> None:
1670            ack_outstanding.set()
1671            await asyncio.Event().wait()
1672
1673        player.stream.start = AsyncMock(side_effect=start)
1674
1675    with (
1676        patch.object(session, "_start_client", side_effect=setup_pending_ack),
1677        patch.object(session, "stop_client", new_callable=AsyncMock) as stop_client,
1678    ):
1679        join = asyncio.create_task(session.add_client(player))
1680        await asyncio.wait_for(ack_outstanding.wait(), timeout=5)
1681        join.cancel()
1682        with pytest.raises(asyncio.CancelledError):
1683            await join
1684
1685    assert player not in session.sync_clients
1686    stop_client.assert_awaited_once_with(player, reason="late joiner start cancelled")
1687
1688
1689@pytest.mark.asyncio
1690async def test_late_join_maps_content_from_the_acked_instant() -> None:
1691    """A binary that acks later than commanded gets its content mapped to the acked instant."""
1692    # Freeze time so both the test and the code under test agree on `now`.
1693    now = 1_000_000.0
1694    start_time = now - 5.0
1695    session = _make_session(start_time, 5.0)
1696    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 5)
1697    player = _make_late_joiner()
1698    # A sync_adjust rides on top of the commanded instant, so it has to be taken
1699    # back out of the ack before the content is mapped onto it.
1700    adjust_ms = 200
1701    player.config.get_value = MagicMock(return_value=adjust_ms)
1702    deferral_ms = 1500
1703
1704    def setup_deferred_ack(*_args: Any, **_kwargs: Any) -> None:
1705        _setup_stream(player)()
1706
1707        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1708            assert join is True
1709            return start_unix_ms + deferral_ms
1710
1711        player.stream.start = AsyncMock(side_effect=start)
1712
1713    written_chunks: list[bytes] = []
1714
1715    async def capture_write(_player: Any, chunk: bytes) -> None:
1716        written_chunks.append(chunk)
1717
1718    with (
1719        patch.object(session, "_start_client", side_effect=setup_deferred_ack),
1720        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1721        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1722    ):
1723        await session.add_client(player)
1724
1725    commanded_ms = int((now + AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000) * 1000) + adjust_ms
1726    assert player.stream.start.await_args.args[0] == commanded_ms
1727    # The ack lands 1.5s later than commanded, so the due position is 9.0s of
1728    # feed instead of 7.5s: the joiner skips 4.0s of the live feed.
1729    assert written_chunks == []
1730    skip_seconds = session._client_skip_bytes[player.player_id] / PCM_SAMPLE_SIZE
1731    assert skip_seconds == pytest.approx(4.0, abs=0.01)
1732    # Progress is reported against the sample that lands on the acked instant,
1733    # not the one that would have landed on the commanded instant.
1734    player.stream.rebase_position.assert_called_once_with(9000)
1735
1736
1737@pytest.mark.asyncio
1738async def test_write_chunk_drains_skip_counter_across_chunks() -> None:
1739    """A per-client skip is consumed across chunks, slicing the partial one."""
1740    session = _make_session(0, 0)
1741    player: Any = session.sync_clients[0]
1742    ffmpeg = MagicMock(closed=False)
1743    ffmpeg.write = AsyncMock()
1744    session._player_ffmpeg[player.player_id] = ffmpeg
1745    # skip 1.5s of a 1s-per-chunk feed
1746    session._client_skip_bytes[player.player_id] = PCM_SAMPLE_SIZE * 3 // 2
1747
1748    chunk_one = b"\x01" * PCM_SAMPLE_SIZE
1749    chunk_two = b"\x02" * PCM_SAMPLE_SIZE
1750    chunk_three = b"\x03" * PCM_SAMPLE_SIZE
1751    for chunk in (chunk_one, chunk_two, chunk_three):
1752        await session._write_chunk_to_player(player, chunk)
1753
1754    written = [call_args.args[0] for call_args in ffmpeg.write.await_args_list]
1755    # first chunk fully consumed, second chunk sliced in half, third whole
1756    assert written == [b"\x02" * (PCM_SAMPLE_SIZE // 2), chunk_three]
1757    assert session._client_skip_bytes[player.player_id] == 0
1758
1759
1760def test_effective_start_time_adds_reference_member_shift() -> None:
1761    """The effective anchor adds the first sync client's accumulated shift."""
1762    session = _make_session(100.0, 0)
1763    reference: Any = session.sync_clients[0]
1764    reference.stream.cumulative_shift_seconds = 1.539
1765    assert session.effective_start_time == pytest.approx(101.539)
1766
1767    # the reference transfers to whichever client is first
1768    other = MagicMock()
1769    other.stream.cumulative_shift_seconds = 0.5
1770    session.sync_clients.insert(0, other)
1771    assert session.effective_start_time == pytest.approx(100.5)
1772
1773    # a missing stream falls back to the raw anchor
1774    other.stream = None
1775    assert session.effective_start_time == pytest.approx(100.0)
1776
1777
1778@pytest.mark.asyncio
1779async def test_stop_client_clears_skip_and_shift_state() -> None:
1780    """Tearing a client down drops its skip counter and resets its playout shift."""
1781    session = _make_session(0, 0)
1782    player = _make_late_joiner()
1783    player.stream = _stream_defaults(MagicMock())
1784    player.stream.session = session
1785    player.stream.stop = AsyncMock()
1786    session._client_skip_bytes[player.player_id] = 12_345
1787
1788    await session.stop_client(player)
1789
1790    assert player.player_id not in session._client_skip_bytes
1791    player.stream.reset_reanchor_shift.assert_called_once_with()
1792    player.stream.stop.assert_awaited_once_with(force=True)
1793
1794
1795@pytest.mark.asyncio
1796async def test_replace_clears_skip_and_shift_state() -> None:
1797    """A warm replace re-anchors everyone, so skip counters and shifts reset."""
1798    session = _make_session(0, 0)
1799    player: Any = session.sync_clients[0]
1800    player.config.get_value = MagicMock(return_value=0)
1801    stream = player.stream
1802    stream.running = True
1803    stream.connected = True
1804    stream.flush = AsyncMock(return_value=True)
1805    stream.wait_audio_present = AsyncMock(return_value=True)
1806    session._client_skip_bytes[player.player_id] = 999
1807
1808    with (
1809        patch.object(session, "_start_player_ffmpeg", new_callable=AsyncMock),
1810        patch.object(session, "_audio_streamer", new_callable=AsyncMock),
1811        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=100.0),
1812    ):
1813        assert await session.replace(MagicMock(), MagicMock(elapsed_time=0))
1814
1815    assert session._client_skip_bytes == {}
1816    stream.reset_reanchor_shift.assert_called()
1817
1818
1819@pytest.mark.asyncio
1820async def test_cleanup_after_removal_skips_idle_when_player_has_new_session_stream() -> None:
1821    """Cleanup must not idle a player that was already re-added to another session."""
1822    now = time.time()
1823    session = _make_session(now - 10, 12.5)
1824    player = _make_late_joiner()
1825    other_session = object()
1826    player.set_state_from_stream = MagicMock()
1827    player.stream = _stream_defaults(MagicMock())
1828    player.stream.session = other_session
1829    session.sync_clients.clear()
1830
1831    with (
1832        patch.object(session, "stop_client", new_callable=AsyncMock),
1833        patch.object(session, "stop", new_callable=AsyncMock),
1834    ):
1835        await session._cleanup_after_removal(player)
1836
1837    player.set_state_from_stream.assert_not_called()
1838
1839
1840@pytest.mark.asyncio
1841async def test_late_join_pads_with_silence_when_the_ring_ran_out_under_a_committed_anchor() -> None:
1842    """A due position lost to the ring after the START is covered with silence, not a moved anchor."""
1843    # Freeze time so both the test and the code under test agree on `now`.
1844    now = 1_000_000.0
1845    start_time = now - 100.0
1846    session = _make_session(start_time, 110.0)
1847    session._pcm_total_fed = int(110.0 * PCM_SAMPLE_SIZE)
1848    # An 8s ring against a 10s write-head lead: wide enough when the anchor is
1849    # planned, too narrow by the time the binary owns the instant.
1850    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 8)
1851    session._pcm_buffer_max = PCM_SAMPLE_SIZE * 8
1852    # Freeze the adaptive sizing so this test covers the cap, not the growth.
1853    session._peak_lead_seconds = 1e6
1854    logger = MagicMock()
1855    session.prov.logger = logger
1856    player = _make_late_joiner()
1857
1858    written: list[tuple[str, bytes]] = []
1859
1860    async def capture_write(target: Any, chunk: bytes) -> None:
1861        written.append((target.player_id, chunk))
1862
1863    def setup_with_feed(*_args: Any, **_kwargs: Any) -> None:
1864        _setup_stream(player)()
1865
1866        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1867            assert join is True
1868            # The group keeps being fed while the START ack is outstanding: this
1869            # is what pushes the joiner's due position off the back of the ring.
1870            await session._write_chunk_to_all_players(b"\x02" * int(2.0 * PCM_SAMPLE_SIZE))
1871            return start_unix_ms
1872
1873        player.stream.start = AsyncMock(side_effect=start)
1874
1875    with (
1876        patch.object(session, "_start_client", side_effect=setup_with_feed),
1877        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1878        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1879    ):
1880        await session.add_client(player)
1881
1882    # The anchor is the one that was commanded: an acked instant is never moved.
1883    headroom = AIRPLAY_LATE_JOIN_MIN_HEADROOM_MS / 1000
1884    assert _captured_start_at(player) == pytest.approx(now + headroom, abs=0.001)
1885    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1886    # due is 102.5s of feed and the write head reached 112.0s, so 9.5s is owed
1887    # while only 8s survives in the ring: the missing 1.5s opens as silence.
1888    assert len(prime) / PCM_SAMPLE_SIZE == pytest.approx(9.5, abs=0.001)
1889    pad = int(1.5 * PCM_SAMPLE_SIZE)
1890    assert prime[:pad] == bytes(pad), "the missing head must be silence"
1891    assert set(prime[pad:]) == {1, 2}, "the rest must be the buffered feed"
1892    assert pad % session._pcm_frame_size == 0
1893    assert session._client_skip_bytes[player.player_id] == 0
1894    # Position still reports where the GROUP is at that instant, because the
1895    # real content lands exactly where it would have without the shortfall.
1896    player.stream.rebase_position.assert_not_called()
1897    logger.warning.assert_called_once()
1898    assert "silence" in logger.warning.call_args.args[0]
1899
1900
1901@pytest.mark.asyncio
1902async def test_late_join_ring_shortfall_keeps_a_misaligned_ring_head_frame_aligned() -> None:
1903    """A silence pad over a mid-frame ring head still lands the feed on frame boundaries."""
1904    now = 1_000_000.0
1905    session = _make_session(now - 100.0, 110.0)
1906    # Both the write head and the ring head sit mid-frame.
1907    session._pcm_total_fed = int(110.0 * PCM_SAMPLE_SIZE) + 3
1908    session._pcm_buffer = bytearray(b"\x01" * (PCM_SAMPLE_SIZE * 8 + 2))
1909    session._pcm_buffer_max = len(session._pcm_buffer)
1910    session._peak_lead_seconds = 1e6
1911    player = _make_late_joiner()
1912
1913    written: list[tuple[str, bytes]] = []
1914
1915    async def capture_write(target: Any, chunk: bytes) -> None:
1916        written.append((target.player_id, chunk))
1917
1918    def setup_with_feed(*_args: Any, **_kwargs: Any) -> None:
1919        _setup_stream(player)()
1920
1921        async def start(start_unix_ms: int, _position_ms: int, *, join: bool = False) -> int:
1922            assert join is True
1923            await session._write_chunk_to_all_players(b"\x02" * (int(2.0 * PCM_SAMPLE_SIZE) + 1))
1924            return start_unix_ms
1925
1926        player.stream.start = AsyncMock(side_effect=start)
1927
1928    with (
1929        patch.object(session, "_start_client", side_effect=setup_with_feed),
1930        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
1931        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
1932    ):
1933        await session.add_client(player)
1934
1935    prime = next(chunk for pid, chunk in written if pid == player.player_id)
1936    frame_size = session._pcm_frame_size
1937    # The prime ends exactly at the write head, so its start - and therefore the
1938    # whole padded prime - has to sit on an absolute frame boundary.
1939    assert (session._pcm_total_fed - len(prime)) % frame_size == 0
1940    pad_len = len(prime) - len(prime.lstrip(b"\x00"))
1941    assert pad_len % frame_size == 0
1942
1943
1944def test_ring_grows_to_the_observed_write_head_lead() -> None:
1945    """The ring tracks the largest lead a session shows, above a floor and under a byte cap."""
1946    now = 1_000_000.0
1947    session = _make_session(now - 100.0, 100.0)
1948    assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1949
1950    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1951        # A 4s lead stays under the floor, which keeps the ring where it was.
1952        session.seconds_streamed = 104.0
1953        session._observe_write_head_lead()
1954        assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1955
1956        # A 15s lead carries the ring past the floor, with the margin on top.
1957        session.seconds_streamed = 115.0
1958        session._observe_write_head_lead()
1959        assert session._peak_lead_seconds == pytest.approx(15.0, abs=0.001)
1960        expected = int((15.0 + AIRPLAY_LATE_JOIN_RING_MARGIN_SECONDS) * PCM_SAMPLE_SIZE)
1961        assert session._pcm_buffer_max == expected
1962
1963        # A lead that falls back never shrinks the ring: the history a joiner
1964        # still needs is already in it.
1965        session.seconds_streamed = 108.0
1966        session._observe_write_head_lead()
1967        assert session._pcm_buffer_max == expected
1968
1969        # Growth is bounded in bytes, so a hi-res rate cannot multiply it out.
1970        session.seconds_streamed = 100.0 + 3600.0
1971        session._observe_write_head_lead()
1972        assert session._pcm_buffer_max == AIRPLAY_LATE_JOIN_RING_MAX_BYTES
1973
1974
1975def test_write_head_lead_is_not_measured_before_the_anchor_arrives() -> None:
1976    """Audio fed inside the start lead is not counted as pipeline depth."""
1977    now = 1_000_000.0
1978    # Anchored 2.5s into the future: nothing is audible yet.
1979    session = _make_session(now + 2.5, 8.0)
1980    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1981        session._observe_write_head_lead()
1982    assert session._peak_lead_seconds == 0.0
1983    assert session._pcm_buffer_max == int(AIRPLAY_LATE_JOIN_RING_MIN_SECONDS * PCM_SAMPLE_SIZE)
1984
1985    unanchored = _make_session(0.0, 8.0)
1986    with patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now):
1987        unanchored._observe_write_head_lead()
1988    assert unanchored._peak_lead_seconds == 0.0
1989
1990
1991@pytest.mark.asyncio
1992async def test_late_join_silence_pad_is_bounded_and_reports_the_residual() -> None:
1993    """An implausible ack is padded only up to the ring bound, and says so."""
1994    now = 1_000_000.0
1995    session = _make_session(now - 400.0, 420.0)
1996    session._pcm_total_fed = int(420.0 * PCM_SAMPLE_SIZE)
1997    session._pcm_buffer = bytearray(b"\x01" * PCM_SAMPLE_SIZE * 2)
1998    session._pcm_buffer_max = PCM_SAMPLE_SIZE * 2
1999    session._peak_lead_seconds = 1e6
2000    logger = MagicMock()
2001    session.prov.logger = logger
2002    player = _make_late_joiner()
2003
2004    written: list[tuple[str, bytes]] = []
2005
2006    async def capture_write(target: Any, chunk: bytes) -> None:
2007        written.append((target.player_id, chunk))
2008
2009    def setup_with_stale_ack(*_args: Any, **_kwargs: Any) -> None:
2010        _setup_stream(player)()
2011        # The binary reports an instant far behind the commanded one, mapping
2012        # the joiner back near the start of the session. 320s of head is owed.
2013        player.stream.start = AsyncMock(return_value=int((now - 300.0) * 1000))
2014
2015    with (
2016        patch.object(session, "_start_client", side_effect=setup_with_stale_ack),
2017        patch.object(session, "_write_chunk_to_player", side_effect=capture_write),
2018        patch("music_assistant.providers.airplay.stream_session.time.time", return_value=now),
2019    ):
2020        await session.add_client(player)
2021
2022    prime = next(chunk for pid, chunk in written if pid == player.player_id)
2023    # Bounded at one ring of silence plus the ring itself - never the 320s owed.
2024    assert len(prime) / PCM_SAMPLE_SIZE == pytest.approx(4.0, abs=0.01)
2025    assert (session._pcm_total_fed - len(prime)) % session._pcm_frame_size == 0
2026    # The joiner cannot be placed exactly, so the log must not claim sync.
2027    logger.warning.assert_called_once()
2028    tail = logger.warning.call_args.args[-1]
2029    assert "ahead of the group" in tail
2030    assert "in sync" not in tail
2031
2032
2033@pytest.mark.asyncio
2034async def test_start_client_releases_a_foreign_mute_latch() -> None:
2035    """A client joining the session gets its foreign mute latch released on start."""
2036    session = _make_session(start_time=0.0, seconds_streamed=0.0)
2037    player = _make_late_joiner()
2038
2039    with (
2040        patch(
2041            "music_assistant.providers.airplay.stream_session.AirPlayStream",
2042            return_value=MagicMock(connect=AsyncMock()),
2043        ),
2044        patch.object(session, "_start_player_ffmpeg", AsyncMock()),
2045    ):
2046        await session._start_client(player, use_shared_ptp=False)
2047
2048    player.release_foreign_mute_latch.assert_called_once_with()
2049