/
/
1"""
2Tests for the crossfade volume-normalization pinning (get_queue_item_stream override).
3
4In per-item crossfade mode the next track is streamed twice: once as the crossfade fade-in
5(at prep time) and once as its own body. If the track's loudness measurement lands between
6those two calls, the body would flip from DYNAMIC (loudnorm) to MEASUREMENT_ONLY while the
7already-baked crossfade intro stays on DYNAMIC, producing an audible volume jump at the seam.
8
9``get_queue_item_stream`` accepts a ``normalization_override`` so the crossfade path can pin
10the body to the same mode the intro was baked with. These tests verify that override behaviour.
11"""
12
13from __future__ import annotations
14
15from collections.abc import AsyncGenerator
16from types import SimpleNamespace
17from typing import Any, cast
18from unittest.mock import AsyncMock, MagicMock
19
20import pytest
21from music_assistant_models.enums import ContentType, MediaType, VolumeNormalizationMode
22from music_assistant_models.media_items import AudioFormat
23from music_assistant_models.streamdetails import StreamDetails
24
25import music_assistant.controllers.streams.audio as audio_mod
26from music_assistant.controllers.streams.audio import StreamsAudio
27
28PCM_FORMAT = AudioFormat(
29 content_type=ContentType.PCM_S16LE,
30 codec_type=ContentType.PCM_S16LE,
31 sample_rate=44100,
32 bit_depth=16,
33 channels=2,
34)
35
36# integrated loudness as if a fresh measurement landed; with target -17 LUFS this yields a
37# MEASUREMENT_ONLY gain of -17 - (-11.4) = -5.6 dB
38MEASURED_LOUDNESS = -11.4
39TARGET_LOUDNESS = -17.0
40
41
42@pytest.fixture
43def audio(monkeypatch: pytest.MonkeyPatch) -> tuple[StreamsAudio, dict[str, list[str]]]:
44 """
45 Build a StreamsAudio with buffer/analysis/config mocked out.
46
47 Returns the controller plus a dict whose ``filter_params`` key captures the ffmpeg
48 filter chain the (faked) audio buffer was asked to apply.
49 """
50 captured: dict[str, list[str]] = {}
51
52 class _FakeBuffer:
53 """AudioBuffer test double: records the filter_params, yields no audio."""
54
55 has_error = False
56 pcm_format = PCM_FORMAT
57
58 @classmethod
59 async def get_buffer(cls, **_kwargs: Any) -> _FakeBuffer:
60 return cls()
61
62 async def get_stream(
63 self,
64 output_format: AudioFormat,
65 seek_position_ms: int = 0,
66 filter_params: list[str] | None = None,
67 ) -> AsyncGenerator[bytes]:
68 captured["filter_params"] = filter_params or []
69 empty: tuple[bytes, ...] = ()
70 for chunk in empty:
71 yield chunk
72
73 monkeypatch.setattr(audio_mod, "AudioBuffer", _FakeBuffer)
74 # when the override path is NOT taken, the mode is re-evaluated to MEASUREMENT_ONLY
75 monkeypatch.setattr(
76 audio_mod,
77 "get_normalization_mode",
78 lambda *_a, **_k: VolumeNormalizationMode.MEASUREMENT_ONLY,
79 )
80
81 controller = StreamsAudio(MagicMock())
82 mass = cast("MagicMock", controller.mass)
83 mass.streams.audio_analysis.get_audio_analysis = AsyncMock(
84 return_value=SimpleNamespace(loudness_integrated=MEASURED_LOUDNESS, loudness_album=None)
85 )
86 mass.streams.config.get_value.return_value = VolumeNormalizationMode.FALLBACK_DYNAMIC.value
87 mass.config.get_player_config = AsyncMock(return_value=MagicMock())
88 return controller, captured
89
90
91def _make_queue_item() -> MagicMock:
92 streamdetails = StreamDetails(
93 provider="test_provider",
94 item_id="track_b",
95 audio_format=PCM_FORMAT,
96 media_type=MediaType.TRACK,
97 )
98 streamdetails.queue_id = "queue_1"
99 streamdetails.target_loudness = TARGET_LOUDNESS
100 streamdetails.loudness = None
101 queue_item = MagicMock()
102 queue_item.media_type = MediaType.TRACK
103 queue_item.streamdetails = streamdetails
104 queue_item.name = "Track B"
105 return queue_item
106
107
108async def _drain(gen: AsyncGenerator[bytes]) -> None:
109 async for _ in gen:
110 pass
111
112
113@pytest.mark.asyncio
114async def test_dynamic_override_forces_loudnorm_and_ignores_measurement(
115 audio: tuple[StreamsAudio, dict[str, list[str]]],
116) -> None:
117 """A DYNAMIC override keeps the body on loudnorm even though a measurement is available."""
118 controller, captured = audio
119 queue_item = _make_queue_item()
120
121 await _drain(
122 controller.get_queue_item_stream(
123 queue_item,
124 PCM_FORMAT,
125 normalization_override=VolumeNormalizationMode.DYNAMIC,
126 )
127 )
128
129 # DYNAMIC -> loudnorm filter, NOT a static measurement gain
130 assert any(p.startswith("loudnorm") for p in captured["filter_params"])
131 assert queue_item.streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC
132 # the just-in-time hydration / re-evaluation is skipped entirely
133 mass = cast("MagicMock", controller.mass)
134 mass.streams.audio_analysis.get_audio_analysis.assert_not_called()
135 mass.streams.config.get_value.assert_not_called()
136
137
138@pytest.mark.asyncio
139async def test_no_override_reevaluates_to_measurement_only(
140 audio: tuple[StreamsAudio, dict[str, list[str]]],
141) -> None:
142 """Without an override, a landed measurement re-evaluates the body to MEASUREMENT_ONLY."""
143 controller, captured = audio
144 queue_item = _make_queue_item()
145
146 await _drain(controller.get_queue_item_stream(queue_item, PCM_FORMAT))
147
148 # MEASUREMENT_ONLY -> static volume gain of target - loudness = -17 - (-11.4) = -5.6 dB
149 assert "volume=-5.6dB" in captured["filter_params"]
150 assert (
151 queue_item.streamdetails.volume_normalization_mode
152 == VolumeNormalizationMode.MEASUREMENT_ONLY
153 )
154 # the measurement was hydrated from analysis
155 cast(
156 "MagicMock", controller.mass
157 ).streams.audio_analysis.get_audio_analysis.assert_called_once()
158