/
/
1"""Test that audio is only ever passed through unconverted when the bytes really match."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.enums import ContentType
8from music_assistant_models.media_items import AudioFormat
9
10from music_assistant.controllers.streams import audio_buffer as audio_buffer_module
11from music_assistant.controllers.streams.audio_buffer import AudioBuffer
12from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
13
14if TYPE_CHECKING:
15 from collections.abc import AsyncGenerator
16
17 import pytest
18
19_S32 = AudioFormat(content_type=ContentType.PCM_S32LE, sample_rate=44100, bit_depth=32, channels=2)
20_F32 = AudioFormat(content_type=ContentType.PCM_F32LE, sample_rate=44100, bit_depth=32, channels=2)
21
22
23def test_integer_and_float_pcm_are_not_interchangeable() -> None:
24 """
25 The gates below rely on the model telling PCM encodings apart.
26
27 Integer and float PCM of one depth share their rate, depth and channel count;
28 passing one through as the other reinterprets every sample.
29 """
30 assert _S32 != _F32
31 same_as_s32 = AudioFormat(
32 content_type=ContentType.PCM_S32LE, sample_rate=44100, bit_depth=32, channels=2
33 )
34 assert same_as_s32 == _S32
35
36
37async def test_the_buffer_converts_rather_than_reinterprets(
38 monkeypatch: pytest.MonkeyPatch,
39) -> None:
40 """A buffer of integer PCM asked for float PCM has to run its conversion."""
41 buffer = AudioBuffer(_S32)
42 took: list[str] = []
43
44 async def _fake_ffmpeg_stream(**_kwargs: object) -> AsyncGenerator[bytes]:
45 took.append("ffmpeg")
46 yield b""
47
48 async def _fake_raw_stream(**_kwargs: object) -> AsyncGenerator[bytes]:
49 took.append("raw")
50 yield b""
51
52 monkeypatch.setattr(audio_buffer_module, "get_ffmpeg_stream", _fake_ffmpeg_stream)
53 monkeypatch.setattr(buffer, "get_raw_stream", _fake_raw_stream)
54
55 async for _ in buffer.get_stream(output_format=_F32):
56 pass
57 assert took == ["ffmpeg"]
58
59 took.clear()
60 async for _ in buffer.get_stream(output_format=_S32):
61 pass
62 assert took == ["raw"]
63
64
65def test_a_warm_airplay_replace_refuses_a_differently_encoded_source() -> None:
66 """A live session must not absorb a source it would then mislabel."""
67 session = object.__new__(AirPlayStreamSession)
68 session.pcm_format = _F32
69 session.sync_clients = []
70 assert session.can_replace([], _S32) is False
71 assert session.can_replace([], _F32) is True
72