music-assistant-server

1"""Tests for SonicAnalysisProvider._finalize base-class contract."""
2
3from __future__ import annotations
4
5import struct
6import time
7from unittest.mock import AsyncMock, MagicMock
8
9import numpy as np
10import pytest
11from music_assistant_models.enums import ContentType, MediaType
12from music_assistant_models.media_items import AudioFormat
13
14from music_assistant.models.audio_analysis import AudioAnalysisData, AudioAnalysisError
15from music_assistant.providers.sonic_analysis import SonicAnalysisProvider, SonicSessionData
16
17# ---------------------------------------------------------------------------
18# Helpers
19# ---------------------------------------------------------------------------
20
21
22def _make_provider() -> tuple[SonicAnalysisProvider, AsyncMock, AsyncMock]:
23    """
24    Construct a SonicAnalysisProvider with mocked MA infrastructure.
25
26    Uses ``__new__`` to bypass ``__init__`` (no model downloads) and manually
27    sets the attributes the finalize path touches.
28
29    :returns: ``(provider, set_audio_analysis_mock, post_analysis_mock)``
30    """
31    set_aa_mock = AsyncMock()
32    post_analysis_mock = AsyncMock()
33
34    mass = MagicMock()
35    mass.streams.audio_analysis.set_audio_analysis = set_aa_mock
36    mass.streams.audio_analysis.get_audio_analysis_version = AsyncMock(return_value=None)
37    mass.streams.audio_analysis.record_analysis_failure = AsyncMock()
38    mass.create_task = MagicMock(side_effect=lambda coro: coro.close() or MagicMock())
39
40    manifest = MagicMock()
41    manifest.domain = "sonic_analysis"
42
43    p = SonicAnalysisProvider.__new__(SonicAnalysisProvider)
44    p.logger = MagicMock()
45    p.mass = mass
46    p.manifest = manifest
47    p._sessions = {}
48    p._finalize_tasks = set()
49    p.unloading = False
50    p._clap_model = None
51    p._clap_prompt_order = []
52    p._clap_text_embeddings = None
53    p.analysis_version = 1
54    p.post_analysis = post_analysis_mock  # type: ignore[method-assign]
55    return p, set_aa_mock, post_analysis_mock
56
57
58def _make_streamdetails(item_id: str = "track-1") -> MagicMock:
59    """Return a minimal streamdetails mock."""
60    sd = MagicMock()
61    sd.item_id = item_id
62    sd.provider = "test_provider"
63    sd.media_type = MediaType.TRACK
64    sd.duration = None
65    return sd
66
67
68def _make_audio_format(
69    sample_rate: int = 22050,
70    bit_depth: int = 16,
71    channels: int = 1,
72    content_type: ContentType = ContentType.PCM_S16LE,
73) -> AudioFormat:
74    """Return a real AudioFormat for 16-bit mono PCM."""
75    return AudioFormat(
76        content_type=content_type,
77        sample_rate=sample_rate,
78        bit_depth=bit_depth,
79        channels=channels,
80    )
81
82
83def _make_pcm_sine(
84    sample_rate: int = 22050,
85    duration_sec: float = 12.0,
86) -> bytes:
87    """Generate a simple sine-wave PCM block sufficient for one 10-second feature block."""
88    n = int(sample_rate * duration_sec)
89    t = np.linspace(0, duration_sec, n, endpoint=False)
90    wave = (np.sin(2 * np.pi * 440 * t) * 0.5 * 32767).astype(np.int16)
91    return struct.pack(f"<{n}h", *wave)
92
93
94def _make_session(
95    provider: SonicAnalysisProvider,
96    session_id: str,
97    *,
98    sample_rate: int = 22050,
99    bit_depth: int = 16,
100    channels: int = 1,
101) -> SonicSessionData:
102    """
103    Register a fresh SonicSessionData for session_id in the provider.
104
105    :param provider: The provider instance to register the session on.
106    :param session_id: The session ID to register.
107    :param sample_rate: PCM sample rate in Hz.
108    :param bit_depth: PCM bit depth.
109    :param channels: PCM channel count.
110    :returns: The created SonicSessionData.
111    """
112    af = _make_audio_format(sample_rate=sample_rate, bit_depth=bit_depth, channels=channels)
113    sd = _make_streamdetails()
114    block_bytes = sample_rate * (bit_depth // 8) * channels * 10
115    session = SonicSessionData(
116        streamdetails=sd,
117        audio_format=af,
118        block_bytes=block_bytes,
119        start_time=time.monotonic(),
120        clap_target_starts=[],
121        clap_target_buffers=[],
122        clap_target_complete=[],
123    )
124    provider._sessions[session_id] = session
125    return session
126
127
128# ---------------------------------------------------------------------------
129# Test 1: happy path — _finalize returns AudioAnalysisData, base class persists
130# ---------------------------------------------------------------------------
131
132
133@pytest.mark.asyncio
134async def test_finalize_calls_set_audio_analysis_and_post_analysis() -> None:
135    """
136    finalize() must call set_audio_analysis and post_analysis exactly once each.
137
138    Before the fix, _finalize returns None so the base class skips both.
139    After the fix, _finalize returns AudioAnalysisData and the base class
140    calls set_audio_analysis + post_analysis exactly once.
141    """
142    provider, set_aa, post_analysis = _make_provider()
143    session_id = "test-session-happy"
144
145    session = _make_session(provider, session_id)
146    af = session.audio_format
147    pcm = _make_pcm_sine(sample_rate=af.sample_rate, duration_sec=12.0)
148
149    # Feed PCM so we have real feature blocks
150    await provider.process_pcm_chunk(session_id, pcm)
151
152    # Call the BASE CLASS finalize (not _finalize directly)
153    await provider.finalize(session_id)
154
155    # set_audio_analysis called once by the base class
156    set_aa.assert_called_once()
157    call_kwargs = set_aa.call_args.kwargs
158    assert call_kwargs["item_id"] == session.streamdetails.item_id
159    assert call_kwargs["aa_provider_domain"] == provider.domain
160    analysis_arg = call_kwargs["analysis"]
161    assert isinstance(analysis_arg, AudioAnalysisData)
162    assert analysis_arg.duration is not None
163    assert analysis_arg.duration > 0
164
165    # post_analysis called once with the persisted analysis
166    post_analysis.assert_called_once()
167    _, post_analysis_arg = post_analysis.call_args.args
168    assert post_analysis_arg is analysis_arg
169
170    # Session cleaned up by base class
171    assert session_id not in provider._sessions
172
173
174# ---------------------------------------------------------------------------
175# Test 2: short-audio early-return — set_audio_analysis and post_analysis not called
176# ---------------------------------------------------------------------------
177
178
179@pytest.mark.asyncio
180async def test_finalize_skips_persist_when_no_feature_blocks() -> None:
181    """
182    finalize() must not persist or fire post_analysis when no feature blocks exist.
183
184    This covers the early-return path when accumulated.rms_frames is empty
185    (e.g. audio too short to produce a single 10-second block).
186    """
187    provider, set_aa, post_analysis = _make_provider()
188    session_id = "test-session-empty"
189
190    # Do NOT call process_pcm_chunk — accumulated.rms_frames stays empty
191    _make_session(provider, session_id)
192
193    await provider.finalize(session_id)
194
195    set_aa.assert_not_called()
196    post_analysis.assert_not_called()
197    # Session is still cleaned up by base class
198    assert session_id not in provider._sessions
199
200
201# ---------------------------------------------------------------------------
202# Test 3: unknown session — no exception, no persist, debug log emitted
203# ---------------------------------------------------------------------------
204
205
206@pytest.mark.asyncio
207async def test_finalize_unknown_session_is_silent() -> None:
208    """
209    finalize() on an unknown session_id must not raise and must log at debug level.
210
211    This verifies the guard at the top of _finalize for sessions that were
212    already cancelled or never started.
213    """
214    provider, set_aa, post_analysis = _make_provider()
215
216    await provider.finalize("nonexistent-session-id")
217
218    set_aa.assert_not_called()
219    post_analysis.assert_not_called()
220    # A debug log should have been emitted for the unknown session
221    provider.logger.debug.assert_called()  # type: ignore[attr-defined]
222    debug_calls = [str(c) for c in provider.logger.debug.call_args_list]  # type: ignore[attr-defined]
223    assert any("nonexistent-session-id" in c for c in debug_calls)
224
225
226# ---------------------------------------------------------------------------
227# Test 4: empty accumulated — _finalize raises AudioAnalysisError
228# ---------------------------------------------------------------------------
229
230
231@pytest.mark.asyncio
232async def test_finalize_raises_when_no_feature_blocks() -> None:
233    """_finalize must raise AudioAnalysisError when no feature blocks were accumulated."""
234    provider, _set_aa, _post_analysis = _make_provider()
235    session_id = "test-session-empty-raise"
236    _make_session(provider, session_id)
237
238    with pytest.raises(AudioAnalysisError, match="no usable audio"):
239        await provider._finalize(session_id)
240