/
/
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.helpers.audio import pcm_formats_match
13from music_assistant.providers.airplay.stream_session import AirPlayStreamSession
14
15if TYPE_CHECKING:
16 from collections.abc import AsyncGenerator
17
18 import pytest
19
20_S32 = AudioFormat(content_type=ContentType.PCM_S32LE, sample_rate=44100, bit_depth=32, channels=2)
21_F32 = AudioFormat(content_type=ContentType.PCM_F32LE, sample_rate=44100, bit_depth=32, channels=2)
22
23
24def test_integer_and_float_pcm_are_not_interchangeable() -> None:
25 """
26 Equality cannot be used to decide a passthrough.
27
28 Every PCM content type renders the same ``output_format_str``, so the model's
29 own ``==`` reports integer and float PCM of one depth as the same format;
30 passing one through as the other reinterprets every sample.
31 """
32 assert _S32 == _F32 # the trap this predicate exists to avoid
33 assert not pcm_formats_match(_S32, _F32)
34 assert pcm_formats_match(_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