/
/
/
1"""Tests for DSP configuration and preset persistence."""
2
3from __future__ import annotations
4
5import base64
6import re
7import subprocess
8from pathlib import Path
9from types import SimpleNamespace
10from typing import TYPE_CHECKING, Any, cast
11from unittest.mock import AsyncMock, MagicMock
12
13import pytest
14from music_assistant_models.dsp import (
15 ConvolutionFilter,
16 DSPConfig,
17 DSPConfigPreset,
18 ToneControlFilter,
19)
20from music_assistant_models.enums import EventType
21from music_assistant_models.errors import InvalidDataError
22
23from music_assistant.controllers.config.dsp import (
24 MAX_IR_BYTES,
25 MAX_IR_SECONDS,
26 DSPConfigMixin,
27)
28
29if TYPE_CHECKING:
30 from music_assistant.mass import MusicAssistant
31
32
33class _DSPConfigStore(DSPConfigMixin):
34 """In-memory DSP configuration store for focused controller tests."""
35
36 def __init__(self) -> None:
37 """Initialize the store and controller dependencies."""
38 self._data: dict[str, Any] = {}
39 self.update_player_dsp_preset = MagicMock()
40 self.signal_event = MagicMock()
41 self.on_player_dsp_change = AsyncMock()
42 self.mass = cast(
43 "MusicAssistant",
44 SimpleNamespace(
45 players=SimpleNamespace(on_player_dsp_change=self.on_player_dsp_change),
46 streams=SimpleNamespace(
47 audio_processing=SimpleNamespace(
48 update_player_dsp_preset=self.update_player_dsp_preset
49 )
50 ),
51 signal_event=self.signal_event,
52 ),
53 )
54
55 def get(self, key: str, default: Any = None) -> Any:
56 """Return a value from the nested test store."""
57 value: Any = self._data
58 for subkey in key.split("/"):
59 if not isinstance(value, dict) or subkey not in value:
60 return default
61 value = value[subkey]
62 return value
63
64 def set(self, key: str, value: Any) -> None:
65 """Set a value in the nested test store."""
66 parent = self._data
67 subkeys = key.split("/")
68 for subkey in subkeys[:-1]:
69 parent = parent.setdefault(subkey, {})
70 parent[subkeys[-1]] = value
71
72 def remove(self, key: str) -> None:
73 """Remove a value from the nested test store."""
74 parent = self._data
75 subkeys = key.split("/")
76 for subkey in subkeys[:-1]:
77 if subkey not in parent:
78 return
79 parent = parent[subkey]
80 parent.pop(subkeys[-1], None)
81
82
83async def test_apply_preset_and_manual_save_reset_identity() -> None:
84 """Preset application persists identity and a manual save clears it."""
85 config = _DSPConfigStore()
86 preset = await config.save_dsp_presets(
87 DSPConfigPreset(
88 name="Warm",
89 preset_id="warm",
90 config=DSPConfig(
91 enabled=True,
92 filters=[ToneControlFilter(enabled=True, bass_level=2.0)],
93 preset_id="other",
94 ),
95 )
96 )
97
98 applied = await config.apply_dsp_preset("player-1", "warm")
99
100 assert preset.config.preset_id is None
101 assert applied.preset_id == "warm"
102 assert config.get_player_dsp_config("player-1") == applied
103 applied.input_gain = -1.5
104 saved = await config.save_dsp_config("player-1", applied)
105 assert saved.preset_id is None
106 assert config.get_player_dsp_config("player-1").preset_id is None
107
108
109async def test_apply_missing_preset_fails() -> None:
110 """Applying an unknown preset reports invalid input."""
111 config = _DSPConfigStore()
112
113 with pytest.raises(KeyError, match="missing"):
114 await config.apply_dsp_preset("player-1", "missing")
115
116
117async def test_preset_setting_update_clears_assignments() -> None:
118 """Changing preset settings clears selection without changing player DSP."""
119 config = _DSPConfigStore()
120 original = DSPConfig(enabled=False, input_gain=-2.0)
121 await config.save_dsp_presets(DSPConfigPreset(name="Quiet", preset_id="quiet", config=original))
122 await config.apply_dsp_preset("player-1", "quiet")
123
124 await config.save_dsp_presets(
125 DSPConfigPreset(
126 name="Quieter",
127 preset_id="quiet",
128 config=DSPConfig(enabled=False, input_gain=-4.0),
129 )
130 )
131
132 player_config = config.get_player_dsp_config("player-1")
133 assert player_config.input_gain == -2.0
134 assert player_config.preset_id is None
135 config.update_player_dsp_preset.assert_called_with("player-1", None)
136
137
138async def test_preset_rename_preserves_assignments() -> None:
139 """Renaming a preset keeps matching player selections."""
140 config = _DSPConfigStore()
141 preset_config = DSPConfig(enabled=False, output_gain=-1.0)
142 await config.save_dsp_presets(
143 DSPConfigPreset(name="Original", preset_id="named", config=preset_config)
144 )
145 await config.apply_dsp_preset("player-1", "named")
146
147 await config.save_dsp_presets(
148 DSPConfigPreset(name="Renamed", preset_id="named", config=preset_config)
149 )
150
151 assert config.get_player_dsp_config("player-1").preset_id == "named"
152
153
154async def test_remove_preset_clears_assignments() -> None:
155 """Removing a preset keeps copied values but clears its selection."""
156 config = _DSPConfigStore()
157 await config.save_dsp_presets(
158 DSPConfigPreset(
159 name="Night",
160 preset_id="night",
161 config=DSPConfig(enabled=False, output_gain=-3.0),
162 )
163 )
164 await config.apply_dsp_preset("player-1", "night")
165
166 await config.remove_dsp_preset("night")
167
168 player_config = config.get_player_dsp_config("player-1")
169 assert player_config.output_gain == -3.0
170 assert player_config.preset_id is None
171 assert await config.get_dsp_presets() == []
172
173
174def _wav_bytes(tmp_path: Path, channels: int = 2, duration: float = 0.1) -> bytes:
175 """Generate a short wav with the given channel count and return its raw bytes."""
176 wav_path = tmp_path / f"source_{channels}ch_{duration}s.wav"
177 subprocess.run( # noqa: S603
178 [ # noqa: S607
179 "ffmpeg",
180 "-y",
181 "-f",
182 "lavfi",
183 "-i",
184 f"sine=frequency=1000:duration={duration}:sample_rate=48000",
185 "-ac",
186 str(channels),
187 str(wav_path),
188 ],
189 check=True,
190 capture_output=True,
191 )
192 return wav_path.read_bytes()
193
194
195def _silent_flac_bytes(tmp_path: Path, duration: float) -> bytes:
196 """Generate a silent flac of the given length, which compresses to very little."""
197 flac_path = tmp_path / f"silence_{duration}s.flac"
198 subprocess.run( # noqa: S603
199 [ # noqa: S607
200 "ffmpeg",
201 "-y",
202 "-f",
203 "lavfi",
204 "-i",
205 "anullsrc=r=48000:cl=stereo",
206 "-t",
207 str(duration),
208 "-c:a",
209 "flac",
210 str(flac_path),
211 ],
212 check=True,
213 capture_output=True,
214 )
215 return flac_path.read_bytes()
216
217
218async def test_upload_list_and_remove_ir(tmp_path: Path) -> None:
219 """An uploaded impulse response is stored, listed and then removed."""
220 config = _DSPConfigStore()
221 config.mass.storage_path = str(tmp_path)
222
223 data = base64.b64encode(_wav_bytes(tmp_path)).decode()
224 record = await config.upload_dsp_ir("My Room", data)
225
226 ir_id = record["ir_id"]
227 assert record["name"] == "My Room"
228 assert record["sample_rate"] == 48000
229 assert record["channels"] == 2
230 assert (tmp_path / "dsp_irs" / f"{ir_id}.wav").is_file()
231 assert config.get_dsp_irs() == [record]
232
233 await config.remove_dsp_ir(ir_id)
234 assert config.get_dsp_irs() == []
235 assert not (tmp_path / "dsp_irs" / f"{ir_id}.wav").exists()
236
237
238async def test_ir_changes_signal_event(tmp_path: Path) -> None:
239 """Uploading or removing an impulse response announces the resulting library."""
240 config = _DSPConfigStore()
241 config.mass.storage_path = str(tmp_path)
242
243 data = base64.b64encode(_wav_bytes(tmp_path)).decode()
244 record = await config.upload_dsp_ir("My Room", data)
245 config.signal_event.assert_called_once_with(EventType.DSP_IRS_UPDATED, data=[record])
246
247 config.signal_event.reset_mock()
248 await config.remove_dsp_ir(record["ir_id"])
249 config.signal_event.assert_called_once_with(EventType.DSP_IRS_UPDATED, data=[])
250
251
252async def test_remove_unused_ir_signals_event(tmp_path: Path) -> None:
253 """Removing an IR that no player or preset uses still announces the change."""
254 config = _DSPConfigStore()
255 config.mass.storage_path = str(tmp_path)
256 await config.save_dsp_config("player-1", DSPConfig(enabled=True, output_gain=-3.0))
257
258 data = base64.b64encode(_wav_bytes(tmp_path)).decode()
259 unused = await config.upload_dsp_ir("Unused", data)
260 config.signal_event.reset_mock()
261
262 await config.remove_dsp_ir(unused["ir_id"])
263
264 config.signal_event.assert_called_once_with(EventType.DSP_IRS_UPDATED, data=[])
265
266
267async def test_upload_ir_rejects_non_audio(tmp_path: Path) -> None:
268 """Uploading data that is not decodable audio raises and stores nothing."""
269 config = _DSPConfigStore()
270 config.mass.storage_path = str(tmp_path)
271
272 data = base64.b64encode(b"this is not an audio file").decode()
273 with pytest.raises(InvalidDataError):
274 await config.upload_dsp_ir("bad", data)
275
276 assert config.get_dsp_irs() == []
277 # the transcode failure must not leave a stored wav or upload temp file behind
278 assert list((tmp_path / "dsp_irs").glob("*")) == []
279
280
281async def test_upload_ir_rejects_multichannel_file(tmp_path: Path) -> None:
282 """A file with more than two channels is rejected rather than silently downmixed."""
283 config = _DSPConfigStore()
284 config.mass.storage_path = str(tmp_path)
285
286 data = base64.b64encode(_wav_bytes(tmp_path, channels=4)).decode()
287 with pytest.raises(InvalidDataError, match="only mono and stereo"):
288 await config.upload_dsp_ir("true stereo", data)
289
290 assert config.get_dsp_irs() == []
291 assert list((tmp_path / "dsp_irs").glob("*")) == []
292
293
294async def test_upload_ir_rejects_long_file(tmp_path: Path) -> None:
295 """An impulse response longer than the limit is rejected, as afir would fail on it."""
296 config = _DSPConfigStore()
297 config.mass.storage_path = str(tmp_path)
298
299 data = base64.b64encode(_wav_bytes(tmp_path, duration=MAX_IR_SECONDS + 1)).decode()
300 with pytest.raises(InvalidDataError, match="seconds"):
301 await config.upload_dsp_ir("too long", data)
302
303 assert config.get_dsp_irs() == []
304 assert list((tmp_path / "dsp_irs").glob("*")) == []
305
306
307async def test_upload_ir_transcode_is_length_bounded(tmp_path: Path) -> None:
308 """A tiny but very long upload is truncated on the way in, not written out in full."""
309 config = _DSPConfigStore()
310 config.mass.storage_path = str(tmp_path)
311
312 data = base64.b64encode(_silent_flac_bytes(tmp_path, duration=3600)).decode()
313 with pytest.raises(InvalidDataError) as excinfo:
314 await config.upload_dsp_ir("very long", data)
315
316 # an unbounded transcode would report the source length, after writing gigabytes
317 reported = re.search(r"is ([\d.]+) seconds", str(excinfo.value))
318 assert reported is not None
319 assert float(reported.group(1)) <= MAX_IR_SECONDS + 1
320
321 assert list((tmp_path / "dsp_irs").glob("*")) == []
322
323
324async def test_upload_ir_cleans_up_after_an_unexpected_failure(
325 tmp_path: Path, monkeypatch: pytest.MonkeyPatch
326) -> None:
327 """A failure the upload does not anticipate still leaves no orphaned file behind."""
328 config = _DSPConfigStore()
329 config.mass.storage_path = str(tmp_path)
330 monkeypatch.setattr(
331 "music_assistant.controllers.config.dsp.async_parse_tags",
332 AsyncMock(side_effect=RuntimeError("boom")),
333 )
334
335 data = base64.b64encode(_wav_bytes(tmp_path)).decode()
336 with pytest.raises(RuntimeError):
337 await config.upload_dsp_ir("broken", data)
338
339 assert config.get_dsp_irs() == []
340 assert list((tmp_path / "dsp_irs").glob("*")) == []
341
342
343async def test_remove_ir_with_nothing_to_remove_is_a_no_op(tmp_path: Path) -> None:
344 """An id with neither a record nor a file leaves the library untouched and silent."""
345 config = _DSPConfigStore()
346 config.mass.storage_path = str(tmp_path)
347
348 await config.remove_dsp_ir("abc123")
349
350 config.signal_event.assert_not_called()
351
352
353async def test_remove_ir_drops_record_with_unusable_id(tmp_path: Path) -> None:
354 """A stored id that no longer passes validation can still be removed from the library."""
355 config = _DSPConfigStore()
356 config.mass.storage_path = str(tmp_path)
357 config.set("player_dsp_irs", {"Legacy ID": {"ir_id": "Legacy ID", "name": "old"}})
358
359 await config.remove_dsp_ir("Legacy ID")
360
361 assert config.get_dsp_irs() == []
362
363
364@pytest.mark.parametrize("evil_id", ["../evil", "../../etc/cron.d/x", "/etc/passwd", "a/b"])
365async def test_remove_ir_rejects_path_traversal(tmp_path: Path, evil_id: str) -> None:
366 """A crafted ir_id cannot delete a file outside the impulse response directory."""
367 config = _DSPConfigStore()
368 config.mass.storage_path = str(tmp_path)
369 # a sentinel file the traversal ids above would resolve to if unguarded
370 sentinel = tmp_path / "evil.wav"
371 sentinel.write_text("keep me")
372
373 with pytest.raises(InvalidDataError):
374 await config.remove_dsp_ir(evil_id)
375
376 assert sentinel.exists()
377
378
379async def test_upload_ir_rejects_oversized_file(tmp_path: Path) -> None:
380 """An upload larger than the size limit is rejected before it is written."""
381 config = _DSPConfigStore()
382 config.mass.storage_path = str(tmp_path)
383
384 oversized = base64.b64encode(b"\x00" * (MAX_IR_BYTES + 1)).decode()
385 with pytest.raises(InvalidDataError):
386 await config.upload_dsp_ir("too big", oversized)
387
388 assert config.get_dsp_irs() == []
389
390
391async def test_save_dsp_config_rejects_unknown_ir() -> None:
392 """A player config naming an impulse response this server does not hold is refused."""
393 config = _DSPConfigStore()
394
395 with pytest.raises(InvalidDataError, match="Unknown impulse response"):
396 await config.save_dsp_config(
397 "player-1",
398 DSPConfig(enabled=True, filters=[ConvolutionFilter(enabled=True, ir_id="gone")]),
399 )
400
401 assert config.get_player_dsp_config("player-1").filters == []
402
403
404async def test_save_dsp_preset_rejects_unknown_ir() -> None:
405 """A preset naming an impulse response this server does not hold is refused."""
406 config = _DSPConfigStore()
407
408 with pytest.raises(InvalidDataError, match="Unknown impulse response"):
409 await config.save_dsp_presets(
410 DSPConfigPreset(
411 name="Room",
412 preset_id="room",
413 config=DSPConfig(
414 enabled=True, filters=[ConvolutionFilter(enabled=True, ir_id="gone")]
415 ),
416 )
417 )
418
419 assert await config.get_dsp_presets() == []
420
421
422async def test_save_dsp_config_allows_blank_ir() -> None:
423 """A convolution filter with nothing selected yet still saves, as removal leaves it blank."""
424 config = _DSPConfigStore()
425
426 saved = await config.save_dsp_config(
427 "player-1",
428 DSPConfig(enabled=True, filters=[ConvolutionFilter(enabled=True, ir_id="")]),
429 )
430
431 saved_filter = saved.filters[0]
432 assert isinstance(saved_filter, ConvolutionFilter)
433 assert saved_filter.ir_id == ""
434
435
436async def test_remove_ir_clears_references(tmp_path: Path) -> None:
437 """Removing an IR blanks its id from any player config or preset that used it."""
438 config = _DSPConfigStore()
439 config.mass.storage_path = str(tmp_path)
440 config.set("player_dsp_irs", {"abc123": {"ir_id": "abc123", "name": "Room"}})
441 await config.save_dsp_config(
442 "player-1",
443 DSPConfig(
444 enabled=True,
445 filters=[ConvolutionFilter(enabled=True, ir_id="abc123", gain=2.0)],
446 ),
447 )
448 await config.save_dsp_presets(
449 DSPConfigPreset(
450 name="Room",
451 preset_id="room",
452 config=DSPConfig(
453 enabled=True,
454 filters=[ConvolutionFilter(enabled=True, ir_id="abc123")],
455 ),
456 )
457 )
458
459 config.signal_event.reset_mock()
460
461 await config.remove_dsp_ir("abc123")
462
463 signalled = [call.args[0] for call in config.signal_event.call_args_list]
464 assert EventType.PLAYER_DSP_CONFIG_UPDATED in signalled
465 # the blanked preset must be announced too, or a client keeps offering the deleted IR
466 assert EventType.DSP_PRESETS_UPDATED in signalled
467 player_filter = config.get_player_dsp_config("player-1").filters[0]
468 assert isinstance(player_filter, ConvolutionFilter)
469 assert player_filter.ir_id == ""
470 assert player_filter.gain == 2.0
471 preset_filter = (await config.get_dsp_presets())[0].config.filters[0]
472 assert isinstance(preset_filter, ConvolutionFilter)
473 assert preset_filter.ir_id == ""
474
475
476async def test_remove_ir_rebuilds_the_stream_of_an_affected_player(tmp_path: Path) -> None:
477 """Blanking a convolution filter reapplies the DSP, as a saved config change would."""
478 config = _DSPConfigStore()
479 config.mass.storage_path = str(tmp_path)
480 config.set("player_dsp_irs", {"abc123": {"ir_id": "abc123", "name": "Room"}})
481 await config.save_dsp_config(
482 "player-1",
483 DSPConfig(enabled=True, filters=[ConvolutionFilter(enabled=True, ir_id="abc123")]),
484 )
485 await config.save_dsp_config("player-2", DSPConfig(enabled=True, output_gain=-3.0))
486 config.on_player_dsp_change.reset_mock()
487
488 await config.remove_dsp_ir("abc123")
489
490 config.on_player_dsp_change.assert_awaited_once_with("player-1")
491
492
493async def test_remove_ir_leaves_a_disabled_player_dsp_alone(tmp_path: Path) -> None:
494 """A player with DSP switched off hears nothing different, so it is not restarted."""
495 config = _DSPConfigStore()
496 config.mass.storage_path = str(tmp_path)
497 config.set("player_dsp_irs", {"abc123": {"ir_id": "abc123", "name": "Room"}})
498 await config.save_dsp_config(
499 "player-1",
500 DSPConfig(enabled=False, filters=[ConvolutionFilter(enabled=True, ir_id="abc123")]),
501 )
502 config.on_player_dsp_change.reset_mock()
503
504 await config.remove_dsp_ir("abc123")
505
506 config.on_player_dsp_change.assert_not_awaited()
507