/
/
/
1"""Test that audio is only ever passed through unconverted when the bytes really match."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import TYPE_CHECKING, cast
7
8from music_assistant_models.enums import ContentType, MediaType, StreamType
9from music_assistant_models.media_items import AudioFormat
10from music_assistant_models.streamdetails import StreamDetails
11
12from music_assistant.controllers.streams import audio_buffer as audio_buffer_module
13from music_assistant.controllers.streams.audio_buffer import AudioBuffer
14from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
15
16if TYPE_CHECKING:
17 from collections.abc import AsyncGenerator
18
19 import pytest
20
21 from music_assistant.controllers.streams.audio import StreamsAudio
22
23_S32 = AudioFormat(content_type=ContentType.PCM_S32LE, sample_rate=44100, bit_depth=32, channels=2)
24_F32 = AudioFormat(content_type=ContentType.PCM_F32LE, sample_rate=44100, bit_depth=32, channels=2)
25
26
27def test_integer_and_float_pcm_are_not_interchangeable() -> None:
28 """
29 The gates below rely on the model telling PCM encodings apart.
30
31 Integer and float PCM of one depth share their rate, depth and channel count;
32 passing one through as the other reinterprets every sample.
33 """
34 assert _S32 != _F32
35 same_as_s32 = AudioFormat(
36 content_type=ContentType.PCM_S32LE, sample_rate=44100, bit_depth=32, channels=2
37 )
38 assert same_as_s32 == _S32
39
40
41async def test_the_buffer_converts_rather_than_reinterprets(
42 monkeypatch: pytest.MonkeyPatch,
43) -> None:
44 """A buffer of integer PCM asked for float PCM has to run its conversion."""
45 buffer = AudioBuffer(_S32)
46 took: list[str] = []
47
48 async def _fake_ffmpeg_stream(**_kwargs: object) -> AsyncGenerator[bytes]:
49 took.append("ffmpeg")
50 yield b""
51
52 async def _fake_raw_stream(**_kwargs: object) -> AsyncGenerator[bytes]:
53 took.append("raw")
54 yield b""
55
56 monkeypatch.setattr(audio_buffer_module, "get_ffmpeg_stream", _fake_ffmpeg_stream)
57 monkeypatch.setattr(buffer, "get_raw_stream", _fake_raw_stream)
58
59 async for _ in buffer.get_stream(output_format=_F32):
60 pass
61 assert took == ["ffmpeg"]
62
63 took.clear()
64 async for _ in buffer.get_stream(output_format=_S32):
65 pass
66 assert took == ["raw"]
67
68
69def test_a_warm_airplay_replace_refuses_a_differently_encoded_source() -> None:
70 """A live session must not absorb a source it would then mislabel."""
71 session = object.__new__(AirPlayStreamSession)
72 session.pcm_format = _F32
73 session.sync_clients = []
74 assert session.can_replace([], _S32) is False
75 assert session.can_replace([], _F32) is True
76
77
78async def _which_path(advertised: AudioFormat, arriving: AudioFormat | None) -> str:
79 """Return which branch of the AudioSource gate a live source is routed through."""
80 from music_assistant.controllers.streams.audio import StreamsAudio # noqa: PLC0415
81
82 took: list[str] = []
83
84 async def _bytes() -> AsyncGenerator[bytes]:
85 yield b"\x00" * 8
86
87 def _fake_open(_streamdetails: object) -> AsyncGenerator[bytes]:
88 took.append("raw")
89 return _bytes()
90
91 async def _fake_ffmpeg(**_kwargs: object) -> AsyncGenerator[bytes]:
92 took.append("ffmpeg")
93 yield b"\x00" * 8
94
95 controller = cast(
96 "StreamsAudio",
97 SimpleNamespace(_open_audio_source_generator=_fake_open, get_media_stream=_fake_ffmpeg),
98 )
99 streamdetails = StreamDetails(
100 provider="test--1",
101 item_id="1",
102 audio_format=advertised,
103 decoded_audio_format=arriving,
104 media_type=MediaType.AUDIO_SOURCE,
105 stream_type=StreamType.NAMED_PIPE,
106 path="/fake/fifo",
107 )
108 async for _ in StreamsAudio._iter_audio_source_pcm(controller, streamdetails, _S32):
109 pass
110 return took[0]
111
112
113async def test_a_codec_advertising_live_source_keeps_ffmpeg_in_the_path() -> None:
114 """
115 The gate reads the advertised format on purpose, not the arriving one.
116
117 A live source that advertises a codec relies on ffmpeg to notice its
118 producer going away - shairport-sync leaves the pipe behind on an unclean
119 disconnect, which a direct read would simply reopen and wait on.
120 """
121 advertised = AudioFormat(
122 content_type=ContentType.OGG,
123 codec_type=ContentType.VORBIS,
124 sample_rate=44100,
125 bit_depth=16,
126 channels=2,
127 bit_rate=160,
128 )
129 assert await _which_path(advertised, _S32) == "ffmpeg"
130
131
132async def test_a_pcm_advertising_live_source_is_read_directly() -> None:
133 """A source that states the PCM it delivers is handed through unconverted."""
134 assert await _which_path(_S32, None) == "raw"
135