pluto
b2c82f36d7
TDD: 3 tests covering silence passthrough, speech→silence segment emission, and short-speech discard. Uses _speech_ms tracking (not total buffer length) for accurate min_speech_ms enforcement. Silero VAD import is try/except'd so tests run without torch via mocker.patch on vad_prob. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
import numpy as np
|
|
import pytest
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
|
|
def _make_chunk(samples: int = 4000, amplitude: float = 0.0) -> bytes:
|
|
"""Create a fake PCM int16 chunk."""
|
|
audio = (np.ones(samples) * amplitude * 32767).astype(np.int16)
|
|
return audio.tobytes()
|
|
|
|
|
|
@pytest.fixture
|
|
def buf(mocker):
|
|
"""RoomVADBuffer with mocked VAD model."""
|
|
mocker.patch(
|
|
"app.pipeline.ingestion.vad_prob",
|
|
side_effect=lambda chunk: 0.9 if np.frombuffer(chunk, np.int16).max() > 100 else 0.1,
|
|
)
|
|
from app.pipeline.ingestion import RoomVADBuffer
|
|
return RoomVADBuffer(silence_threshold_ms=600, min_speech_ms=500, chunk_ms=250)
|
|
|
|
|
|
def test_silence_produces_no_segment(buf):
|
|
ts = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
for _ in range(10):
|
|
result = buf.process_chunk(_make_chunk(amplitude=0.0), ts)
|
|
assert result is None
|
|
|
|
|
|
def test_speech_then_silence_produces_segment(buf):
|
|
ts = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
# 4 speech chunks = 1 second of speech
|
|
for _ in range(4):
|
|
result = buf.process_chunk(_make_chunk(amplitude=0.5), ts)
|
|
assert result is None # still accumulating
|
|
# 3 silence chunks = 750ms silence (> 600ms threshold)
|
|
segment = None
|
|
for _ in range(3):
|
|
segment = buf.process_chunk(_make_chunk(amplitude=0.0), ts)
|
|
assert segment is not None
|
|
assert isinstance(segment, tuple)
|
|
audio_bytes, start_time = segment
|
|
assert len(audio_bytes) > 0
|
|
|
|
|
|
def test_short_speech_discarded(buf):
|
|
ts = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
# 1 speech chunk = 250ms (below 500ms min)
|
|
buf.process_chunk(_make_chunk(amplitude=0.5), ts)
|
|
# Now silence to trigger emission
|
|
result = None
|
|
for _ in range(3):
|
|
result = buf.process_chunk(_make_chunk(amplitude=0.0), ts)
|
|
assert result is None # discarded as too short
|