/
/
/
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 exact_seek: bool = False,
68 ) -> AsyncGenerator[bytes]:
69 del output_format, seek_position_ms, exact_seek
70 captured["filter_params"] = filter_params or []
71 empty: tuple[bytes, ...] = ()
72 for chunk in empty:
73 yield chunk
74
75 monkeypatch.setattr(audio_mod, "AudioBuffer", _FakeBuffer)
76 # when the override path is NOT taken, the mode is re-evaluated to MEASUREMENT_ONLY
77 monkeypatch.setattr(
78 audio_mod,
79 "get_normalization_mode",
80 lambda *_a, **_k: VolumeNormalizationMode.MEASUREMENT_ONLY,
81 )
82
83 controller = StreamsAudio(MagicMock())
84 mass = cast("MagicMock", controller.mass)
85 mass.streams.audio_analysis.get_audio_analysis = AsyncMock(
86 return_value=SimpleNamespace(loudness_integrated=MEASURED_LOUDNESS, loudness_album=None)
87 )
88 mass.streams.config.get_value.return_value = VolumeNormalizationMode.FALLBACK_DYNAMIC.value
89 mass.config.get_player_config = AsyncMock(return_value=MagicMock())
90 return controller, captured
91
92
93def _make_queue_item() -> MagicMock:
94 streamdetails = StreamDetails(
95 provider="test_provider",
96 item_id="track_b",
97 audio_format=PCM_FORMAT,
98 media_type=MediaType.TRACK,
99 )
100 streamdetails.queue_id = "queue_1"
101 streamdetails.target_loudness = TARGET_LOUDNESS
102 streamdetails.loudness = None
103 queue_item = MagicMock()
104 queue_item.media_type = MediaType.TRACK
105 queue_item.streamdetails = streamdetails
106 queue_item.name = "Track B"
107 return queue_item
108
109
110async def _drain(gen: AsyncGenerator[bytes]) -> None:
111 async for _ in gen:
112 pass
113
114
115@pytest.mark.asyncio
116async def test_dynamic_override_forces_loudnorm_and_ignores_measurement(
117 audio: tuple[StreamsAudio, dict[str, list[str]]],
118) -> None:
119 """A DYNAMIC override keeps the body on loudnorm even though a measurement is available."""
120 controller, captured = audio
121 queue_item = _make_queue_item()
122
123 await _drain(
124 controller.get_queue_item_stream(
125 queue_item,
126 PCM_FORMAT,
127 normalization_override=VolumeNormalizationMode.DYNAMIC,
128 )
129 )
130
131 # DYNAMIC -> loudnorm filter, NOT a static measurement gain
132 assert any(p.startswith("loudnorm") for p in captured["filter_params"])
133 assert queue_item.streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC
134 # the just-in-time hydration / re-evaluation is skipped entirely
135 mass = cast("MagicMock", controller.mass)
136 mass.streams.audio_analysis.get_audio_analysis.assert_not_called()
137 mass.streams.config.get_value.assert_not_called()
138
139
140@pytest.mark.asyncio
141async def test_no_override_reevaluates_to_measurement_only(
142 audio: tuple[StreamsAudio, dict[str, list[str]]],
143) -> None:
144 """Without an override, a landed measurement re-evaluates the body to MEASUREMENT_ONLY."""
145 controller, captured = audio
146 queue_item = _make_queue_item()
147
148 await _drain(controller.get_queue_item_stream(queue_item, PCM_FORMAT))
149
150 # MEASUREMENT_ONLY -> static volume gain of target - loudness = -17 - (-11.4) = -5.6 dB
151 assert "volume=-5.6dB" in captured["filter_params"]
152 assert (
153 queue_item.streamdetails.volume_normalization_mode
154 == VolumeNormalizationMode.MEASUREMENT_ONLY
155 )
156 # the measurement was hydrated from analysis
157 cast(
158 "MagicMock", controller.mass
159 ).streams.audio_analysis.get_audio_analysis.assert_called_once()
160