/
/
/
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_daemon(reader_attaches: bool) -> MagicMock:
11 """
12 Build a mock receiver daemon whose audio pipe reports the given reader state.
13
14 :param reader_attaches: Whether a consumer attaches to the audio pipe.
15 """
16 daemon = MagicMock()
17 daemon.audio_pipe.wait_for_reader = AsyncMock(return_value=reader_attaches)
18 daemon.audio_pipe.write = AsyncMock(return_value=True)
19 return daemon
20
21
22async def test_silence_nudge_waits_for_the_returning_reader() -> None:
23 """The nudge is delivered once the stream's consumer has reattached to the pipe."""
24 provider = MagicMock()
25 provider.logger = MagicMock()
26 daemon = _make_daemon(reader_attaches=True)
27
28 await AirPlayReceiverProvider._write_silence_to_unblock_stream(provider, daemon)
29
30 daemon.audio_pipe.wait_for_reader.assert_awaited_once()
31 # one second of PCM_S16LE stereo 44.1kHz, enough for the consumer to emit a chunk
32 assert len(daemon.audio_pipe.write.await_args.args[0]) == 176400
33
34
35async def test_silence_nudge_skipped_when_no_reader_returns() -> None:
36 """A pipe that nothing reads back is left alone instead of taking a dropped write."""
37 provider = MagicMock()
38 provider.logger = MagicMock()
39 daemon = _make_daemon(reader_attaches=False)
40
41 await AirPlayReceiverProvider._write_silence_to_unblock_stream(provider, daemon)
42
43 daemon.audio_pipe.write.assert_not_awaited()
44