/
/
1"""Tests for the silence nudge that unblocks a stopped AirPlay Receiver stream."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock
6
7from music_assistant.providers.airplay_receiver import AirPlayReceiverProvider
8
9
10def _make_provider(reader_attaches: bool) -> MagicMock:
11 """
12 Build a mock provider whose audio pipe reports the given reader state.
13
14 :param reader_attaches: Whether a consumer attaches to the audio pipe.
15 """
16 provider = MagicMock()
17 provider.logger = MagicMock()
18 provider.audio_pipe.wait_for_reader = AsyncMock(return_value=reader_attaches)
19 provider.audio_pipe.write = AsyncMock(return_value=True)
20 return provider
21
22
23async def test_silence_nudge_waits_for_the_returning_reader() -> None:
24 """The nudge is delivered once the stream's consumer has reattached to the pipe."""
25 provider = _make_provider(reader_attaches=True)
26
27 await AirPlayReceiverProvider._write_silence_to_unblock_stream(provider)
28
29 provider.audio_pipe.wait_for_reader.assert_awaited_once()
30 # one second of PCM_S16LE stereo 44.1kHz, enough for the consumer to emit a chunk
31 assert len(provider.audio_pipe.write.await_args.args[0]) == 176400
32
33
34async def test_silence_nudge_skipped_when_no_reader_returns() -> None:
35 """A pipe that nothing reads back is left alone instead of taking a dropped write."""
36 provider = _make_provider(reader_attaches=False)
37
38 await AirPlayReceiverProvider._write_silence_to_unblock_stream(provider)
39
40 provider.audio_pipe.write.assert_not_awaited()
41