/
/
/
1"""Tests for the LoudnessAnalysisProvider._finalize return-value contract."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock
6
7import pytest
8from music_assistant_models.enums import MediaType
9
10from music_assistant.constants import CONF_LOG_LEVEL
11from music_assistant.models.audio_analysis import AudioAnalysisData, AudioAnalysisError
12from music_assistant.models.audio_analysis_provider import AnalysisSessionData
13from music_assistant.providers.loudness_analysis.provider import (
14 CONF_WRITE_REPLAYGAIN_TAGS,
15 MIN_DURATION_SECONDS,
16 LoudnessAnalysisProvider,
17 LoudnessSessionData,
18 _parse_ebur128_metrics,
19)
20
21
22def _make_provider() -> LoudnessAnalysisProvider:
23 """Construct a LoudnessAnalysisProvider with mocked MA infrastructure."""
24 mass = MagicMock()
25 mass.streams.audio_analysis.get_audio_analysis_version = AsyncMock(return_value=None)
26 mass.streams.audio_analysis.set_audio_analysis = AsyncMock()
27 manifest = MagicMock()
28 manifest.domain = "loudness_analysis"
29 config = MagicMock()
30 config.instance_id = "loudness_analysis_test"
31 config.get_value = MagicMock(return_value="GLOBAL")
32 config.values = {}
33 return LoudnessAnalysisProvider(mass, manifest, config, set())
34
35
36def _make_session_data() -> tuple[LoudnessSessionData, MagicMock]:
37 """Return (LoudnessSessionData with mocked ffmpeg, streamdetails mock)."""
38 streamdetails = MagicMock()
39 streamdetails.item_id = "track-1"
40 streamdetails.provider = "test_provider"
41 streamdetails.uri = "test://track-1"
42 streamdetails.media_type = MediaType.TRACK
43
44 ffmpeg = MagicMock()
45 ffmpeg.wait = AsyncMock()
46 ffmpeg.close = AsyncMock()
47 ffmpeg.write_eof = AsyncMock()
48 ffmpeg.log_history = []
49
50 session_data = LoudnessSessionData(ffmpeg=ffmpeg)
51 return session_data, streamdetails
52
53
54@pytest.mark.asyncio
55async def test_finalize_returns_analysis_on_success(monkeypatch: pytest.MonkeyPatch) -> None:
56 """_finalize must return AudioAnalysisData with the parsed metrics when analysis succeeds."""
57 provider = _make_provider()
58 session_id = "test-session-success"
59
60 session_data, streamdetails = _make_session_data()
61 session_data.chunks_received = MIN_DURATION_SECONDS + 1
62 session_data.eof_sent = True # already sent, _send_eof will be a no-op
63
64 provider._data[session_id] = session_data
65 provider._sessions[session_id] = AnalysisSessionData(
66 streamdetails=streamdetails,
67 audio_format=MagicMock(),
68 )
69
70 # Patch _parse_ebur128_metrics to return a valid result above the threshold
71 monkeypatch.setattr(
72 "music_assistant.providers.loudness_analysis.provider._parse_ebur128_metrics",
73 lambda _log: (-14.5, 7.2, -1.2),
74 )
75
76 result = await provider._finalize(session_id)
77
78 assert isinstance(result, AudioAnalysisData)
79 assert result.loudness_integrated == -14.5
80
81
82@pytest.mark.asyncio
83async def test_finalize_raises_when_below_threshold(monkeypatch: pytest.MonkeyPatch) -> None:
84 """_finalize must raise AudioAnalysisError when measured loudness is below the reliability floor."""
85 provider = _make_provider()
86 session_id = "test-session-quiet"
87
88 session_data, streamdetails = _make_session_data()
89 session_data.chunks_received = MIN_DURATION_SECONDS + 1
90 session_data.eof_sent = True
91
92 provider._data[session_id] = session_data
93 provider._sessions[session_id] = AnalysisSessionData(
94 streamdetails=streamdetails,
95 audio_format=MagicMock(),
96 )
97
98 # ebur128 reports ~-70 LUFS on near-silent tracks, below the reliability floor.
99 monkeypatch.setattr(
100 "music_assistant.providers.loudness_analysis.provider._parse_ebur128_metrics",
101 lambda _log: (-70.0, 5.0, -1.0),
102 )
103
104 with pytest.raises(AudioAnalysisError, match="quiet"):
105 await provider._finalize(session_id)
106
107
108@pytest.mark.asyncio
109async def test_finalize_raises_when_insufficient_duration() -> None:
110 """_finalize must raise AudioAnalysisError when chunks_received is below the minimum."""
111 provider = _make_provider()
112 session_id = "test-session-short"
113
114 session_data, streamdetails = _make_session_data()
115 session_data.chunks_received = MIN_DURATION_SECONDS - 1
116 session_data.eof_sent = True
117
118 provider._data[session_id] = session_data
119 provider._sessions[session_id] = AnalysisSessionData(
120 streamdetails=streamdetails,
121 audio_format=MagicMock(),
122 )
123
124 with pytest.raises(AudioAnalysisError, match="too short"):
125 await provider._finalize(session_id)
126
127
128# ---------------------------------------------------------------------------
129# post_analysis tests
130# ---------------------------------------------------------------------------
131
132
133def _make_loudness_provider(*, write_replaygain_tags: bool) -> LoudnessAnalysisProvider:
134 """Construct a LoudnessAnalysisProvider with a config gated on write_replaygain_tags."""
135 mass = MagicMock()
136 manifest = MagicMock()
137 manifest.domain = "loudness_analysis"
138 config = MagicMock()
139 config.instance_id = "loudness_analysis_test"
140 config.values = {}
141 config.get_value = MagicMock(
142 side_effect=lambda key: {
143 CONF_LOG_LEVEL: "GLOBAL",
144 CONF_WRITE_REPLAYGAIN_TAGS: write_replaygain_tags,
145 }.get(key, "GLOBAL")
146 )
147 return LoudnessAnalysisProvider(mass, manifest, config, supported_features=set())
148
149
150@pytest.mark.asyncio
151async def test_post_analysis_writes_tag_when_path_writable_and_config_on(
152 monkeypatch: pytest.MonkeyPatch,
153) -> None:
154 """post_analysis writes ReplayGain tag when path is filesystem-writable AND config is on."""
155 provider = _make_loudness_provider(write_replaygain_tags=True)
156 streamdetails = MagicMock()
157 streamdetails.path = "/music/test.flac"
158 analysis = AudioAnalysisData(loudness_integrated=-14.0)
159
160 write_mock = AsyncMock(return_value=True)
161 monkeypatch.setattr(
162 "music_assistant.providers.loudness_analysis.provider.write_replaygain_track_gain",
163 write_mock,
164 )
165
166 await provider.post_analysis(streamdetails, analysis)
167
168 # ReplayGain 2.0: track_gain_db = -18 - loudness_lufs = -18 - (-14) = -4
169 write_mock.assert_awaited_once_with("/music/test.flac", -4.0)
170
171
172@pytest.mark.asyncio
173async def test_post_analysis_skips_when_path_not_writable(
174 monkeypatch: pytest.MonkeyPatch,
175) -> None:
176 """post_analysis is a no-op when streamdetails.path is None or non-string."""
177 provider = _make_loudness_provider(write_replaygain_tags=True)
178 streamdetails = MagicMock()
179 streamdetails.path = None
180 analysis = AudioAnalysisData(loudness_integrated=-14.0)
181
182 write_mock = AsyncMock(return_value=True)
183 monkeypatch.setattr(
184 "music_assistant.providers.loudness_analysis.provider.write_replaygain_track_gain",
185 write_mock,
186 )
187
188 await provider.post_analysis(streamdetails, analysis)
189
190 write_mock.assert_not_awaited()
191
192
193@pytest.mark.asyncio
194async def test_post_analysis_skips_when_config_off(
195 monkeypatch: pytest.MonkeyPatch,
196) -> None:
197 """post_analysis is a no-op when write_replaygain_tags config is False."""
198 provider = _make_loudness_provider(write_replaygain_tags=False)
199 streamdetails = MagicMock()
200 streamdetails.path = "/music/test.flac"
201 analysis = AudioAnalysisData(loudness_integrated=-14.0)
202
203 write_mock = AsyncMock(return_value=True)
204 monkeypatch.setattr(
205 "music_assistant.providers.loudness_analysis.provider.write_replaygain_track_gain",
206 write_mock,
207 )
208
209 await provider.post_analysis(streamdetails, analysis)
210
211 write_mock.assert_not_awaited()
212
213
214@pytest.mark.asyncio
215async def test_post_analysis_skips_when_loudness_missing(
216 monkeypatch: pytest.MonkeyPatch,
217) -> None:
218 """post_analysis is a no-op when analysis.loudness_integrated is None."""
219 provider = _make_loudness_provider(write_replaygain_tags=True)
220 streamdetails = MagicMock()
221 streamdetails.path = "/music/test.flac"
222 analysis = AudioAnalysisData(loudness_integrated=None)
223
224 write_mock = AsyncMock(return_value=True)
225 monkeypatch.setattr(
226 "music_assistant.providers.loudness_analysis.provider.write_replaygain_track_gain",
227 write_mock,
228 )
229
230 await provider.post_analysis(streamdetails, analysis)
231
232 write_mock.assert_not_awaited()
233
234
235# ---------------------------------------------------------------------------
236# true peak measurement tests
237# ---------------------------------------------------------------------------
238
239# verbatim ffmpeg output, do not hand-edit
240_FFMPEG_SUMMARY_WITH_PEAK = [
241 "[Parsed_ebur128_0 @ 0x8b8c05080] Summary:",
242 "",
243 " Integrated loudness:",
244 " I: -21.8 LUFS",
245 " Threshold: -31.8 LUFS",
246 "",
247 " Loudness range:",
248 " LRA: 0.0 LU",
249 " Threshold: -41.8 LUFS",
250 " LRA low: -21.8 LUFS",
251 " LRA high: -21.8 LUFS",
252 "",
253 " True peak:",
254 " Peak: -18.1 dBFS",
255]
256
257
258def test_parse_metrics_extracts_true_peak_from_ffmpeg_summary() -> None:
259 """All three metrics must be parsed from a real ebur128 summary."""
260 integrated, lra, true_peak = _parse_ebur128_metrics(_FFMPEG_SUMMARY_WITH_PEAK)
261
262 assert integrated == -21.8
263 assert lra == 0.0
264 assert true_peak == -18.1
265
266
267@pytest.mark.asyncio
268async def test_start_analysis_requests_peak_measurement(
269 monkeypatch: pytest.MonkeyPatch,
270) -> None:
271 """ebur128 only reports true peak when explicitly asked, so the filter must request it."""
272 provider = _make_provider()
273 streamdetails = MagicMock()
274 streamdetails.volume_normalization_mode = None
275
276 fake_ffmpeg = MagicMock()
277 fake_ffmpeg.start = AsyncMock()
278 ffmpeg_cls = MagicMock(return_value=fake_ffmpeg)
279 monkeypatch.setattr("music_assistant.providers.loudness_analysis.provider.FFMpeg", ffmpeg_cls)
280
281 assert await provider._start_analysis("session-peak", streamdetails, MagicMock()) is True
282
283 filter_params = ffmpeg_cls.call_args.kwargs["filter_params"]
284 assert any("peak=true" in param for param in filter_params)
285