music-assistant-server

1"""Unit tests for plugin setup-time behavior, status text, and action dispatch."""
2
3from __future__ import annotations
4
5import asyncio
6from types import SimpleNamespace
7from typing import TYPE_CHECKING
8from unittest.mock import AsyncMock, MagicMock
9
10import pytest
11from music_assistant_models.enums import EventType
12from music_assistant_models.errors import ActionUnavailable, SetupFailedError
13
14if TYPE_CHECKING:
15    from collections.abc import Callable
16    from pathlib import Path
17    from typing import Any
18
19from music_assistant.providers.sonic_similarity import SonicSimilarityPlugin
20from music_assistant.providers.sonic_similarity import clap_index as clap_index_module
21from music_assistant.providers.sonic_similarity.constants import (
22    ACTION_REBUILD_18DIM,
23    ACTION_REBUILD_CLAP,
24    SUPPORTED_FEATURES,
25)
26
27
28class TestCollectStatusText:
29    """Tests for the _collect_status_text helper used by the plugin's status rows."""
30
31    @pytest.mark.asyncio
32    async def test_reports_pending_state_when_index_not_built(
33        self, make_plugin: Callable[..., Any]
34    ) -> None:
35        """A loaded plugin with no indexed corpus reports empty counts and disabled engines."""
36        plugin = make_plugin()  # no signatures → empty corpus, no search index
37
38        eighteen, clap, text = await plugin._collect_status_text()
39
40        assert "0 tracks indexed" in eighteen
41        assert "corpus stats pending" in eighteen
42        assert clap == "Character engine: disabled"
43        assert text == "Text encoder: disabled"
44
45    @pytest.mark.asyncio
46    async def test_returns_populated_18dim_status_when_provider_is_loaded(
47        self, make_plugin: Callable[..., Any]
48    ) -> None:
49        """Loaded plugin with a primed corpus yields a populated 18-dim status line."""
50        plugin = make_plugin(
51            signatures={
52                ("spotify", "a"): [0.1] * 18,
53                ("spotify", "b"): [0.2] * 18,
54            }
55        )
56
57        eighteen, _clap, _text = await plugin._collect_status_text()
58
59        assert "2 tracks indexed" in eighteen
60        assert ("corpus stats ready" in eighteen) or ("2 signatures cached" in eighteen)
61
62    @pytest.mark.asyncio
63    async def test_clap_engine_disabled_when_clap_index_is_none(
64        self, make_plugin: Callable[..., Any]
65    ) -> None:
66        """Without CLAP enabled the clap status string is the disabled sentinel."""
67        plugin = make_plugin(signatures={("spotify", "a"): [0.1] * 18})
68
69        _eighteen, clap, _text = await plugin._collect_status_text()
70
71        assert clap == "Character engine: disabled"
72
73    @pytest.mark.asyncio
74    async def test_clap_engine_status_reports_size_when_enabled(
75        self, make_plugin: Callable[..., Any]
76    ) -> None:
77        """When CLAP is enabled the clap line reports the index size."""
78        plugin = make_plugin(
79            clap_enabled=True,
80            signatures={("spotify", "a"): [0.1] * 18},
81        )
82        plugin._clap_index.__len__ = MagicMock(return_value=42)
83
84        _eighteen, clap, _text = await plugin._collect_status_text()
85
86        assert "42 embeddings indexed" in clap
87
88    @pytest.mark.asyncio
89    async def test_text_encoder_cold_message_when_enabled_and_encoder_none(
90        self, make_plugin: Callable[..., Any]
91    ) -> None:
92        """Text-search-enabled with no encoder loaded reports a cold-state message."""
93        plugin = make_plugin(
94            text_search_enabled=True,
95            signatures={("spotify", "a"): [0.1] * 18},
96        )
97
98        _eighteen, _clap, text = await plugin._collect_status_text()
99
100        lowered = text.lower()
101        assert ("cold" in lowered) or ("downloads on first query" in lowered)
102
103    @pytest.mark.asyncio
104    async def test_coverage_pct_included_when_get_coverage_returns_counts(
105        self, mock_mass: MagicMock, make_plugin: Callable[..., Any]
106    ) -> None:
107        """A real coverage object produces a percentage substring in the 18-dim line."""
108        coverage = SimpleNamespace(analyzed=80, pending=20, stale_version=0, analysis_version=1)
109        mock_mass.streams.audio_analysis.get_coverage = AsyncMock(return_value=coverage)
110        plugin = make_plugin(signatures={("spotify", "a"): [0.1] * 18})
111
112        eighteen, _clap, _text = await plugin._collect_status_text()
113
114        assert "80.0%" in eighteen
115
116
117class TestConfigEntriesActions:
118    """Tests for handle_config_action (the one-shot rebuild buttons)."""
119
120    @pytest.mark.asyncio
121    async def test_action_rebuild_18dim_dispatches_to_provider(
122        self, mock_mass: MagicMock, make_plugin: Callable[..., Any]
123    ) -> None:
124        """The 18-dim rebuild action fires create_task once and returns None."""
125        plugin = make_plugin(signatures={("spotify", "a"): [0.1] * 18})
126        plugin._rebuild_search_index = AsyncMock()
127
128        result = await plugin.handle_config_action(ACTION_REBUILD_18DIM)
129
130        assert mock_mass.create_task.call_count == 1
131        assert result is None
132
133    @pytest.mark.asyncio
134    async def test_action_rebuild_clap_dispatches_when_clap_enabled(
135        self, mock_mass: MagicMock, make_plugin: Callable[..., Any]
136    ) -> None:
137        """The CLAP rebuild action fires create_task when _clap_index is present."""
138        plugin = make_plugin(
139            clap_enabled=True,
140            signatures={("spotify", "a"): [0.1] * 18},
141        )
142        plugin._rebuild_clap_index_from_database = AsyncMock()
143
144        await plugin.handle_config_action(ACTION_REBUILD_CLAP)
145
146        assert mock_mass.create_task.call_count == 1
147
148    @pytest.mark.asyncio
149    async def test_action_rebuild_clap_reports_failure_when_index_missing(
150        self, mock_mass: MagicMock, make_plugin: Callable[..., Any]
151    ) -> None:
152        """Without a CLAP index the rebuild action reports failure instead of silently passing."""
153        plugin = make_plugin(signatures={("spotify", "a"): [0.1] * 18})
154
155        with pytest.raises(ActionUnavailable) as exc_info:
156            await plugin.handle_config_action(ACTION_REBUILD_CLAP)
157
158        assert exc_info.value.translation_key == "clap_index_unavailable"
159        assert exc_info.value.translation_owner == "provider.sonic_similarity"
160        assert mock_mass.create_task.call_count == 0
161
162
163def _build_plugin_for_init(mock_mass: MagicMock) -> Any:
164    """
165    Construct a plugin without going through handle_async_init / loaded_in_mass.
166
167    Returned as ``Any`` to match the project's existing test convention for
168    plugin instances whose private methods get mock-swapped.
169    """
170    manifest = MagicMock()
171    manifest.instance_id = "iid"
172    manifest.domain = "sonic_similarity"
173    config = MagicMock()
174    config_values = {"log_level": "GLOBAL"}
175    config.get_value = lambda key: config_values.get(key)
176    return SonicSimilarityPlugin(mock_mass, manifest, config, SUPPORTED_FEATURES)
177
178
179class TestHandleAsyncInit:
180    """handle_async_init surfaces rebuild failures as SetupFailedError so MA's loader sees them."""
181
182    @pytest.mark.asyncio
183    async def test_raises_setup_failed_when_rebuild_raises(self, mock_mass: MagicMock) -> None:
184        """A rebuild failure during initial setup must surface as SetupFailedError."""
185        plugin = _build_plugin_for_init(mock_mass)
186        plugin._rebuild_search_index = AsyncMock(side_effect=RuntimeError("boom"))
187
188        with pytest.raises(SetupFailedError, match="Traits search index"):
189            await plugin.handle_async_init()
190
191    @pytest.mark.asyncio
192    async def test_succeeds_when_rebuild_succeeds(self, mock_mass: MagicMock) -> None:
193        """The happy path: rebuild runs once, no exception escapes."""
194        plugin = _build_plugin_for_init(mock_mass)
195        plugin._rebuild_search_index = AsyncMock()
196
197        await plugin.handle_async_init()
198
199        plugin._rebuild_search_index.assert_awaited_once()
200
201
202def _build_plugin_for_loaded(mock_mass: MagicMock, *, clap_enabled: bool) -> Any:
203    """Construct a plugin configured to attempt CLAP setup in loaded_in_mass."""
204    manifest = MagicMock()
205    manifest.instance_id = "iid"
206    manifest.domain = "sonic_similarity"
207    config = MagicMock()
208    config_values = {
209        "log_level": "GLOBAL",
210        "enable_clap_index": clap_enabled,
211        "enable_text_search": False,
212    }
213    config.get_value = lambda key: config_values.get(key)
214    return SonicSimilarityPlugin(mock_mass, manifest, config, SUPPORTED_FEATURES)
215
216
217class TestLoadedInMassClapResilience:
218    """loaded_in_mass must keep the 18-dim engine alive when optional CLAP setup fails."""
219
220    @pytest.mark.asyncio
221    async def test_clap_setup_failure_leaves_plugin_loaded(
222        self, mock_mass: MagicMock, monkeypatch: pytest.MonkeyPatch
223    ) -> None:
224        """A ClapIndex.load() failure is swallowed; _clap_index ends up None."""
225
226        async def _boom(_self: Any) -> None:
227            raise RuntimeError("usearch missing")
228
229        monkeypatch.setattr(clap_index_module.ClapIndex, "load", _boom)
230
231        plugin = _build_plugin_for_loaded(mock_mass, clap_enabled=True)
232        await plugin.loaded_in_mass()
233
234        assert plugin._clap_index is None
235
236
237class TestSafeRebuild:
238    """_safe_rebuild swallows background-task failures into _last_rebuild_error."""
239
240    @pytest.mark.asyncio
241    async def test_failure_populates_last_rebuild_error(self, mock_mass: MagicMock) -> None:
242        """A raising rebuild fn is caught; the error message lands in the dict."""
243        plugin = _build_plugin_for_init(mock_mass)
244
245        async def _boom() -> None:
246            raise RuntimeError("disk full")
247
248        await plugin._safe_rebuild("Traits", _boom)
249
250        assert plugin._last_rebuild_error == {"Traits": "disk full"}
251
252    @pytest.mark.asyncio
253    async def test_success_clears_prior_error(self, mock_mass: MagicMock) -> None:
254        """A subsequent successful rebuild removes the stale error entry."""
255        plugin = _build_plugin_for_init(mock_mass)
256        plugin._last_rebuild_error["Traits"] = "earlier failure"
257
258        async def _ok() -> None:
259            return None
260
261        await plugin._safe_rebuild("Traits", _ok)
262
263        assert "Traits" not in plugin._last_rebuild_error
264
265    @pytest.mark.asyncio
266    async def test_errors_are_per_label(self, mock_mass: MagicMock) -> None:
267        """A CLAP failure does not clobber an unrelated 18-dim error entry."""
268        plugin = _build_plugin_for_init(mock_mass)
269        plugin._last_rebuild_error["Traits"] = "existing 18-dim error"
270
271        async def _boom() -> None:
272            raise RuntimeError("clap broke")
273
274        await plugin._safe_rebuild("Character", _boom)
275
276        assert plugin._last_rebuild_error == {
277            "Traits": "existing 18-dim error",
278            "Character": "clap broke",
279        }
280
281    @pytest.mark.asyncio
282    async def test_success_signals_providers_updated(self, mock_mass: MagicMock) -> None:
283        """A finished rebuild signals PROVIDERS_UPDATED so the status labels re-render."""
284        plugin = _build_plugin_for_init(mock_mass)
285
286        async def _ok() -> None:
287            return None
288
289        await plugin._safe_rebuild("Traits", _ok)
290
291        mock_mass.signal_event.assert_called_once_with(
292            EventType.PROVIDERS_UPDATED, data=mock_mass.get_providers.return_value
293        )
294
295    @pytest.mark.asyncio
296    async def test_failure_signals_providers_updated(self, mock_mass: MagicMock) -> None:
297        """A failed rebuild still signals, so the recorded error reaches the status labels."""
298        plugin = _build_plugin_for_init(mock_mass)
299
300        async def _boom() -> None:
301            raise RuntimeError("disk full")
302
303        await plugin._safe_rebuild("Traits", _boom)
304
305        mock_mass.signal_event.assert_called_once_with(
306            EventType.PROVIDERS_UPDATED, data=mock_mass.get_providers.return_value
307        )
308
309
310class TestStatusTextRebuildErrors:
311    """_collect_status_text surfaces _last_rebuild_error entries on the matching engine line."""
312
313    @pytest.mark.asyncio
314    async def test_18dim_error_appears_on_18dim_line(self, mock_mass: MagicMock) -> None:
315        """A 18-dim rebuild error is appended to the 18-dim status line."""
316        plugin = _build_plugin_for_init(mock_mass)
317        plugin._last_rebuild_error["Traits"] = "disk full"
318
319        eighteen, _clap, _text = await plugin._collect_status_text()
320
321        assert "last rebuild failed: disk full" in eighteen
322
323    @pytest.mark.asyncio
324    async def test_clap_error_appears_on_clap_line(self, mock_mass: MagicMock) -> None:
325        """A CLAP rebuild error is appended to the CLAP status line."""
326        plugin = _build_plugin_for_init(mock_mass)
327        plugin._clap_index = MagicMock()
328        plugin._clap_index.__len__ = MagicMock(return_value=42)
329        plugin._last_rebuild_error["Character"] = "usearch native crash"
330
331        _eighteen, clap, _text = await plugin._collect_status_text()
332
333        assert "last rebuild failed: usearch native crash" in clap
334
335
336def _make_analysis_data() -> Any:
337    """Return an AudioAnalysisData with every assemble_vector-required field set."""
338    from music_assistant.models.audio_analysis import AudioAnalysisData  # noqa: PLC0415
339
340    return AudioAnalysisData(
341        bpm=120.0,
342        energy=0.5,
343        danceability=0.6,
344        loudness_integrated=-9.0,
345        loudness_range=4.0,
346        brightness=0.7,
347        harmonic_complexity=0.3,
348        roughness=0.2,
349        rhythmic_regularity=0.8,
350        key="C",
351        mode="major",
352    )
353
354
355class TestRebuildSearchIndexLocked:
356    """The 18-dim rebuild path: corpus stats, atomic swap, versioned file writes."""
357
358    @pytest.mark.asyncio
359    async def test_empty_iter_preserves_prior_state(self, mock_mass: MagicMock) -> None:
360        """No rows from the controller → early return, prior state untouched."""
361        plugin = _build_plugin_for_init(mock_mass)
362        plugin._signature_cache = {("spotify", "old"): [0.5] * 18}
363        plugin.corpus_means = [0.0] * 18
364        plugin.corpus_stds = [1.0] * 18
365        mock_mass._iter_merged_audio_analysis_rows_data = []
366
367        await plugin._rebuild_search_index_locked()
368
369        # Prior state preserved (no swap happened).
370        assert plugin._signature_cache == {("spotify", "old"): [0.5] * 18}
371        assert plugin.corpus_means == [0.0] * 18
372
373    @pytest.mark.asyncio
374    async def test_rows_present_but_unassemblable_preserves_prior_state(
375        self, mock_mass: MagicMock
376    ) -> None:
377        """Rows without the required scalar fields are skipped; index isn't rebuilt."""
378        plugin = _build_plugin_for_init(mock_mass)
379        plugin._signature_cache = {("spotify", "old"): [0.5] * 18}
380        plugin.corpus_means = [0.0] * 18
381
382        from music_assistant.models.audio_analysis import AudioAnalysisData  # noqa: PLC0415
383
384        unassemblable = AudioAnalysisData()  # all fields None
385        mock_mass._iter_merged_audio_analysis_rows_data = [
386            ("track_a", "spotify", unassemblable),
387        ]
388
389        await plugin._rebuild_search_index_locked()
390
391        # Prior state preserved — assemble_vector returned None for every row.
392        assert plugin._signature_cache == {("spotify", "old"): [0.5] * 18}
393
394    @pytest.mark.asyncio
395    async def test_happy_path_populates_state_and_writes_file(
396        self, mock_mass: MagicMock, tmp_path: Path
397    ) -> None:
398        """Valid rows produce a populated signature cache, corpus stats, and an on-disk index file."""
399        plugin = _build_plugin_for_init(mock_mass)
400        mock_mass.storage_path = str(tmp_path)
401        mock_mass._iter_merged_audio_analysis_rows_data = [
402            ("track_a", "spotify", _make_analysis_data()),
403            ("track_b", "tidal", _make_analysis_data()),
404        ]
405
406        await plugin._rebuild_search_index_locked()
407
408        assert ("track_a", "spotify") in plugin._signature_cache
409        assert ("track_b", "tidal") in plugin._signature_cache
410        assert plugin.corpus_means is not None
411        assert len(plugin.corpus_means) == 18
412        assert plugin.corpus_stds is not None
413        assert len(plugin.corpus_stds) == 18
414        assert len(plugin._search_index) == 2
415        # Versioned file matches the domain-aware template.
416        files = list(tmp_path.glob("sonic_signatures_sonic_analysis_*.usearch"))
417        assert len(files) == 1
418
419    @pytest.mark.asyncio
420    async def test_rebuild_cleans_up_prior_versioned_file(
421        self, mock_mass: MagicMock, tmp_path: Path
422    ) -> None:
423        """A successful rebuild unlinks the previous versioned file for the same domain."""
424        plugin = _build_plugin_for_init(mock_mass)
425        mock_mass.storage_path = str(tmp_path)
426        mock_mass._iter_merged_audio_analysis_rows_data = [
427            ("track_a", "spotify", _make_analysis_data()),
428        ]
429
430        await plugin._rebuild_search_index_locked()
431        first_files = list(tmp_path.glob("sonic_signatures_sonic_analysis_*.usearch"))
432        assert len(first_files) == 1
433
434        # Sleep one ms so the next rebuild's version timestamp differs.
435        await asyncio.sleep(0.002)
436
437        await plugin._rebuild_search_index_locked()
438        after_files = list(tmp_path.glob("sonic_signatures_sonic_analysis_*.usearch"))
439        assert len(after_files) == 1
440        # The new file replaces the old one (different timestamp).
441        assert after_files[0] != first_files[0]
442