music-assistant-server

1"""Tests for the MilkDrop visualizer audio tap."""
2
3from __future__ import annotations
4
5import struct
6from unittest.mock import MagicMock, Mock
7
8from aiosendspin.models.color import SessionUpdateColor
9from aiosendspin.models.types import UndefinedField
10from aiosendspin.server.roles import AudioChunk
11from orjson import loads
12
13from music_assistant.providers.milkdrop_visualizer.tap import (
14    _COLOR_FIELDS,
15    CONF_COLOR_TINT,
16    MILKDROP_ROLE_ID,
17    WAVE_SAMPLES,
18    MilkdropWaveRole,
19    Tap,
20    TapManager,
21    ViewerQueue,
22    _extract_color_update,
23)
24from music_assistant.providers.sendspin.bridge_role import COLOR_BRIDGE_ROLE_ID
25
26
27def _stereo_chunk(mono_values: list[int], *, dangling_sample: bool = False) -> AudioChunk:
28    """Build an AudioChunk from mono int16 values, duplicated across L/R."""
29    samples: list[int] = []
30    for value in mono_values:
31        samples.extend((value, value))
32    if dangling_sample:
33        # One unpaired int16 (even byte count, odd sample count): the reshape
34        # would fail without the role's drop-the-dangling-sample guard.
35        samples.append(0)
36    data = struct.pack(f"<{len(samples)}h", *samples)
37    return AudioChunk(data=data, timestamp_us=1_000, duration_us=21_000, byte_count=len(data))
38
39
40def _role() -> tuple[MilkdropWaveRole, list[tuple[int, bytes]]]:
41    """Return a wave role wired to a capturing callback."""
42    role = MilkdropWaveRole(client=Mock(client_id="milkdrop-test"))
43    captured: list[tuple[int, bytes]] = []
44    role.set_wave_callback(lambda ts, samples: captured.append((ts, samples)))
45    return role, captured
46
47
48def test_emits_one_tail_per_full_buffer() -> None:
49    """A chunk with at least WAVE_SAMPLES mono samples emits a single 1024-byte tail."""
50    role, captured = _role()
51    role.on_audio_chunk(_stereo_chunk([0] * WAVE_SAMPLES))
52    assert len(captured) == 1
53    ts_us, samples = captured[0]
54    assert len(samples) == WAVE_SAMPLES
55    # timestamp is the chunk end (start + duration)
56    assert ts_us == 1_000 + 21_000
57
58
59def test_no_emit_before_warmup() -> None:
60    """A short chunk buffers silently until enough samples have accumulated."""
61    role, captured = _role()
62    role.on_audio_chunk(_stereo_chunk([0] * (WAVE_SAMPLES // 2)))
63    assert captured == []
64    role.on_audio_chunk(_stereo_chunk([0] * (WAVE_SAMPLES // 2)))
65    assert len(captured) == 1
66
67
68def test_quantization_is_symmetric_offset_binary() -> None:
69    """Full-scale +/- and zero map to 255 / 1 / 128, never wrapping."""
70    role, captured = _role()
71    role.on_audio_chunk(_stereo_chunk([32767] * WAVE_SAMPLES))
72    role.on_audio_chunk(_stereo_chunk([-32768] * WAVE_SAMPLES))
73    role.on_audio_chunk(_stereo_chunk([0] * WAVE_SAMPLES))
74    assert set(captured[0][1]) == {255}
75    assert set(captured[1][1]) == {1}
76    assert set(captured[2][1]) == {128}
77
78
79def test_odd_length_chunk_does_not_raise() -> None:
80    """A truncated (odd int16 count) chunk drops the dangling sample instead of failing."""
81    role, captured = _role()
82    role.on_audio_chunk(_stereo_chunk([0] * WAVE_SAMPLES, dangling_sample=True))
83    # Still produced a tail from the surviving even-length buffer.
84    assert len(captured) == 1
85    assert len(captured[0][1]) == WAVE_SAMPLES
86
87
88def test_stream_start_resets_the_rolling_buffer() -> None:
89    """A stream boundary clears buffered samples so a partial tail never carries over."""
90    role, captured = _role()
91    role.on_audio_chunk(_stereo_chunk([0] * (WAVE_SAMPLES - 1)))
92    role.on_stream_start()
93    role.on_audio_chunk(_stereo_chunk([0] * 1))
94    # Without the reset the two chunks would have summed past the threshold.
95    assert captured == []
96
97
98def test_viewer_queue_evicts_oldest_waveform_not_control() -> None:
99    """When full, the queue drops the oldest binary frame and keeps control messages."""
100    queue = ViewerQueue(capacity=3)
101    queue.push(b"wave-1")
102    queue.push('{"type": "stream/clear"}')
103    queue.push(b"wave-2")
104    # Full now; the next push must evict the oldest *binary* frame (wave-1).
105    queue.push(b"wave-3")
106    drained = [queue._items[i] for i in range(len(queue._items))]
107    assert b"wave-1" not in drained
108    assert '{"type": "stream/clear"}' in drained
109    assert b"wave-2" in drained
110    assert b"wave-3" in drained
111
112
113def test_viewer_queue_evicts_control_only_when_no_binary_left() -> None:
114    """With nothing but control messages queued, eviction falls back to the oldest item."""
115    queue = ViewerQueue(capacity=2)
116    queue.push('{"type": "stream/clear"}')
117    queue.push('{"type": "stream/end"}')
118    queue.push('{"type": "stream/start"}')
119    drained = [queue._items[i] for i in range(len(queue._items))]
120    assert len(drained) == 2
121    assert '{"type": "stream/start"}' in drained
122
123
124def _color_payload(**fields: tuple[int, int, int] | None) -> Mock:
125    """Build a ServerStatePayload-shaped mock whose color carries only the given fields."""
126    color = Mock(spec=SessionUpdateColor)
127    for name in _COLOR_FIELDS:
128        setattr(color, name, fields.get(name, UndefinedField()))
129    return Mock(color=color)
130
131
132def test_extract_color_update_skips_undefined_fields() -> None:
133    """Only fields the server actually touched make it into the update."""
134    payload = _color_payload(primary=(10, 20, 30))
135    assert _extract_color_update(payload) == {"primary": (10, 20, 30)}
136
137
138def test_extract_color_update_keeps_explicit_nulls() -> None:
139    """A field explicitly cleared to None is forwarded, not dropped like an undefined one."""
140    payload = _color_payload(primary=None)
141    assert _extract_color_update(payload) == {"primary": None}
142
143
144def test_extract_color_update_returns_empty_when_payload_has_no_color() -> None:
145    """A server/state message that doesn't touch color yields nothing to forward."""
146    assert _extract_color_update(Mock(color=None)) == {}
147
148
149def test_tap_apply_color_merges_into_cached_palette() -> None:
150    """Successive partial updates accumulate rather than overwrite the whole palette."""
151    tap = Tap("milkdrop-test")
152    tap.apply_color({"primary": (10, 20, 30)})
153    tap.apply_color({"accent": (1, 2, 3)})
154    assert tap.last_color == {"primary": (10, 20, 30), "accent": (1, 2, 3)}
155
156
157def test_tap_apply_color_fans_out_only_the_update() -> None:
158    """Viewers receive just what changed, not the whole accumulated palette."""
159    tap = Tap("milkdrop-test")
160    queue = ViewerQueue()
161    tap.queues.add(queue)
162    tap.apply_color({"primary": (10, 20, 30)})
163    tap.apply_color({"accent": (1, 2, 3)})
164    first = loads(queue._items[0])
165    second = loads(queue._items[1])
166    assert first == {"type": "color", "payload": {"primary": [10, 20, 30]}}
167    assert second == {"type": "color", "payload": {"accent": [1, 2, 3]}}
168
169
170def test_tap_apply_color_ignores_an_empty_update() -> None:
171    """An update with nothing new (e.g. an undefined-only payload) fans out nothing."""
172    tap = Tap("milkdrop-test")
173    queue = ViewerQueue()
174    tap.queues.add(queue)
175    tap.apply_color({})
176    assert not queue._items
177
178
179def _tap_manager(*, color_tint: bool) -> tuple[TapManager, MagicMock, MagicMock, MagicMock]:
180    """Build a TapManager against mocks, returning it with the sendspin api and roles."""
181    provider = MagicMock()
182    provider.config.get_value.side_effect = lambda key: (
183        color_tint if key == CONF_COLOR_TINT else None
184    )
185    sendspin = MagicMock()
186    provider.mass.get_provider.return_value = sendspin
187    wave_role = MagicMock()
188    color_role = MagicMock()
189    viz_client = MagicMock()
190    viz_client.roles_by_family.side_effect = lambda family: {
191        "visualizer": [wave_role],
192        "color": [color_role],
193    }[family]
194    sendspin.server_api.register_external_player.return_value = viz_client
195    return TapManager(provider), sendspin, wave_role, color_role
196
197
198def test_register_client_includes_color_role_when_tint_enabled() -> None:
199    """With color tint on, the tap registers the color role and wires its callback."""
200    manager, sendspin, _wave_role, color_role = _tap_manager(color_tint=True)
201    manager._register_client(Tap("milkdrop-test"))
202    hello = sendspin.server_api.register_external_player.call_args.args[0]
203    assert hello.supported_roles == [MILKDROP_ROLE_ID, COLOR_BRIDGE_ROLE_ID]
204    color_role.set_callbacks.assert_called_once()
205
206
207def test_register_client_omits_color_role_when_tint_disabled() -> None:
208    """With color tint off, no color role is registered and nothing is wired to it."""
209    manager, sendspin, _wave_role, color_role = _tap_manager(color_tint=False)
210    manager._register_client(Tap("milkdrop-test"))
211    hello = sendspin.server_api.register_external_player.call_args.args[0]
212    assert hello.supported_roles == [MILKDROP_ROLE_ID]
213    color_role.set_callbacks.assert_not_called()
214