/
/
/
1"""Tests for the synchronous CLAP model load in SonicAnalysisProvider."""
2
3from __future__ import annotations
4
5from collections.abc import Generator
6from typing import Any
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from music_assistant_models.enums import ContentType
11from music_assistant_models.errors import SetupFailedError
12from music_assistant_models.media_items import AudioFormat
13
14from music_assistant.providers.sonic_analysis import (
15 CLAP_SAMPLING_FAST,
16 SonicAnalysisProvider,
17)
18
19# ---------------------------------------------------------------------------
20# Helpers
21# ---------------------------------------------------------------------------
22
23
24def _make_provider() -> SonicAnalysisProvider:
25 """
26 Construct a SonicAnalysisProvider with mocked MA infrastructure.
27
28 Uses ``__new__`` to bypass ``__init__`` (no model downloads) and manually
29 sets the attributes the load + start-analysis paths touch.
30 """
31 mass = MagicMock()
32
33 manifest = MagicMock()
34 manifest.domain = "sonic_analysis"
35
36 p = SonicAnalysisProvider.__new__(SonicAnalysisProvider)
37 p.logger = MagicMock()
38 p.mass = mass
39 p.manifest = manifest
40 p.config = MagicMock()
41 p.config.get_value = MagicMock(return_value=CLAP_SAMPLING_FAST)
42 p._sessions = {}
43 p._clap_model = None
44 p._clap_text_embeddings = None
45 p._clap_prompt_order = []
46 p.analysis_version = 1
47 return p
48
49
50def _make_audio_format(
51 sample_rate: int = 22050,
52 bit_depth: int = 16,
53 channels: int = 1,
54) -> AudioFormat:
55 """Return a real AudioFormat for 16-bit mono PCM."""
56 return AudioFormat(
57 content_type=ContentType.PCM_S16LE,
58 sample_rate=sample_rate,
59 bit_depth=bit_depth,
60 channels=channels,
61 )
62
63
64def _make_streamdetails(
65 item_id: str = "track-1",
66 duration: float | None = 60.0,
67) -> MagicMock:
68 """Return a minimal streamdetails mock."""
69 sd = MagicMock()
70 sd.item_id = item_id
71 sd.provider = "test_provider"
72 sd.duration = duration
73 return sd
74
75
76@pytest.fixture(autouse=True)
77def _stub_ml_inference_gate() -> Generator[None]:
78 """Stub the hardware gate so these unit tests never spawn the real capability probe."""
79 with patch(
80 "music_assistant.providers.sonic_analysis.verify_system_meets_requirements",
81 new=AsyncMock(),
82 ):
83 yield
84
85
86# ---------------------------------------------------------------------------
87# handle_async_init populates state on success
88# ---------------------------------------------------------------------------
89
90
91@pytest.mark.asyncio
92async def test_handle_async_init_populates_state_on_success() -> None:
93 """On success ``handle_async_init`` must populate model/embeddings/prompt order."""
94 provider = _make_provider()
95 fake_model = MagicMock(name="clap_model")
96 fake_embeddings = MagicMock(name="text_embeddings")
97 fake_prompt_order: list[tuple[str, tuple[str, str]]] = [
98 ("danceable", ("danceable", "not danceable")),
99 ("energetic", ("energetic", "calm")),
100 ]
101
102 with patch.object(
103 provider,
104 "_load_clap",
105 return_value=(fake_model, fake_embeddings, fake_prompt_order),
106 ):
107 await provider.handle_async_init()
108
109 assert provider._clap_model is fake_model
110 assert provider._clap_text_embeddings is fake_embeddings
111 assert provider._clap_prompt_order == fake_prompt_order
112
113 provider.logger.info.assert_called_once() # type: ignore[attr-defined]
114 info_call = provider.logger.info.call_args # type: ignore[attr-defined]
115 assert info_call.args[1] == len(fake_prompt_order)
116
117
118# ---------------------------------------------------------------------------
119# handle_async_init propagates load failure so provider.available stays False
120# ---------------------------------------------------------------------------
121
122
123@pytest.mark.asyncio
124async def test_handle_async_init_propagates_load_failure() -> None:
125 """
126 Synchronous load: failures must propagate, not be swallowed.
127
128 The AudioAnalysisController gates work on ``provider.available``, which
129 stays ``False`` if ``handle_async_init`` raises. Swallowing here would
130 flip the provider to ``available=True`` despite ``_clap_model is None``.
131 """
132 provider = _make_provider()
133 err = RuntimeError("hf network unreachable")
134
135 with (
136 patch.object(provider, "_load_clap", side_effect=err),
137 pytest.raises(RuntimeError, match="hf network unreachable"),
138 ):
139 await provider.handle_async_init()
140
141 assert provider._clap_model is None
142
143
144# ---------------------------------------------------------------------------
145# handle_async_init offloads the blocking load to a worker thread
146# ---------------------------------------------------------------------------
147
148
149@pytest.mark.asyncio
150async def test_handle_async_init_offloads_load_to_thread() -> None:
151 """``handle_async_init`` must offload ``_load_clap`` via asyncio.to_thread."""
152 provider = _make_provider()
153 fake_state: tuple[Any, Any, list[Any]] = (MagicMock(), MagicMock(), [])
154
155 with patch(
156 "music_assistant.providers.sonic_analysis.asyncio.to_thread",
157 new=AsyncMock(return_value=fake_state),
158 ) as to_thread_mock:
159 await provider.handle_async_init()
160
161 to_thread_mock.assert_called_once()
162 # First positional arg passed to to_thread is the callable being offloaded.
163 # Use ``==`` (not ``is``): each ``provider._load_clap`` access yields a fresh
164 # bound-method object, but bound methods of the same (instance, function)
165 # compare equal.
166 assert to_thread_mock.call_args.args[0] == provider._load_clap
167
168
169# ---------------------------------------------------------------------------
170# _start_analysis gating: declines tracks while CLAP is unavailable
171# ---------------------------------------------------------------------------
172
173
174@pytest.mark.asyncio
175async def test_start_analysis_returns_false_when_clap_not_loaded() -> None:
176 """
177 ``_start_analysis`` must decline tracks while CLAP is unavailable.
178
179 Defensive: with synchronous loading + raise-on-failure, the normal path
180 keeps ``_clap_model`` populated whenever the provider is available. This
181 guards the edge case of being invoked after ``unload`` cleared state.
182 """
183 provider = _make_provider()
184 provider._clap_model = None
185
186 af = _make_audio_format()
187 sd = _make_streamdetails(item_id="skip-me")
188
189 result = await provider._start_analysis("session-skip", sd, af)
190
191 assert result is False
192 assert provider._sessions == {}
193
194 provider.logger.debug.assert_called() # type: ignore[attr-defined]
195 debug_msgs = [str(c) for c in provider.logger.debug.call_args_list] # type: ignore[attr-defined]
196 assert any("CLAP model not yet available" in c for c in debug_msgs)
197
198
199@pytest.mark.asyncio
200async def test_start_analysis_proceeds_when_clap_loaded() -> None:
201 """``_start_analysis`` must create a session when CLAP is available."""
202 provider = _make_provider()
203 provider._clap_model = MagicMock(name="clap_model")
204
205 af = _make_audio_format()
206 sd = _make_streamdetails(item_id="go-ahead")
207
208 result = await provider._start_analysis("session-ok", sd, af)
209
210 assert result is True
211 assert "session-ok" in provider._sessions
212
213
214@pytest.mark.asyncio
215@pytest.mark.parametrize("duration", [None, 0, 0.0])
216async def test_start_analysis_returns_false_without_duration(
217 duration: float | None,
218) -> None:
219 """
220 ``_start_analysis`` must decline tracks without a usable duration.
221
222 Without duration, CLAP windows can't be planned and the resulting record
223 would be librosa-only. Rejecting at start keeps the retry path open for a
224 later analysis attempt once duration is known.
225 """
226 provider = _make_provider()
227 provider._clap_model = MagicMock(name="clap_model")
228
229 af = _make_audio_format()
230 sd = _make_streamdetails(item_id="no-duration", duration=duration)
231
232 result = await provider._start_analysis("session-no-duration", sd, af)
233
234 assert result is False
235 assert provider._sessions == {}
236
237 debug_msgs = [str(c) for c in provider.logger.debug.call_args_list] # type: ignore[attr-defined]
238 assert any("duration missing or zero" in c for c in debug_msgs)
239
240
241async def test_handle_async_init_raises_when_requirements_not_met() -> None:
242 """Setup fails before any model load when the system does not meet requirements."""
243 provider = _make_provider()
244 with (
245 patch(
246 "music_assistant.providers.sonic_analysis.verify_system_meets_requirements",
247 side_effect=SetupFailedError("unsupported system"),
248 ),
249 patch.object(SonicAnalysisProvider, "_load_clap") as load_clap_mock,
250 pytest.raises(SetupFailedError),
251 ):
252 await provider.handle_async_init()
253 load_clap_mock.assert_not_called()
254