music-assistant-server

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