music-assistant-server

1"""Tests for the MilkDrop visualizer audio tap."""
2
3from __future__ import annotations
4
5import struct
6from unittest.mock import Mock
7
8from aiosendspin.server.roles import AudioChunk
9
10from music_assistant.providers.milkdrop_visualizer.tap import (
11    WAVE_SAMPLES,
12    MilkdropWaveRole,
13    ViewerQueue,
14)
15
16
17def _stereo_chunk(mono_values: list[int], *, dangling_sample: bool = False) -> AudioChunk:
18    """Build an AudioChunk from mono int16 values, duplicated across L/R."""
19    samples: list[int] = []
20    for value in mono_values:
21        samples.extend((value, value))
22    if dangling_sample:
23        # One unpaired int16 (even byte count, odd sample count): the reshape
24        # would fail without the role's drop-the-dangling-sample guard.
25        samples.append(0)
26    data = struct.pack(f"<{len(samples)}h", *samples)
27    return AudioChunk(data=data, timestamp_us=1_000, duration_us=21_000, byte_count=len(data))
28
29
30def _role() -> tuple[MilkdropWaveRole, list[tuple[int, bytes]]]:
31    """Return a wave role wired to a capturing callback."""
32    role = MilkdropWaveRole(client=Mock(client_id="milkdrop-test"))
33    captured: list[tuple[int, bytes]] = []
34    role.set_wave_callback(lambda ts, samples: captured.append((ts, samples)))
35    return role, captured
36
37
38def test_emits_one_tail_per_full_buffer() -> None:
39    """A chunk with at least WAVE_SAMPLES mono samples emits a single 1024-byte tail."""
40    role, captured = _role()
41    role.on_audio_chunk(_stereo_chunk([0] * WAVE_SAMPLES))
42    assert len(captured) == 1
43    ts_us, samples = captured[0]
44    assert len(samples) == WAVE_SAMPLES
45    # timestamp is the chunk end (start + duration)
46    assert ts_us == 1_000 + 21_000
47
48
49def test_no_emit_before_warmup() -> None:
50    """A short chunk buffers silently until enough samples have accumulated."""
51    role, captured = _role()
52    role.on_audio_chunk(_stereo_chunk([0] * (WAVE_SAMPLES // 2)))
53    assert captured == []
54    role.on_audio_chunk(_stereo_chunk([0] * (WAVE_SAMPLES // 2)))
55    assert len(captured) == 1
56
57
58def test_quantization_is_symmetric_offset_binary() -> None:
59    """Full-scale +/- and zero map to 255 / 1 / 128, never wrapping."""
60    role, captured = _role()
61    role.on_audio_chunk(_stereo_chunk([32767] * WAVE_SAMPLES))
62    role.on_audio_chunk(_stereo_chunk([-32768] * WAVE_SAMPLES))
63    role.on_audio_chunk(_stereo_chunk([0] * WAVE_SAMPLES))
64    assert set(captured[0][1]) == {255}
65    assert set(captured[1][1]) == {1}
66    assert set(captured[2][1]) == {128}
67
68
69def test_odd_length_chunk_does_not_raise() -> None:
70    """A truncated (odd int16 count) chunk drops the dangling sample instead of failing."""
71    role, captured = _role()
72    role.on_audio_chunk(_stereo_chunk([0] * WAVE_SAMPLES, dangling_sample=True))
73    # Still produced a tail from the surviving even-length buffer.
74    assert len(captured) == 1
75    assert len(captured[0][1]) == WAVE_SAMPLES
76
77
78def test_stream_start_resets_the_rolling_buffer() -> None:
79    """A stream boundary clears buffered samples so a partial tail never carries over."""
80    role, captured = _role()
81    role.on_audio_chunk(_stereo_chunk([0] * (WAVE_SAMPLES - 1)))
82    role.on_stream_start()
83    role.on_audio_chunk(_stereo_chunk([0] * 1))
84    # Without the reset the two chunks would have summed past the threshold.
85    assert captured == []
86
87
88def test_viewer_queue_evicts_oldest_waveform_not_control() -> None:
89    """When full, the queue drops the oldest binary frame and keeps control messages."""
90    queue = ViewerQueue(capacity=3)
91    queue.push(b"wave-1")
92    queue.push('{"type": "stream/clear"}')
93    queue.push(b"wave-2")
94    # Full now; the next push must evict the oldest *binary* frame (wave-1).
95    queue.push(b"wave-3")
96    drained = [queue._items[i] for i in range(len(queue._items))]
97    assert b"wave-1" not in drained
98    assert '{"type": "stream/clear"}' in drained
99    assert b"wave-2" in drained
100    assert b"wave-3" in drained
101
102
103def test_viewer_queue_evicts_control_only_when_no_binary_left() -> None:
104    """With nothing but control messages queued, eviction falls back to the oldest item."""
105    queue = ViewerQueue(capacity=2)
106    queue.push('{"type": "stream/clear"}')
107    queue.push('{"type": "stream/end"}')
108    queue.push('{"type": "stream/start"}')
109    drained = [queue._items[i] for i in range(len(queue._items))]
110    assert len(drained) == 2
111    assert '{"type": "stream/start"}' in drained
112