/
/
/
1"""Tests for the sonic similarity plugin API parameter handling and metadata filters."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9
10from music_assistant.providers.sonic_similarity.constants import (
11 METADATA_BONUS_SCALE,
12 SIMILARITY_PRESETS,
13)
14from music_assistant.providers.sonic_similarity.helpers import (
15 _parse_similar_params,
16 _parse_weights,
17 apply_filters,
18 format_text_query,
19)
20from music_assistant.providers.sonic_similarity.similarity import Candidate, ScoredCandidate
21from music_assistant.providers.sonic_similarity.vectors import FEATURE_GROUPS
22from tests.providers.sonic_similarity.conftest import make_track
23
24if TYPE_CHECKING:
25 from collections.abc import Callable
26 from typing import Any
27
28
29class TestParseSimilarParams:
30 """Tests for _parse_similar_params validation and normalization."""
31
32 def test_item_id_alias(self) -> None:
33 """Single item_id string wraps into item_ids list."""
34 params = _parse_similar_params(item_id="abc")
35 assert params.item_ids == ["abc"]
36
37 def test_item_ids_list(self) -> None:
38 """item_ids list passes through directly."""
39 params = _parse_similar_params(item_ids=["a", "b"])
40 assert params.item_ids == ["a", "b"]
41
42 def test_item_id_and_item_ids_prefers_ids(self) -> None:
43 """When both provided, item_ids takes precedence."""
44 params = _parse_similar_params(item_id="old", item_ids=["new"])
45 assert params.item_ids == ["new"]
46
47 def test_no_ids_raises(self) -> None:
48 """Must provide at least item_id or item_ids."""
49 with pytest.raises(ValueError, match="item_id"):
50 _parse_similar_params()
51
52 def test_limit_clamped(self) -> None:
53 """Limit is clamped to [1, 100]."""
54 assert _parse_similar_params(item_id="x", limit=0).limit == 1
55 assert _parse_similar_params(item_id="x", limit=200).limit == 100
56 assert _parse_similar_params(item_id="x", limit=50).limit == 50
57
58 def test_depth_clamped(self) -> None:
59 """Depth is clamped to [1, 5]."""
60 assert _parse_similar_params(item_id="x", depth=0).depth == 1
61 assert _parse_similar_params(item_id="x", depth=10).depth == 5
62
63 def test_diversity_clamped(self) -> None:
64 """Diversity is clamped to [0.0, 1.0]."""
65 assert _parse_similar_params(item_id="x", diversity=-1.0).diversity == 0.0
66 assert _parse_similar_params(item_id="x", diversity=5.0).diversity == 1.0
67
68 def test_blend_mode_validated(self) -> None:
69 """Invalid blend_mode falls back to centroid."""
70 assert _parse_similar_params(item_id="x", blend_mode="centroid").blend_mode == "centroid"
71 assert _parse_similar_params(item_id="x", blend_mode="union").blend_mode == "union"
72 assert _parse_similar_params(item_id="x", blend_mode="invalid").blend_mode == "centroid"
73
74 def test_seed_weights_length_validated(self) -> None:
75 """seed_weights length must match item_ids."""
76 with pytest.raises(ValueError, match="seed_weights"):
77 _parse_similar_params(item_ids=["a", "b"], seed_weights=[1.0])
78
79 def test_candidates_scaled_with_filters(self) -> None:
80 """Candidates doubled when filters are active."""
81 no_filter = _parse_similar_params(item_id="x", candidates=50)
82 with_filter = _parse_similar_params(item_id="x", candidates=50, filter_genres=["jazz"])
83 assert with_filter.candidates == no_filter.candidates * 2
84
85 def test_defaults(self) -> None:
86 """Verify all default values."""
87 params = _parse_similar_params(item_id="x")
88 assert params.limit == 25
89 assert params.depth == 1
90 assert params.branch_factor == 5
91 assert params.blend_mode == "centroid"
92 assert params.candidates == 200
93 assert params.seed_weights is None
94 assert params.diversity == 0.0
95 assert params.preset == "balanced"
96 assert params.resolve is False
97 assert params.filter_genres is None
98 assert params.filter_providers is None
99 assert params.exclude_track_ids is None
100 assert params.exclude_artists is None
101
102
103class TestFormatTextQuery:
104 """Tests for the CLAP query template helper."""
105
106 def test_appends_music_suffix(self) -> None:
107 """A bare query is framed toward CLAP's caption form."""
108 assert format_text_query("aggressive metal") == "aggressive metal music"
109
110 def test_skips_when_music_present(self) -> None:
111 """No double 'music' when the query already mentions it."""
112 assert format_text_query("upbeat dance music") == "upbeat dance music"
113 assert format_text_query("Music for studying") == "Music for studying"
114
115 def test_matches_word_not_substring(self) -> None:
116 """The skip guard matches the word 'music', not substrings like 'musician'."""
117 assert format_text_query("musician") == "musician music"
118 assert format_text_query("musical theatre") == "musical theatre music"
119
120 def test_strips_whitespace(self) -> None:
121 """Surrounding whitespace is trimmed before framing."""
122 assert format_text_query(" jazzy ") == "jazzy music"
123
124 def test_empty_stays_empty(self) -> None:
125 """An empty/whitespace query is left as an empty string."""
126 assert format_text_query(" ") == ""
127
128
129class TestApplyFilters:
130 """Tests for post-ANN filter pipeline."""
131
132 def test_no_filters_passes_all(self) -> None:
133 """All candidates pass with no active filters."""
134 candidates = [ScoredCandidate("a", "prov1", 0.1), ScoredCandidate("b", "prov2", 0.2)]
135 result = apply_filters(
136 candidates, seed_ids=set(), exclude_track_ids=None, filter_providers=None
137 )
138 assert len(result) == 2
139
140 def test_exclude_seed_ids(self) -> None:
141 """Seed IDs are always excluded."""
142 candidates = [ScoredCandidate("seed", "prov1", 0.1), ScoredCandidate("other", "prov1", 0.2)]
143 result = apply_filters(
144 candidates, seed_ids={"seed"}, exclude_track_ids=None, filter_providers=None
145 )
146 assert len(result) == 1
147 assert result[0].item_id == "other"
148
149 def test_exclude_track_ids(self) -> None:
150 """Explicitly excluded track IDs are removed."""
151 candidates = [
152 ScoredCandidate("a", "p", 0.1),
153 ScoredCandidate("b", "p", 0.2),
154 ScoredCandidate("c", "p", 0.3),
155 ]
156 result = apply_filters(
157 candidates, seed_ids=set(), exclude_track_ids={"a", "c"}, filter_providers=None
158 )
159 assert [r.item_id for r in result] == ["b"]
160
161 def test_filter_providers(self) -> None:
162 """Only candidates from listed providers are kept."""
163 candidates = [
164 ScoredCandidate("a", "prov1", 0.1),
165 ScoredCandidate("b", "prov2", 0.2),
166 ScoredCandidate("c", "prov1", 0.3),
167 ]
168 result = apply_filters(
169 candidates, seed_ids=set(), exclude_track_ids=None, filter_providers={"prov1"}
170 )
171 assert [r.item_id for r in result] == ["a", "c"]
172
173 def test_all_filters_combined(self) -> None:
174 """Filters stack: seed exclusion + exclude_track_ids + filter_providers."""
175 candidates = [
176 ScoredCandidate("seed", "prov1", 0.0),
177 ScoredCandidate("a", "prov1", 0.1),
178 ScoredCandidate("b", "prov2", 0.2),
179 ScoredCandidate("c", "prov1", 0.3),
180 ]
181 result = apply_filters(
182 candidates, seed_ids={"seed"}, exclude_track_ids={"c"}, filter_providers={"prov1"}
183 )
184 assert [r.item_id for r in result] == ["a"]
185
186
187class TestParseWeights:
188 """Tests for _parse_weights dict-based weight parsing."""
189
190 def test_default_preset(self) -> None:
191 """Empty params return balanced preset defaults."""
192 result = _parse_weights({})
193 balanced = SIMILARITY_PRESETS["balanced"]
194 assert result == balanced
195
196 def test_named_preset(self) -> None:
197 """Selecting a preset by name uses its values."""
198 result = _parse_weights({"preset": "party"})
199 party = SIMILARITY_PRESETS["party"]
200 assert result["rhythm"] == party["rhythm"]
201 assert result["timbre"] == party["timbre"]
202
203 def test_unknown_preset_falls_back(self) -> None:
204 """Unknown preset name falls back to balanced."""
205 result = _parse_weights({"preset": "nonexistent"})
206 assert result == SIMILARITY_PRESETS["balanced"]
207
208 def test_individual_override(self) -> None:
209 """Individual weight overrides take precedence over preset."""
210 result = _parse_weights({"preset": "balanced", "rhythm_weight": "0.3"})
211 assert abs(result["rhythm"] - 0.3) < 0.01
212 assert result["timbre"] == SIMILARITY_PRESETS["balanced"]["timbre"]
213
214 def test_clamping(self) -> None:
215 """Values outside [0, 1] are clamped."""
216 result = _parse_weights({"rhythm_weight": "1.5", "timbre_weight": "-0.3"})
217 assert result["rhythm"] == 1.0
218 assert result["timbre"] == 0.0
219
220 def test_invalid_string_falls_back(self) -> None:
221 """Non-numeric string falls back to preset default."""
222 result = _parse_weights({"preset": "vibe", "rhythm_weight": "abc"})
223 assert result["rhythm"] == SIMILARITY_PRESETS["vibe"]["rhythm"]
224
225 def test_returns_dict(self) -> None:
226 """Result is a plain dict with every preset weight key (7 audio groups + 2 metadata)."""
227 result = _parse_weights({})
228 assert isinstance(result, dict)
229 for key in (
230 "rhythm",
231 "loudness",
232 "timbre",
233 "regularity",
234 "mood",
235 "tonal",
236 "dynamics",
237 "genre",
238 "era",
239 ):
240 assert key in result
241
242
243class TestSimilarityPresets:
244 """
245 Structural invariants over the SIMILARITY_PRESETS weight dicts.
246
247 Each preset is a hand-edited dict; these guard against a typo (a dropped,
248 misspelled or extra key, or an out-of-range value) shipping as a runtime
249 KeyError or a degenerate weighting.
250 """
251
252 @pytest.mark.parametrize("preset", SIMILARITY_PRESETS.values(), ids=SIMILARITY_PRESETS.keys())
253 def test_exact_key_set(self, preset: dict[str, float]) -> None:
254 """A preset weights exactly the feature groups plus the genre/era knobs."""
255 assert set(preset) == set(FEATURE_GROUPS) | {"genre", "era"}
256
257 @pytest.mark.parametrize("preset", SIMILARITY_PRESETS.values(), ids=SIMILARITY_PRESETS.keys())
258 def test_weights_in_unit_range(self, preset: dict[str, float]) -> None:
259 """Every weight sits in [0, 1] (preset baselines are not clamped at use)."""
260 assert all(0.0 <= w <= 1.0 for w in preset.values())
261
262 @pytest.mark.parametrize("preset", SIMILARITY_PRESETS.values(), ids=SIMILARITY_PRESETS.keys())
263 def test_has_a_nonzero_audio_group(self, preset: dict[str, float]) -> None:
264 """At least one audio group is weighted, so the distance stays meaningful."""
265 assert any(preset[group] > 0.0 for group in FEATURE_GROUPS)
266
267
268class TestSingleSeedAPI:
269 """Cover the single-seed `item_id=` API: parsing, options, and parity with `item_ids=[...]`."""
270
271 def test_single_and_multi_seed_param_parse(self) -> None:
272 """item_id='abc' produces same params as item_ids=['abc']."""
273 single = _parse_similar_params(item_id="abc")
274 multi = _parse_similar_params(item_ids=["abc"])
275 assert single.item_ids == multi.item_ids
276 assert single.limit == multi.limit
277 assert single.depth == multi.depth
278
279 def test_single_seed_with_limit_and_preset(self) -> None:
280 """Single-seed call with extra kwargs works."""
281 params = _parse_similar_params(item_id="abc", limit=10, preset="vibe")
282 assert params.item_ids == ["abc"]
283 assert params.limit == 10
284 assert params.preset == "vibe"
285
286 def test_single_seed_with_weight_overrides(self) -> None:
287 """Weight kwargs pass through."""
288 params = _parse_similar_params(item_id="abc", timbre_weight="0.5")
289 assert params.weight_overrides["timbre_weight"] == "0.5"
290
291
292class TestHandleSimilarReason:
293 """_handle_similar surfaces a `reason` that distinguishes empty-response causes."""
294
295 @pytest.mark.asyncio
296 async def test_corpus_not_ready_reason(self, make_plugin: Callable[..., Any]) -> None:
297 """No corpus â reason = corpus_not_ready (regardless of seed match)."""
298 plugin = make_plugin() # no signatures â corpus_means/stds stay None
299
300 result = await plugin._handle_similar(item_id="anything")
301
302 assert result["analyzed"] is False
303 assert result["reason"] == "corpus_not_ready"
304 assert result["items"] == []
305
306 @pytest.mark.asyncio
307 async def test_seed_not_in_index_reason(self, make_plugin: Callable[..., Any]) -> None:
308 """Corpus ready but no matching seed â reason = seed_not_in_index."""
309 plugin = make_plugin(signatures={("spotify", "known_seed"): [0.1] * 18})
310
311 result = await plugin._handle_similar(item_id="unknown_seed")
312
313 assert result["analyzed"] is False
314 assert result["reason"] == "seed_not_in_index"
315 assert result["items"] == []
316
317
318def _track_with_metadata(
319 item_id: str,
320 *,
321 provider: str = "spotify",
322 genres: list[str] | None = None,
323 artist_names: tuple[str, ...] = (),
324) -> MagicMock:
325 """Build a Track-like mock with ``metadata.genres`` set (make_track doesn't)."""
326 track = make_track(item_id, provider=provider, artists=artist_names)
327 if genres is not None:
328 track.metadata = MagicMock()
329 track.metadata.genres = genres
330 else:
331 track.metadata = None
332 return track
333
334
335def _make_candidate(item_id: str, *, distance: float = 0.5) -> Candidate:
336 """Build a Candidate with the minimum fields needed for filter/rerank tests."""
337 return Candidate(
338 item_id=item_id, provider="spotify", features=[0.0] * 18, distance=distance, generation=0
339 )
340
341
342class TestApplyMetadataFilters:
343 """_apply_metadata_filters drops candidates that don't pass the genre/artist gates."""
344
345 @pytest.mark.asyncio
346 async def test_no_filters_returns_input_unchanged(
347 self, make_plugin: Callable[..., Any]
348 ) -> None:
349 """With no filter_genres and no exclude_artists the input passes through."""
350 plugin = make_plugin(signatures={("spotify", "a"): [0.1] * 18})
351 cands = [_make_candidate("a"), _make_candidate("b")]
352
353 result = await plugin._apply_metadata_filters(cands, {})
354
355 assert result == cands
356
357 @pytest.mark.asyncio
358 async def test_genre_filter_drops_non_matching(self, make_plugin: Callable[..., Any]) -> None:
359 """Candidates without overlapping genres are dropped."""
360 plugin = make_plugin(signatures={("spotify", "a"): [0.1] * 18})
361 resolved = {
362 ("a", "spotify"): _track_with_metadata("a", genres=["Rock", "Indie"]),
363 ("b", "spotify"): _track_with_metadata("b", genres=["Pop"]),
364 }
365
366 result = await plugin._apply_metadata_filters(
367 [_make_candidate("a"), _make_candidate("b")],
368 resolved,
369 filter_genres=["rock"],
370 )
371
372 assert [c.item_id for c in result] == ["a"]
373
374 @pytest.mark.asyncio
375 async def test_artist_exclusion_drops_matching(self, make_plugin: Callable[..., Any]) -> None:
376 """Candidates whose artist is in exclude_artists are dropped (case-insensitive)."""
377 plugin = make_plugin(signatures={("spotify", "a"): [0.1] * 18})
378 resolved = {
379 ("a", "spotify"): _track_with_metadata("a", artist_names=("Banned Artist",)),
380 ("b", "spotify"): _track_with_metadata("b", artist_names=("Other Artist",)),
381 }
382
383 result = await plugin._apply_metadata_filters(
384 [_make_candidate("a"), _make_candidate("b")],
385 resolved,
386 exclude_artists=["banned artist"],
387 )
388
389 assert [c.item_id for c in result] == ["b"]
390
391 @pytest.mark.asyncio
392 async def test_unresolved_tracks_are_dropped(self, make_plugin: Callable[..., Any]) -> None:
393 """Candidates whose track resolve returned None are silently dropped."""
394 plugin = make_plugin(signatures={("spotify", "a"): [0.1] * 18})
395 resolved = {
396 ("a", "spotify"): None, # unresolved miss
397 ("b", "spotify"): _track_with_metadata("b", genres=["rock"]),
398 }
399
400 result = await plugin._apply_metadata_filters(
401 [_make_candidate("a"), _make_candidate("b")],
402 resolved,
403 filter_genres=["rock"],
404 )
405
406 assert [c.item_id for c in result] == ["b"]
407
408
409class TestApplyMetadataReranking:
410 """_apply_metadata_reranking shifts distances based on shared genre/year metadata."""
411
412 @pytest.mark.asyncio
413 async def test_no_seed_tracks_returns_unchanged(
414 self, make_plugin: Callable[..., Any], mock_mass: MagicMock
415 ) -> None:
416 """If every seed resolve fails, return the input list as-is."""
417 from music_assistant_models.errors import MusicAssistantError # noqa: PLC0415
418
419 plugin = make_plugin(signatures={("spotify", "seed"): [0.1] * 18})
420 mock_mass.music.tracks.get = AsyncMock(side_effect=MusicAssistantError("seed missing"))
421 cands = [_make_candidate("a", distance=0.5)]
422
423 result = await plugin._apply_metadata_reranking(["seed"], cands, {"genre": 1.0}, {})
424
425 assert result == cands
426
427 @pytest.mark.asyncio
428 async def test_genre_overlap_reduces_distance(
429 self, make_plugin: Callable[..., Any], mock_mass: MagicMock
430 ) -> None:
431 """A candidate sharing all genres with the seed gets a negative bonus."""
432 plugin = make_plugin(signatures={("spotify", "seed"): [0.1] * 18})
433 mock_mass.music.tracks.get = AsyncMock(
434 return_value=_track_with_metadata("seed", genres=["rock", "indie"])
435 )
436 resolved = {("a", "spotify"): _track_with_metadata("a", genres=["rock", "indie"])}
437
438 result = await plugin._apply_metadata_reranking(
439 ["seed"], [_make_candidate("a", distance=0.5)], {"genre": 1.0}, resolved
440 )
441
442 # Full overlap (jaccard = 1.0) with weight 1.0 â bonus = -METADATA_BONUS_SCALE
443 assert result[0].distance == pytest.approx(0.5 - METADATA_BONUS_SCALE)
444
445 @pytest.mark.asyncio
446 async def test_zero_weight_leaves_distance_unchanged(
447 self, make_plugin: Callable[..., Any], mock_mass: MagicMock
448 ) -> None:
449 """genre/era weights of 0 produce no rerank bonus even with full overlap."""
450 plugin = make_plugin(signatures={("spotify", "seed"): [0.1] * 18})
451 mock_mass.music.tracks.get = AsyncMock(
452 return_value=_track_with_metadata("seed", genres=["rock"])
453 )
454 resolved = {("a", "spotify"): _track_with_metadata("a", genres=["rock"])}
455
456 result = await plugin._apply_metadata_reranking(
457 ["seed"], [_make_candidate("a", distance=0.5)], {"genre": 0.0, "era": 0.0}, resolved
458 )
459
460 assert result[0].distance == pytest.approx(0.5)
461
462 @pytest.mark.asyncio
463 async def test_results_sorted_by_distance_after_rerank(
464 self, make_plugin: Callable[..., Any], mock_mass: MagicMock
465 ) -> None:
466 """After rerank, candidates are sorted by the new distance ascending."""
467 plugin = make_plugin(signatures={("spotify", "seed"): [0.1] * 18})
468 mock_mass.music.tracks.get = AsyncMock(
469 return_value=_track_with_metadata("seed", genres=["rock"])
470 )
471 resolved = {
472 ("a", "spotify"): _track_with_metadata("a", genres=["pop"]), # no overlap â no bonus
473 ("b", "spotify"): _track_with_metadata("b", genres=["rock"]), # overlap â bonus
474 }
475
476 result = await plugin._apply_metadata_reranking(
477 ["seed"],
478 [_make_candidate("a", distance=0.4), _make_candidate("b", distance=0.45)],
479 {"genre": 1.0},
480 resolved,
481 )
482
483 # b started farther (0.45) but got the -0.1 bonus â ends ahead of a (0.4).
484 assert [c.item_id for c in result] == ["b", "a"]
485
486 @pytest.mark.asyncio
487 async def test_bonus_is_bounded_by_scale(
488 self, make_plugin: Callable[..., Any], mock_mass: MagicMock
489 ) -> None:
490 """Even max overlap with weight=1.0 doesn't shift distance more than METADATA_BONUS_SCALE."""
491 plugin = make_plugin(signatures={("spotify", "seed"): [0.1] * 18})
492 mock_mass.music.tracks.get = AsyncMock(
493 return_value=_track_with_metadata("seed", genres=["rock", "indie", "alt"])
494 )
495 resolved = {("a", "spotify"): _track_with_metadata("a", genres=["rock", "indie", "alt"])}
496
497 result = await plugin._apply_metadata_reranking(
498 ["seed"], [_make_candidate("a", distance=0.5)], {"genre": 1.0}, resolved
499 )
500
501 shift = 0.5 - result[0].distance
502 assert shift <= METADATA_BONUS_SCALE + 1e-9
503