music-assistant-server

1"""Unit tests for the native AirPlay announcement orchestration."""
2
3import asyncio
4import logging
5import time
6from collections.abc import AsyncGenerator, Iterator
7from typing import Any
8from unittest.mock import AsyncMock, MagicMock, patch
9
10import pytest
11from music_assistant_models.enums import ContentType, PlaybackState
12from music_assistant_models.errors import PlayerCommandFailed
13from music_assistant_models.media_items import AudioFormat
14
15from music_assistant.providers.airplay import announce
16from music_assistant.providers.airplay.constants import (
17    AIRPLAY_ANNOUNCE_AT_MARGIN_MS,
18    AIRPLAY_ANNOUNCE_DUCK_DB,
19    AIRPLAY_ANNOUNCE_FALLBACK_SPAN_MS,
20    AIRPLAY_PCM_FORMAT,
21)
22
23ANNOUNCE_DATA = {
24    "announcement_url": "http://ma.local/tts.mp3",
25    "pre_announce": True,
26    "pre_announce_url": "http://ma.local/chime.mp3",
27    "announce_player_id": None,
28}
29HIRES_PCM_FORMAT = AudioFormat(content_type=ContentType.PCM_S32LE, sample_rate=48000, bit_depth=24)
30
31
32@pytest.fixture(autouse=True)
33def _zero_restore_pad() -> Iterator[None]:
34    """
35    Zero the audible-end pad for every test.
36
37    The live path holds its return (and the volume restore) until the clip's
38    audible end plus this pad; the streams below ack a long-past instant, so
39    only the pad would otherwise add real wall-clock time to every test. The
40    audible-end test overrides it with its own value.
41    """
42    with patch.object(announce, "AIRPLAY_ANNOUNCE_VOLUME_RESTORE_PAD_MS", 0):
43        yield
44
45
46def _make_render(duration: float = 1.5) -> MagicMock:
47    """Build a mock announcement render that finished with the given duration."""
48    render = MagicMock()
49    render.duration = duration
50    render.wait_finished = AsyncMock(return_value=duration)
51
52    async def get_stream(_output_format: Any) -> AsyncGenerator[bytes]:
53        yield b"\x00" * 64
54
55    render.get_stream = get_stream
56    return render
57
58
59def _make_stream(
60    pcm_format: AudioFormat = AIRPLAY_PCM_FORMAT,
61    ack: tuple[int, int] | None = (1, 1),
62) -> MagicMock:
63    """
64    Build a mock member stream whose announce arm resolves with the given ack.
65
66    The default ack reports a long-past audible instant, so the audible-end
67    hold at the end of the live path resolves without waiting.
68    """
69    stream = MagicMock()
70    stream.running = True
71    stream.connected = True
72    stream.pcm_format = pcm_format
73    stream.warm_lead_ms = 0
74    stream.latency_lead_ms = 0
75    stream.announce = AsyncMock(return_value=True)
76    stream.wait_announce_started = AsyncMock(return_value=ack)
77    stream.wait_announce_done = AsyncMock(return_value=True)
78    return stream
79
80
81def _make_player(player_id: str, stream: MagicMock | None = None) -> MagicMock:
82    """Build a mock AirPlay player for announcement orchestration tests."""
83    player = MagicMock()
84    player.player_id = player_id
85    player.display_name = player_id
86    player.synced_to = None
87    player.state.active_group = None
88    player.protocol_parent_id = None
89    player.playback_state = PlaybackState.PLAYING
90    player.volume_level = 30
91    player.volume_set = AsyncMock()
92    player.stream = stream
93    player._lock = asyncio.Lock()
94    player.logger = logging.getLogger("test.airplay.announce")
95    player.mass.create_task = MagicMock(
96        side_effect=lambda coro, *_args, **_kwargs: asyncio.get_running_loop().create_task(coro)
97    )
98    player.provider._announce_plans = {}
99    player.provider.bridge_manager.get_bridge = MagicMock(return_value=None)
100    renderer = player.mass.streams.announcement_renderer
101    renderer.acquire = MagicMock(return_value=_make_render())
102    renderer.release = AsyncMock()
103    return player
104
105
106def _make_playing_group(*streams: MagicMock) -> list[MagicMock]:
107    """Build a playing session of one member per given stream; first one leads."""
108    members = [_make_player(f"member_{i}", stream=stream) for i, stream in enumerate(streams)]
109    session = MagicMock()
110    session.sync_clients = members
111    for member in members:
112        member.stream.session = session
113    return members
114
115
116def _make_announcement() -> MagicMock:
117    """Build the announcement PlayerMedia handed down by the player controller."""
118    return MagicMock(custom_data=dict(ANNOUNCE_DATA))
119
120
121def test_member_span_prefers_the_warm_lead() -> None:
122    """A splice-timeline member's span is its warm lead, not the device lead."""
123    stream = _make_stream()
124    stream.warm_lead_ms = 600
125    stream.latency_lead_ms = 1900
126
127    assert announce._member_span_ms(stream) == 600
128
129
130def test_member_span_uses_the_device_lead_without_a_warm_lead() -> None:
131    """Without a warm lead, the reported device lead bounds the delivery head."""
132    stream = _make_stream()
133    stream.latency_lead_ms = 1900
134
135    assert announce._member_span_ms(stream) == 1900
136
137
138def test_member_span_falls_back_when_nothing_was_reported() -> None:
139    """Both leads at 0 mean unreported: assume the binary's default playback lead."""
140    assert announce._member_span_ms(_make_stream()) == AIRPLAY_ANNOUNCE_FALLBACK_SPAN_MS
141
142
143def test_shared_instant_covers_the_slowest_member() -> None:
144    """The shared instant sits past the LARGEST member span plus the fan-out margin."""
145    fast = _make_stream()
146    fast.warm_lead_ms = 600
147    slow = _make_stream()
148    slow.latency_lead_ms = 1900
149    unreported = _make_stream()
150
151    before_ms = int(time.time() * 1000)
152    at_unix_ms = announce._shared_announce_instant([fast, slow, unreported])
153    after_ms = time.time() * 1000
154
155    # the unreported member's fallback span (2000) is the largest of the three
156    expected_lead = AIRPLAY_ANNOUNCE_FALLBACK_SPAN_MS + AIRPLAY_ANNOUNCE_AT_MARGIN_MS
157    assert before_ms + expected_lead <= at_unix_ms <= after_ms + expected_lead
158
159
160@pytest.mark.asyncio
161async def test_live_announcement_arms_every_member_at_one_shared_instant() -> None:
162    """A playing session arms every member with the same instant, duck and clip file."""
163    streams = [_make_stream(), _make_stream()]
164    members = _make_playing_group(*streams)
165    leader = members[0]
166
167    with patch.object(announce, "_announce_with_session", new_callable=AsyncMock) as session_path:
168        await announce.play_announcement(leader, _make_announcement(), None)
169
170    session_path.assert_not_awaited()
171    arms = [stream.announce.call_args for stream in streams]
172    for arm in arms:
173        assert arm is not None
174    # one shared instant and the default duck for every member
175    assert len({arm.args[1] for arm in arms}) == 1
176    assert {arm.args[2] for arm in arms} == {AIRPLAY_ANNOUNCE_DUCK_DB}
177    # both members share one stdin format, so they share one clip file
178    assert len({arm.args[0] for arm in arms}) == 1
179    assert arms[0].args[0].endswith(".pcm")
180    for stream in streams:
181        stream.wait_announce_done.assert_awaited_once()
182    leader.mass.streams.announcement_renderer.release.assert_awaited_once()
183
184
185@pytest.mark.asyncio
186async def test_live_announcement_renders_one_clip_per_distinct_format() -> None:
187    """Members on different stdin formats each get a clip in exactly their format."""
188    streams = [_make_stream(), _make_stream(pcm_format=HIRES_PCM_FORMAT)]
189    members = _make_playing_group(*streams)
190
191    with patch.object(announce, "_announce_with_session", new_callable=AsyncMock):
192        await announce.play_announcement(members[0], _make_announcement(), None)
193
194    clip_paths = {stream.announce.call_args.args[0] for stream in streams}
195    assert len(clip_paths) == 2
196
197
198@pytest.mark.asyncio
199async def test_idle_player_takes_the_session_path() -> None:
200    """Without live playback the announcement runs as its own stream session."""
201    player = _make_player("solo")
202    player.playback_state = PlaybackState.IDLE
203    announcement = _make_announcement()
204
205    with patch.object(announce, "_announce_with_session", new_callable=AsyncMock) as session_path:
206        await announce.play_announcement(player, announcement, 40)
207
208    session_path.assert_awaited_once()
209    args = session_path.call_args.args
210    assert args[0] is player
211    assert args[1] is announcement
212    assert args[3] == 1.5  # the render's exact duration bounds the session waits
213    assert args[4] == 40
214
215
216@pytest.mark.asyncio
217async def test_no_member_arming_fails_without_killing_the_music() -> None:
218    """
219    Live members that never arm the clip fail the announcement, not the music.
220
221    An outdated cliairplay silently ignores the arm command; a fallback session
222    would stop the user's playing session over a clip that could not be mixed.
223    """
224    streams = [_make_stream(ack=None), _make_stream(ack=None)]
225    members = _make_playing_group(*streams)
226    live_session = streams[0].session
227    live_session.stop = AsyncMock()
228
229    with (
230        patch.object(announce, "_announce_with_session", new_callable=AsyncMock) as session_path,
231        pytest.raises(PlayerCommandFailed, match="may not support announcements"),
232    ):
233        await announce.play_announcement(members[0], _make_announcement(), None)
234
235    for stream in streams:
236        stream.announce.assert_awaited_once()
237        stream.wait_announce_done.assert_not_awaited()
238    session_path.assert_not_awaited()
239    live_session.stop.assert_not_awaited()
240    members[0].mass.streams.announcement_renderer.release.assert_awaited_once()
241
242
243@pytest.mark.asyncio
244async def test_partial_success_warns_and_does_not_fall_back(
245    caplog: pytest.LogCaptureFixture,
246) -> None:
247    """One member playing the clip is a success; the members that did not are named."""
248    streams = [_make_stream(), _make_stream(ack=None)]
249    members = _make_playing_group(*streams)
250
251    with (
252        patch.object(announce, "_announce_with_session", new_callable=AsyncMock) as session_path,
253        caplog.at_level(logging.WARNING),
254    ):
255        await announce.play_announcement(members[0], _make_announcement(), None)
256
257    session_path.assert_not_awaited()
258    assert "member_1" in caplog.text
259    streams[0].wait_announce_done.assert_awaited_once()
260    streams[1].wait_announce_done.assert_not_awaited()
261
262
263@pytest.mark.asyncio
264async def test_volume_is_scheduled_on_the_acked_instant() -> None:
265    """The volume bump lands at the acked audible instant, the restore after the clip."""
266    ack_at_unix_ms = int(time.time() * 1000) + 200
267    # the acked duration includes the 1s ducked-silence tail: 2s of content
268    stream = _make_stream(ack=(ack_at_unix_ms, 3000))
269    members = _make_playing_group(stream)
270    member = members[0]
271    member.volume_level = 30
272
273    with patch.object(announce, "_announce_with_session", new_callable=AsyncMock):
274        await announce.play_announcement(member, _make_announcement(), 55)
275
276    scheduled = member.mass.call_later.call_args_list
277    assert len(scheduled) == 2
278    bump, restore = scheduled
279    assert bump.args[1:] == (member.volume_set, 55)
280    assert restore.args[1:] == (member.volume_set, 30)
281    # the bump derives from the acked instant (~0.2s out) plus the 0.3s
282    # into-the-clip bias; the restore follows the CONTENT length (the acked
283    # duration minus the silence tail) plus the (zeroed) pad, so it lands
284    # inside the ducked cushion - 1.7s after the bump here
285    assert 0.2 < bump.args[0] <= 0.5
286    assert restore.args[0] - bump.args[0] == pytest.approx(1.7)
287
288
289def test_member_duck_compensates_the_volume_bump() -> None:
290    """
291    The duck deepens by exactly the device-volume bump so the music never rises.
292
293    38 -> 61 volume points is +6.9 dB on the AirPlay dB scale; the -18 dB duck
294    becomes -24.9 dB so the music's perceived level stays at the configured
295    duck depth (the regression heard as "the music was not ducked"). A bump
296    down shallows it symmetrically, and without a bump the base duck applies.
297    """
298    member = _make_player("m")
299    member.volume_level = 38
300    assert announce._member_duck_db(member, 61) == pytest.approx(-24.9)
301    assert announce._member_duck_db(member, 18) == pytest.approx(-12.0)
302    assert announce._member_duck_db(member, None) == pytest.approx(-18.0)
303    assert announce._member_duck_db(member, 38) == pytest.approx(-18.0)
304    # extreme bumps clamp to the binary's usable range (never boost the music)
305    member.volume_level = 0
306    assert announce._member_duck_db(member, 100) == pytest.approx(-48.0)
307    assert announce._member_duck_db(member, 0) == pytest.approx(-18.0)
308    member.volume_level = 100
309    assert announce._member_duck_db(member, 0) == pytest.approx(0.0)
310
311
312@pytest.mark.asyncio
313async def test_no_volume_level_leaves_the_volume_alone() -> None:
314    """Without an announcement volume nothing is scheduled on any member."""
315    members = _make_playing_group(_make_stream())
316
317    with patch.object(announce, "_announce_with_session", new_callable=AsyncMock):
318        await announce.play_announcement(members[0], _make_announcement(), None)
319
320    members[0].mass.call_later.assert_not_called()
321
322
323@pytest.mark.asyncio
324async def test_session_path_plays_the_clip_and_restores_the_volume() -> None:
325    """The dedicated session serves the clip to the configured group at the given volume."""
326    player = _make_player("solo")
327    player.playback_state = PlaybackState.IDLE
328    player.volume_level = 25
329    player._get_sync_clients = MagicMock(return_value=[player])
330    player._get_session_pcm_format = AsyncMock(return_value=AIRPLAY_PCM_FORMAT)
331    announcement = _make_announcement()
332    render = _make_render(duration=0.01)
333    player.mass.streams.announcement_renderer.acquire = MagicMock(return_value=render)
334
335    with (
336        patch.object(announce, "AirPlayStreamSession") as session_cls,
337        patch.object(announce, "AIRPLAY_ANNOUNCE_SESSION_DRAIN_S", 0.0),
338    ):
339        session = session_cls.return_value
340        session.start = AsyncMock()
341        session.stop = AsyncMock()
342        session.start_time = 0.0
343        await announce.play_announcement(player, announcement, 60)
344
345    assert session_cls.call_args.args == (
346        player.provider,
347        [player],
348        AIRPLAY_PCM_FORMAT,
349        announcement,
350    )
351    session.start.assert_awaited_once()
352    session.stop.assert_awaited_once()
353    # announcement volume before the session, previous level restored after
354    assert [call.args for call in player.volume_set.await_args_list] == [(60,), (25,)]
355    # the player ends idle without still showing media (like player.stop())
356    assert player._attr_current_media is None
357    player.update_state.assert_called()
358
359
360@pytest.mark.asyncio
361async def test_session_path_stops_a_parked_session_first() -> None:
362    """A parked (paused) session is stopped before the dedicated announcement session."""
363    parked_stream = _make_stream()
364    player = _make_player("solo", stream=parked_stream)
365    player.playback_state = PlaybackState.PAUSED
366    player._get_sync_clients = MagicMock(return_value=[player])
367    player._get_session_pcm_format = AsyncMock(return_value=AIRPLAY_PCM_FORMAT)
368    parked_session = parked_stream.session
369    parked_session.stop = AsyncMock()
370    render = _make_render(duration=0.01)
371    player.mass.streams.announcement_renderer.acquire = MagicMock(return_value=render)
372
373    with (
374        patch.object(announce, "AirPlayStreamSession") as session_cls,
375        patch.object(announce, "AIRPLAY_ANNOUNCE_SESSION_DRAIN_S", 0.0),
376    ):
377        session = session_cls.return_value
378        session.start = AsyncMock()
379        session.stop = AsyncMock()
380        session.start_time = 0.0
381        await announce.play_announcement(player, _make_announcement(), None)
382
383    parked_session.stop.assert_awaited_once()
384    assert player.stream is None
385    session.start.assert_awaited_once()
386
387
388@pytest.mark.asyncio
389async def test_group_entity_fanout_arms_each_member_at_one_shared_instant() -> None:
390    """
391    A group-entity fan-out arms each member once, at one shared instant.
392
393    The controller forwards a group-entity announcement per member; each call
394    arms only its OWN member, and all of them share one audible instant through
395    the provider's plan registry, so every room renders in sync.
396    """
397    streams = [_make_stream(), _make_stream(), _make_stream()]
398    members = _make_playing_group(*streams)
399    leader, member_1, member_2 = members
400    provider = leader.provider
401    render = _make_render()
402    for member in members:
403        member.provider = provider
404        member.state.active_group = "syncgroup_1"
405        if member is not leader:
406            member.synced_to = leader.player_id
407        member.mass.streams.announcement_renderer.acquire = MagicMock(return_value=render)
408    # the second member reports a far larger span: the shared instant must
409    # clear it for every sibling arm
410    streams[1].latency_lead_ms = 2000
411    announcement = _make_announcement()
412
413    await asyncio.gather(
414        announce.play_announcement(leader, announcement, None),
415        announce.play_announcement(member_1, announcement, None),
416        announce.play_announcement(member_2, announcement, None),
417    )
418
419    instants = set()
420    for stream in streams:
421        stream.announce.assert_awaited_once()
422        stream.wait_announce_done.assert_awaited_once()
423        instants.add(stream.announce.await_args.args[1])
424    assert len(instants) == 1
425    # the shared instant cleared the largest member span (2s + margin)
426    assert next(iter(instants)) >= int(time.time() * 1000) + 2000
427
428
429@pytest.mark.asyncio
430async def test_group_entity_session_leader_announces_alone() -> None:
431    """
432    A group entity's session leader addressed individually announces alone.
433
434    The entity itself is the whole-group handle, so leading the underlying
435    session does not widen an individual announcement.
436    """
437    streams = [_make_stream(), _make_stream()]
438    members = _make_playing_group(*streams)
439    leader = members[0]
440    leader.state.active_group = "syncgroup_1"
441    members[1].synced_to = leader.player_id
442
443    await announce.play_announcement(leader, _make_announcement(), None)
444
445    streams[0].announce.assert_awaited_once()
446    streams[1].announce.assert_not_awaited()
447
448
449@pytest.mark.asyncio
450async def test_protocol_child_reads_group_ownership_from_its_parent() -> None:
451    """
452    A protocol child leading a syncgroup's session still announces alone.
453
454    Protocol players never carry active_group themselves - the model keeps the
455    group state on the device player they render for - so the whole-group
456    handle is found through the protocol parent (the regression heard as an
457    individual announcement playing on the whole syncgroup).
458    """
459    streams = [_make_stream(), _make_stream()]
460    members = _make_playing_group(*streams)
461    leader = members[0]
462    leader.protocol_parent_id = "milo_parent"
463    parent = MagicMock()
464    parent.state.active_group = "syncgroup_1"
465    leader.mass.players.get_player = MagicMock(return_value=parent)
466    members[1].synced_to = leader.player_id
467
468    await announce.play_announcement(leader, _make_announcement(), None)
469
470    streams[0].announce.assert_awaited_once()
471    streams[1].announce.assert_not_awaited()
472
473
474@pytest.mark.asyncio
475async def test_member_announced_directly_arms_itself() -> None:
476    """A direct announcement to one synced member mixes on just that member."""
477    stream = _make_stream()
478    members = _make_playing_group(stream)
479    member = members[0]
480    member.synced_to = "some_leader"
481
482    await announce.play_announcement(member, _make_announcement(), None)
483
484    stream.announce.assert_awaited_once()
485
486
487@pytest.mark.asyncio
488async def test_synced_member_without_live_playback_is_refused() -> None:
489    """
490    A parked group member must not announce by tearing down the shared session.
491
492    Its stream belongs to the leader's parked session; stopping that for a
493    single-member announcement would silence every room in the group.
494    """
495    parked_stream = _make_stream()
496    members = _make_playing_group(parked_stream)
497    member = members[0]
498    member.synced_to = "some_leader"
499    member.playback_state = PlaybackState.PAUSED
500    parked_session = parked_stream.session
501    parked_session.stop = AsyncMock()
502
503    with pytest.raises(PlayerCommandFailed, match="without live playback"):
504        await announce.play_announcement(member, _make_announcement(), None)
505
506    parked_session.stop.assert_not_awaited()
507
508
509@pytest.mark.asyncio
510async def test_return_holds_until_the_audible_end() -> None:
511    """
512    The call returns only once the clip is audibly over, not when it is mixed.
513
514    announce_done reports MIX completion at the delivery head - ahead of
515    audibility - and the caller re-mutes muted players the moment this returns.
516    """
517    now_unix_ms = int(time.time() * 1000)
518    stream = _make_stream(ack=(now_unix_ms + 250, 100))
519    members = _make_playing_group(stream)
520
521    with patch.object(announce, "AIRPLAY_ANNOUNCE_VOLUME_RESTORE_PAD_MS", 100):
522        started = time.monotonic()
523        await announce.play_announcement(members[0], _make_announcement(), None)
524        elapsed = time.monotonic() - started
525
526    # audible end = acked instant + clip duration (0.35s out) + the 0.1s pad
527    assert elapsed >= 0.4
528
529
530@pytest.mark.asyncio
531async def test_bridged_player_with_live_stream_arms_itself() -> None:
532    """A Sendspin-bridged player mixes the clip into the bridge-owned stream."""
533    stream = _make_stream()
534    stream.session = None
535    player = _make_player("bridged", stream=stream)
536    bridge = MagicMock(owns_airplay_stream=True)
537    player.provider.bridge_manager.get_bridge = MagicMock(return_value=bridge)
538
539    await announce.play_announcement(player, _make_announcement(), None)
540
541    stream.announce.assert_awaited_once()
542    stream.wait_announce_done.assert_awaited_once()
543
544
545@pytest.mark.asyncio
546async def test_bridge_configured_player_mixes_over_its_own_session() -> None:
547    """
548    A bridge that is merely configured never blocks the live mix.
549
550    The regression this pins: a player with a Sendspin bridge set up but
551    playing its own (session-backed) AirPlay stream mixes the clip like any
552    unbridged player - the idle bridge is a bystander.
553    """
554    (member,) = _make_playing_group(_make_stream())
555    bridge = MagicMock(owns_airplay_stream=False)
556    member.provider.bridge_manager.get_bridge = MagicMock(return_value=bridge)
557
558    await announce.play_announcement(member, _make_announcement(), None)
559
560    member.stream.announce.assert_awaited_once()
561    member.stream.wait_announce_done.assert_awaited_once()
562
563
564@pytest.mark.asyncio
565async def test_bridge_streaming_player_never_runs_a_session_fallback() -> None:
566    """A player whose bridge owns the live stream fails instead of seizing the device."""
567    player = _make_player("bridged")
568    player.playback_state = PlaybackState.IDLE
569    bridge = MagicMock(owns_airplay_stream=True)
570    player.provider.bridge_manager.get_bridge = MagicMock(return_value=bridge)
571
572    with (
573        patch.object(announce, "AirPlayStreamSession") as session_cls,
574        pytest.raises(PlayerCommandFailed, match="Sendspin bridge"),
575    ):
576        await announce.play_announcement(player, _make_announcement(), None)
577
578    session_cls.assert_not_called()
579
580
581@pytest.mark.asyncio
582async def test_announcement_without_data_is_refused() -> None:
583    """An announcement without its announce data cannot be rendered."""
584    player = _make_player("solo")
585
586    with pytest.raises(PlayerCommandFailed, match="carries no announcement data"):
587        await announce.play_announcement(player, MagicMock(custom_data=None), None)
588
589
590@pytest.mark.asyncio
591async def test_announcement_without_audio_plays_nothing(
592    caplog: pytest.LogCaptureFixture,
593) -> None:
594    """A render that produced no audio is skipped, releasing the render regardless."""
595    stream = _make_stream()
596    members = _make_playing_group(stream)
597    player = members[0]
598    render = _make_render(duration=0.0)
599    player.mass.streams.announcement_renderer.acquire = MagicMock(return_value=render)
600
601    with caplog.at_level(logging.WARNING):
602        await announce.play_announcement(player, _make_announcement(), None)
603
604    stream.announce.assert_not_awaited()
605    assert "produced no audio" in caplog.text
606    player.mass.streams.announcement_renderer.release.assert_awaited_once()
607