/
/
/
1"""Unit tests for the finalize-time flush of partially filled CLAP windows."""
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 SonicAnalysisProvider,
12 SonicSessionData,
13)
14
15SR = 22050
16WINDOW_SAMPLES = CLAP_WINDOW_SECONDS * SR
17
18
19def _make_provider() -> tuple[SonicAnalysisProvider, list[np.ndarray]]:
20 """
21 Stub provider whose create_task records the window each dispatch was handed.
22
23 :returns: ``(provider, dispatched_windows)``.
24 """
25 dispatched: list[np.ndarray] = []
26 p = SonicAnalysisProvider.__new__(SonicAnalysisProvider)
27 p.logger = MagicMock()
28 p.mass = MagicMock()
29 p.mass.create_task = MagicMock(side_effect=lambda _coro: MagicMock())
30
31 def _record(_session: SonicSessionData, window_audio: np.ndarray, _source_sr: int) -> MagicMock:
32 """Capture the window instead of running inference on it."""
33 dispatched.append(window_audio)
34 return MagicMock()
35
36 p._run_single_clap_window = _record # type: ignore[method-assign,assignment]
37 return p, dispatched
38
39
40def _make_session(target_starts: list[int]) -> SonicSessionData:
41 """Build a SonicSessionData with the given target starts and matching buffer lists."""
42 return SonicSessionData(
43 streamdetails=MagicMock(),
44 audio_format=MagicMock(),
45 clap_target_starts=list(target_starts),
46 clap_target_buffers=[[] for _ in target_starts],
47 clap_target_complete=[False] * len(target_starts),
48 )
49
50
51def test_partial_window_is_flushed() -> None:
52 """A window holding under 7s of audio is dispatched and marked complete."""
53 p, dispatched = _make_provider()
54 session = _make_session([0])
55 partial = np.ones(SR * 3, dtype=np.float32)
56 session.clap_target_buffers[0] = [partial]
57
58 p._flush_incomplete_clap_windows(session, SR)
59
60 assert len(dispatched) == 1
61 assert len(dispatched[0]) == SR * 3
62 assert session.clap_target_complete == [True]
63 assert session.clap_target_buffers[0] == []
64 assert len(session.clap_inference_tasks) == 1
65
66
67def test_multiple_buffered_chunks_are_concatenated() -> None:
68 """Chunks accumulated across dispatches are joined in order for the flush."""
69 p, dispatched = _make_provider()
70 session = _make_session([0])
71 session.clap_target_buffers[0] = [
72 np.arange(0, SR, dtype=np.float32),
73 np.arange(SR, 2 * SR, dtype=np.float32),
74 ]
75
76 p._flush_incomplete_clap_windows(session, SR)
77
78 np.testing.assert_array_equal(dispatched[0], np.arange(0, 2 * SR, dtype=np.float32))
79
80
81def test_already_complete_window_is_not_redispatched() -> None:
82 """Windows that already reached the 7s gate are left alone."""
83 p, dispatched = _make_provider()
84 session = _make_session([0, WINDOW_SAMPLES])
85 session.clap_target_complete = [True, False]
86 session.clap_target_buffers = [[], [np.ones(SR, dtype=np.float32)]]
87
88 p._flush_incomplete_clap_windows(session, SR)
89
90 assert len(dispatched) == 1
91 assert len(dispatched[0]) == SR
92
93
94def test_window_with_no_audio_stays_incomplete() -> None:
95 """A planned window the stream never reached is not faked as complete."""
96 p, dispatched = _make_provider()
97 session = _make_session([0, WINDOW_SAMPLES])
98 session.clap_target_buffers = [[np.ones(SR, dtype=np.float32)], []]
99
100 p._flush_incomplete_clap_windows(session, SR)
101
102 assert len(dispatched) == 1
103 assert session.clap_target_complete == [True, False]
104
105
106def test_no_targets_is_a_no_op() -> None:
107 """A session with no planned windows dispatches nothing."""
108 p, dispatched = _make_provider()
109 session = _make_session([])
110
111 p._flush_incomplete_clap_windows(session, SR)
112
113 assert dispatched == []
114 assert session.clap_inference_tasks == []
115