/
/
/
1"""Unit tests for the CLAP chunk dispatch state machine."""
2
3from __future__ import annotations
4
5from unittest.mock import MagicMock
6
7import numpy as np
8
9from music_assistant.providers.sonic_analysis import (
10 CLAP_WINDOW_SECONDS,
11 SonicSessionData,
12 _dispatch_clap_chunk,
13)
14
15SR = 22050
16WINDOW_SAMPLES = CLAP_WINDOW_SECONDS * SR
17
18
19def _make_session(target_starts: list[int]) -> SonicSessionData:
20 """Build a SonicSessionData with the given target starts and matching buffer lists."""
21 return SonicSessionData(
22 streamdetails=MagicMock(),
23 audio_format=MagicMock(),
24 clap_target_starts=list(target_starts),
25 clap_target_buffers=[[] for _ in target_starts],
26 clap_target_complete=[False] * len(target_starts),
27 )
28
29
30def _ramp(n: int) -> np.ndarray:
31 """Build a monotonically increasing float32 array so slice positions are visible."""
32 return np.arange(n, dtype=np.float32)
33
34
35def test_returns_empty_when_no_targets() -> None:
36 """Sessions with no planned target windows skip dispatch entirely."""
37 session = _make_session([])
38 completed = _dispatch_clap_chunk(session, _ramp(SR), SR)
39 assert completed == []
40
41
42def test_chunk_before_first_target_buffers_nothing() -> None:
43 """Chunks that arrive before the first target start don't touch any buffer."""
44 session = _make_session([10 * SR])
45 completed = _dispatch_clap_chunk(session, _ramp(5 * SR), SR)
46 assert completed == []
47 assert session.clap_target_buffers[0] == []
48 assert session.clap_position_samples == 5 * SR
49
50
51def test_chunk_inside_window_buffers_partial() -> None:
52 """A chunk landing mid-window appends but doesn't trigger completion."""
53 session = _make_session([0]) # target at sample 0, ends at WINDOW_SAMPLES
54 chunk = _ramp(3 * SR) # 3s â less than 7s window
55 completed = _dispatch_clap_chunk(session, chunk, SR)
56 assert completed == []
57 assert session.clap_target_complete[0] is False
58 assert sum(len(a) for a in session.clap_target_buffers[0]) == 3 * SR
59
60
61def test_window_completes_in_single_chunk() -> None:
62 """A chunk that fully covers a target window emits the completed window."""
63 session = _make_session([0])
64 chunk = _ramp(WINDOW_SAMPLES)
65 completed = _dispatch_clap_chunk(session, chunk, SR)
66 assert len(completed) == 1
67 assert len(completed[0]) == WINDOW_SAMPLES
68 assert session.clap_target_complete[0] is True
69 assert session.clap_target_buffers[0] == [] # freed on completion
70
71
72def test_window_completes_across_multiple_chunks() -> None:
73 """Three 3s chunks that span a 7s window complete it on the third."""
74 session = _make_session([0])
75 chunk_size = 3 * SR
76 completed_first = _dispatch_clap_chunk(session, _ramp(chunk_size), SR)
77 completed_second = _dispatch_clap_chunk(session, _ramp(chunk_size), SR)
78 completed_third = _dispatch_clap_chunk(session, _ramp(chunk_size), SR)
79 assert completed_first == []
80 assert completed_second == []
81 assert len(completed_third) == 1
82 assert len(completed_third[0]) == WINDOW_SAMPLES
83
84
85def test_chunk_overlapping_two_windows() -> None:
86 """A long chunk that spans the boundary between two adjacent windows fills both."""
87 target_a = 0
88 target_b = WINDOW_SAMPLES # back-to-back, no gap
89 session = _make_session([target_a, target_b])
90 chunk = _ramp(2 * WINDOW_SAMPLES)
91 completed = _dispatch_clap_chunk(session, chunk, SR)
92 assert len(completed) == 2
93 assert all(len(w) == WINDOW_SAMPLES for w in completed)
94 assert session.clap_target_complete == [True, True]
95
96
97def test_chunk_after_all_windows_complete_is_no_op_for_buffers() -> None:
98 """Once every target is complete, further chunks update position but don't buffer."""
99 session = _make_session([0])
100 _dispatch_clap_chunk(session, _ramp(WINDOW_SAMPLES), SR)
101 pre_buffers = [list(b) for b in session.clap_target_buffers]
102
103 completed = _dispatch_clap_chunk(session, _ramp(SR), SR)
104 assert completed == []
105 assert [list(b) for b in session.clap_target_buffers] == pre_buffers
106 assert session.clap_position_samples == WINDOW_SAMPLES + SR
107
108
109def test_completed_window_is_exactly_window_samples_long() -> None:
110 """If the buffer briefly accumulates more than 7s, the emitted window is trimmed."""
111 session = _make_session([0])
112 # one chunk slightly larger than the window
113 chunk = _ramp(WINDOW_SAMPLES + 1024)
114 completed = _dispatch_clap_chunk(session, chunk, SR)
115 assert len(completed) == 1
116 assert len(completed[0]) == WINDOW_SAMPLES
117
118
119def test_position_samples_advances_monotonically() -> None:
120 """clap_position_samples grows by exactly len(decoded_audio) per call."""
121 session = _make_session([0])
122 _dispatch_clap_chunk(session, _ramp(SR), SR)
123 _dispatch_clap_chunk(session, _ramp(2 * SR), SR)
124 _dispatch_clap_chunk(session, _ramp(3 * SR), SR)
125 assert session.clap_position_samples == 6 * SR
126
127
128def test_target_buffer_freed_on_completion() -> None:
129 """clap_target_buffers[i] is reset to [] the moment window i completes."""
130 session = _make_session([0])
131 # Fill mostly, then finish
132 _dispatch_clap_chunk(session, _ramp(WINDOW_SAMPLES - SR), SR)
133 assert session.clap_target_buffers[0] != []
134 _dispatch_clap_chunk(session, _ramp(SR), SR)
135 assert session.clap_target_buffers[0] == []
136
137
138def test_independent_targets_complete_independently() -> None:
139 """Two non-adjacent targets each complete on their own boundary chunks."""
140 session = _make_session([0, 10 * SR])
141 # Chunk 1: fills target 0 fully, lands inside [0, 7s]
142 completed_1 = _dispatch_clap_chunk(session, _ramp(WINDOW_SAMPLES), SR)
143 assert len(completed_1) == 1
144 assert session.clap_target_complete == [True, False]
145
146 # Chunk 2: lands at 7s..10s â between targets, fills nothing
147 completed_2 = _dispatch_clap_chunk(session, _ramp(3 * SR), SR)
148 assert completed_2 == []
149 assert session.clap_target_complete == [True, False]
150
151 # Chunk 3: lands at 10s..17s â fills target 1 exactly
152 completed_3 = _dispatch_clap_chunk(session, _ramp(WINDOW_SAMPLES), SR)
153 assert len(completed_3) == 1
154 assert session.clap_target_complete == [True, True]
155