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