/
/
/
1"""Unit tests for the lazy CLAP text encoder and _handle_text_search dispatcher hook."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from unittest.mock import AsyncMock, MagicMock
7
8import numpy as np
9import pytest
10from music_assistant_models.errors import MusicAssistantError
11
12from music_assistant.providers.sonic_similarity.similarity import ScoredCandidate
13from tests.providers.sonic_similarity.conftest import make_track
14
15if TYPE_CHECKING:
16 from collections.abc import Callable
17 from typing import Any
18
19
20def _make_tensor_chain(vector: np.ndarray) -> MagicMock:
21 """Build a tensor-like MagicMock whose .detach().cpu().numpy() yields vector."""
22 chain = MagicMock()
23 chain.detach.return_value = chain
24 chain.cpu.return_value = chain
25 chain.numpy.return_value = vector
26 return chain
27
28
29def _make_mock_encoder(*vectors: np.ndarray) -> MagicMock:
30 """
31 Build a CLAP-like encoder returning one tensor chain per prompt.
32
33 get_text_embeddings(prompts) yields a chain for each prompt, drawn in order
34 from the supplied vectors, so multi-prompt calls (query + exclude) work.
35 """
36 chains = [_make_tensor_chain(v) for v in vectors]
37 encoder = MagicMock()
38 encoder.get_text_embeddings.side_effect = lambda prompts: chains[: len(prompts)]
39 return encoder
40
41
42def _unit_vector(*components: float) -> np.ndarray:
43 """Return a 1024-dim float32 vector with the leading components set, rest zero."""
44 vec = np.zeros((1024,), dtype=np.float32)
45 vec[: len(components)] = components
46 return vec
47
48
49class TestGetTextEncoder:
50 """Tests for SonicSimilarityPlugin._get_text_encoder (lazy load + cache)."""
51
52 @pytest.mark.asyncio
53 async def test_returns_cached_encoder_on_second_call(
54 self, make_plugin: Callable[..., Any]
55 ) -> None:
56 """Second call returns the cached encoder without reloading."""
57 plugin = make_plugin()
58 loader = MagicMock(return_value="ENCODER_SENTINEL")
59 plugin._load_text_encoder = loader
60
61 first = await plugin._get_text_encoder()
62 second = await plugin._get_text_encoder()
63
64 assert first == "ENCODER_SENTINEL"
65 assert second == "ENCODER_SENTINEL"
66 assert loader.call_count == 1
67
68 @pytest.mark.asyncio
69 async def test_returns_none_when_load_raises(self, make_plugin: Callable[..., Any]) -> None:
70 """Encoder load failure is swallowed; method returns None and caches None."""
71 plugin = make_plugin()
72 plugin._load_text_encoder = MagicMock(side_effect=RuntimeError("boom"))
73
74 result = await plugin._get_text_encoder()
75
76 assert result is None
77 assert plugin._text_encoder is None
78
79
80class TestEmbedTextQuery:
81 """Tests for SonicSimilarityPlugin._embed_text_query (template + exclusion math)."""
82
83 @pytest.mark.asyncio
84 async def test_returns_unit_vector(self, make_plugin: Callable[..., Any]) -> None:
85 """A plain query yields the unit-normalised encoder embedding."""
86 plugin = make_plugin(clap_enabled=True)
87 plugin._load_text_encoder = lambda: _make_mock_encoder(_unit_vector(3.0, 4.0))
88
89 result = await plugin._embed_text_query("disco")
90
91 assert result is not None
92 np.testing.assert_allclose(result[:2], [0.6, 0.8], atol=1e-6)
93
94 @pytest.mark.asyncio
95 async def test_exclude_subtracts_normalised_direction(
96 self, make_plugin: Callable[..., Any]
97 ) -> None:
98 """Exclude subtracts the unit exclude direction: [1,0]-[0,1] -> [.707,-.707]."""
99 plugin = make_plugin(clap_enabled=True)
100 keep, excl = _unit_vector(1.0), _unit_vector(0.0, 1.0)
101 plugin._load_text_encoder = lambda: _make_mock_encoder(keep, excl)
102
103 result = await plugin._embed_text_query("loud", exclude="fast", exclude_weight=1.0)
104
105 assert result is not None
106 np.testing.assert_allclose(result[:2], [0.70710677, -0.70710677], atol=1e-6)
107
108 @pytest.mark.asyncio
109 async def test_exclude_weight_zero_returns_keep(self, make_plugin: Callable[..., Any]) -> None:
110 """weight=0 leaves the query embedding unchanged."""
111 plugin = make_plugin(clap_enabled=True)
112 keep, excl = _unit_vector(1.0), _unit_vector(0.0, 1.0)
113 plugin._load_text_encoder = lambda: _make_mock_encoder(keep, excl)
114
115 result = await plugin._embed_text_query("loud", exclude="fast", exclude_weight=0.0)
116
117 assert result is not None
118 np.testing.assert_allclose(result[:2], [1.0, 0.0], atol=1e-6)
119
120 @pytest.mark.asyncio
121 async def test_query_equals_exclude_returns_none(self, make_plugin: Callable[..., Any]) -> None:
122 """Identical keep/exclude cancel to a zero vector -> None (cosine-unsafe)."""
123 plugin = make_plugin(clap_enabled=True)
124 same = _unit_vector(1.0)
125 plugin._load_text_encoder = lambda: _make_mock_encoder(same, same)
126
127 result = await plugin._embed_text_query("loud", exclude="loud", exclude_weight=1.0)
128
129 assert result is None
130
131 @pytest.mark.asyncio
132 async def test_empty_query_returns_none_without_encoding(
133 self, make_plugin: Callable[..., Any]
134 ) -> None:
135 """A whitespace-only query short-circuits to None and never encodes."""
136 plugin = make_plugin(clap_enabled=True)
137 encoder = _make_mock_encoder(_unit_vector(1.0))
138 plugin._load_text_encoder = lambda: encoder
139
140 result = await plugin._embed_text_query(" ")
141
142 assert result is None
143 encoder.get_text_embeddings.assert_not_called()
144
145 @pytest.mark.asyncio
146 async def test_zero_norm_embedding_returns_none(self, make_plugin: Callable[..., Any]) -> None:
147 """A zero-norm embedding cannot be ranked by cosine -> None."""
148 plugin = make_plugin(clap_enabled=True)
149 plugin._load_text_encoder = lambda: _make_mock_encoder(np.zeros((1024,), dtype=np.float32))
150
151 result = await plugin._embed_text_query("disco")
152
153 assert result is None
154
155
156class TestHandleTextSearch:
157 """Tests for SonicSimilarityPlugin._handle_text_search."""
158
159 @pytest.mark.asyncio
160 async def test_empty_query_reports_empty_query(self, make_plugin: Callable[..., Any]) -> None:
161 """A whitespace-only query is rejected with the empty_query reason."""
162 plugin = make_plugin(clap_enabled=True)
163 plugin._clap_index.__len__ = MagicMock(return_value=5)
164
165 result = await plugin._handle_text_search(" ")
166
167 assert result["analyzed"] is False
168 assert result["reason"] == "empty_query"
169
170 @pytest.mark.asyncio
171 async def test_exclude_weight_clamped_to_range(self, make_plugin: Callable[..., Any]) -> None:
172 """exclude_weight is clamped to [0, 2] before reaching the embedder."""
173 plugin = make_plugin(clap_enabled=True)
174 plugin._clap_index.__len__ = MagicMock(return_value=5)
175 plugin._embed_text_query = AsyncMock(return_value=None)
176
177 await plugin._handle_text_search("disco", exclude="vocals", exclude_weight=9.0)
178
179 assert plugin._embed_text_query.await_args.kwargs["exclude_weight"] == 2.0
180
181 @pytest.mark.asyncio
182 async def test_returns_clap_index_empty_when_no_index(
183 self, make_plugin: Callable[..., Any]
184 ) -> None:
185 """Missing CLAP index short-circuits with clap_index_empty."""
186 plugin = make_plugin()
187
188 result = await plugin._handle_text_search("disco")
189
190 assert result["analyzed"] is False
191 assert result["reason"] == "clap_index_empty"
192 assert result["items"] == []
193
194 @pytest.mark.asyncio
195 async def test_returns_clap_index_empty_when_index_is_empty(
196 self, make_plugin: Callable[..., Any]
197 ) -> None:
198 """Empty CLAP index (len == 0) short-circuits with clap_index_empty."""
199 plugin = make_plugin(clap_enabled=True)
200
201 result = await plugin._handle_text_search("disco")
202
203 assert result["analyzed"] is False
204 assert result["reason"] == "clap_index_empty"
205
206 @pytest.mark.asyncio
207 async def test_text_encoder_unavailable_when_load_fails(
208 self, make_plugin: Callable[..., Any]
209 ) -> None:
210 """When the encoder fails to load the hook reports text_encoder_unavailable."""
211 plugin = make_plugin(clap_enabled=True)
212 plugin._clap_index.__len__ = MagicMock(return_value=5)
213 plugin._load_text_encoder = MagicMock(side_effect=RuntimeError("nope"))
214
215 result = await plugin._handle_text_search("disco")
216
217 assert result["analyzed"] is False
218 assert result["reason"] == "text_encoder_unavailable"
219 assert result["items"] == []
220
221 @pytest.mark.asyncio
222 async def test_happy_path_returns_ranked_items_without_resolve(
223 self, make_plugin: Callable[..., Any]
224 ) -> None:
225 """resolve=False returns ranked (provider, item_id, distance) entries."""
226 plugin = make_plugin(clap_enabled=True)
227 plugin._clap_index.__len__ = MagicMock(return_value=5)
228 vector = _unit_vector(1.0)
229 plugin._load_text_encoder = lambda: _make_mock_encoder(vector)
230 plugin._clap_index.search = AsyncMock(
231 return_value=[
232 ScoredCandidate("track_a", "spotify", 0.1),
233 ScoredCandidate("track_b", "tidal", 0.2),
234 ]
235 )
236
237 result = await plugin._handle_text_search("disco", limit=5)
238
239 assert result["analyzed"] is True
240 assert result["query"] == "disco"
241 assert result["items"] == [
242 {"provider": "spotify", "item_id": "track_a", "distance": 0.1},
243 {"provider": "tidal", "item_id": "track_b", "distance": 0.2},
244 ]
245
246 @pytest.mark.asyncio
247 async def test_happy_path_resolve_true_adds_name_and_artist(
248 self, make_plugin: Callable[..., Any], mock_mass: MagicMock
249 ) -> None:
250 """resolve=True augments each entry with name + comma-joined artist string."""
251 plugin = make_plugin(clap_enabled=True)
252 plugin._clap_index.__len__ = MagicMock(return_value=5)
253 vector = _unit_vector(1.0)
254 plugin._load_text_encoder = lambda: _make_mock_encoder(vector)
255 plugin._clap_index.search = AsyncMock(
256 return_value=[
257 ScoredCandidate("track_a", "spotify", 0.1),
258 ScoredCandidate("track_b", "tidal", 0.2),
259 ]
260 )
261 mock_mass.music.tracks.get.side_effect = [
262 make_track("track_a", provider="spotify", name="X", artists=("A1", "A2")),
263 make_track("track_b", provider="tidal", name="Y", artists=("B1",)),
264 ]
265
266 result = await plugin._handle_text_search("disco", resolve=True)
267
268 assert result["analyzed"] is True
269 items_by_id = {entry["item_id"]: entry for entry in result["items"]}
270 assert items_by_id["track_a"]["name"] == "X"
271 assert items_by_id["track_a"]["artist"] == "A1, A2"
272 assert items_by_id["track_b"]["name"] == "Y"
273 assert items_by_id["track_b"]["artist"] == "B1"
274
275 @pytest.mark.asyncio
276 async def test_resolve_true_handles_music_assistant_error(
277 self, make_plugin: Callable[..., Any], mock_mass: MagicMock
278 ) -> None:
279 """A failed resolve falls back to '(unknown)'/'' but still returns the entry."""
280 plugin = make_plugin(clap_enabled=True)
281 plugin._clap_index.__len__ = MagicMock(return_value=5)
282 vector = _unit_vector(1.0)
283 plugin._load_text_encoder = lambda: _make_mock_encoder(vector)
284 plugin._clap_index.search = AsyncMock(
285 return_value=[
286 ScoredCandidate("track_a", "spotify", 0.1),
287 ScoredCandidate("track_b", "tidal", 0.2),
288 ]
289 )
290 mock_mass.music.tracks.get.side_effect = [
291 MusicAssistantError("not found"),
292 make_track("track_b", provider="tidal", name="Y", artists=("B1",)),
293 ]
294
295 result = await plugin._handle_text_search("disco", resolve=True)
296
297 assert result["analyzed"] is True
298 items_by_id = {entry["item_id"]: entry for entry in result["items"]}
299 assert items_by_id["track_a"]["name"] == "(unknown)"
300 assert items_by_id["track_a"]["artist"] == ""
301 assert items_by_id["track_b"]["name"] == "Y"
302 assert items_by_id["track_b"]["artist"] == "B1"
303