m4
c2743251e9
Self-evolving AI scientist framework built on LangGraph/LangChain with CLI/TUI core, FastAPI gateway, and Next.js frontend. Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>
285 lines
8.3 KiB
Python
285 lines
8.3 KiB
Python
"""Tests for STT (speech-to-text) module and channel integration.
|
|
|
|
Unit tests mock the heavy ML backends so they run without GPU/models.
|
|
See bottom of file for manual integration test instructions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from EvoScientist.stt import STT_MODELS, is_audio_file, transcribe_file
|
|
from tests.conftest import run_async
|
|
|
|
# ── is_audio_file ─────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "expected"),
|
|
[
|
|
("voice.ogg", True),
|
|
("audio.mp3", True),
|
|
("audio.wav", True),
|
|
("audio.m4a", True),
|
|
("audio.opus", True),
|
|
("image.jpg", False),
|
|
("video.mp4", False),
|
|
("doc.pdf", False),
|
|
("noext", False),
|
|
],
|
|
)
|
|
def test_is_audio_file(path, expected):
|
|
assert is_audio_file(path) is expected
|
|
|
|
|
|
# ── STT_MODELS mapping ────────────────────────────────────────────────
|
|
|
|
|
|
def test_stt_models_keys():
|
|
assert "zh" in STT_MODELS
|
|
assert "en" in STT_MODELS
|
|
assert "auto" in STT_MODELS
|
|
assert "faster-whisper" in STT_MODELS["zh"].lower()
|
|
assert "faster-whisper" in STT_MODELS["en"].lower()
|
|
assert "faster-whisper" in STT_MODELS["auto"].lower()
|
|
|
|
|
|
# ── transcribe_file: non-audio path ──────────────────────────────────
|
|
|
|
|
|
def test_transcribe_non_audio_returns_none():
|
|
result = run_async(transcribe_file("photo.jpg", language="auto"))
|
|
assert result is None
|
|
|
|
|
|
# ── transcribe_file: whisper backend ─────────────────────────────────
|
|
|
|
|
|
def _make_whisper_mock(text: str):
|
|
seg = MagicMock()
|
|
seg.text = text
|
|
seg.no_speech_prob = 0.0 # confident speech
|
|
model = MagicMock()
|
|
model.transcribe.return_value = ([seg], MagicMock())
|
|
return model
|
|
|
|
|
|
def _patch_whisper(whisper_model):
|
|
return patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"faster_whisper": MagicMock(
|
|
WhisperModel=MagicMock(return_value=whisper_model)
|
|
)
|
|
},
|
|
)
|
|
|
|
|
|
def test_transcribe_en_uses_whisper():
|
|
import EvoScientist.stt as stt_mod
|
|
|
|
stt_mod._engine = None
|
|
with _patch_whisper(_make_whisper_mock("Hello world")):
|
|
result = run_async(transcribe_file("voice.mp3", language="en"))
|
|
stt_mod._engine = None
|
|
assert result == "Hello world"
|
|
|
|
|
|
def test_transcribe_auto_uses_whisper():
|
|
import EvoScientist.stt as stt_mod
|
|
|
|
stt_mod._engine = None
|
|
with _patch_whisper(_make_whisper_mock("Bonjour monde")):
|
|
result = run_async(transcribe_file("voice.ogg", language="auto"))
|
|
stt_mod._engine = None
|
|
assert result == "Bonjour monde"
|
|
|
|
|
|
def test_transcribe_zh_uses_whisper():
|
|
import EvoScientist.stt as stt_mod
|
|
|
|
stt_mod._engine = None
|
|
with _patch_whisper(_make_whisper_mock("你好世界")):
|
|
result = run_async(transcribe_file("voice.ogg", language="zh"))
|
|
stt_mod._engine = None
|
|
assert result == "你好世界"
|
|
|
|
|
|
def test_transcribe_custom_model_override():
|
|
"""stt_model config overrides the default model mapping."""
|
|
import EvoScientist.stt as stt_mod
|
|
|
|
stt_mod._engine = None
|
|
captured_model_id = []
|
|
|
|
def patched_init(self, model_id, device, compute_type):
|
|
captured_model_id.append(model_id)
|
|
# don't actually load the model
|
|
self._model = _make_whisper_mock("test")
|
|
|
|
with patch.object(stt_mod._WhisperEngine, "__init__", patched_init):
|
|
run_async(
|
|
transcribe_file(
|
|
"voice.ogg", language="auto", model="openai/whisper-large-v3"
|
|
)
|
|
)
|
|
stt_mod._engine = None
|
|
assert captured_model_id == ["openai/whisper-large-v3"]
|
|
|
|
|
|
# ── transcribe_file: missing dependency ──────────────────────────────
|
|
|
|
|
|
def test_transcribe_missing_dep_returns_none():
|
|
import sys
|
|
|
|
import EvoScientist.stt as stt_mod
|
|
|
|
stt_mod._engine = None
|
|
saved = sys.modules.pop("faster_whisper", None)
|
|
try:
|
|
with patch.dict("sys.modules", {"faster_whisper": None}):
|
|
result = run_async(transcribe_file("voice.mp3", language="auto"))
|
|
finally:
|
|
if saved is not None:
|
|
sys.modules["faster_whisper"] = saved
|
|
stt_mod._engine = None
|
|
|
|
assert result is None
|
|
|
|
|
|
# ── Channel _enqueue_raw STT integration ─────────────────────────────
|
|
|
|
|
|
def _make_channel():
|
|
from EvoScientist.channels.telegram.channel import TelegramChannel, TelegramConfig
|
|
|
|
cfg = TelegramConfig(bot_token="dummy")
|
|
ch = TelegramChannel(cfg)
|
|
captured: list = []
|
|
|
|
async def _fake_build(raw):
|
|
captured.append(raw)
|
|
return None
|
|
|
|
ch._build_inbound_async = _fake_build # type: ignore[method-assign]
|
|
return ch, captured
|
|
|
|
|
|
def test_enqueue_raw_stt_prepends_transcript():
|
|
"""_enqueue_raw prepends STT transcript to raw.text when stt_enabled."""
|
|
from EvoScientist.channels.base import RawIncoming
|
|
|
|
ch, captured = _make_channel()
|
|
ch._stt_enabled = True
|
|
ch._stt_language = "zh"
|
|
ch._stt_model = ""
|
|
ch._stt_device = "cpu"
|
|
ch._stt_compute_type = "int8"
|
|
|
|
raw = RawIncoming(
|
|
sender_id="123",
|
|
chat_id="456",
|
|
text="",
|
|
media_files=["/tmp/voice.ogg"],
|
|
content_annotations=["[voice: /tmp/voice.ogg]"],
|
|
timestamp=datetime.now(),
|
|
)
|
|
|
|
async def _run():
|
|
with (
|
|
patch(
|
|
"EvoScientist.stt.transcribe_file", new=AsyncMock(return_value="你好")
|
|
),
|
|
patch("EvoScientist.stt.is_audio_file", return_value=True),
|
|
):
|
|
await ch._enqueue_raw(raw)
|
|
|
|
run_async(_run())
|
|
assert captured[0].text == "你好"
|
|
# annotation should be removed after transcription
|
|
assert captured[0].content_annotations == []
|
|
|
|
|
|
def test_enqueue_raw_stt_disabled_skips_transcription():
|
|
"""When stt_enabled=False, transcription is not called."""
|
|
from EvoScientist.channels.base import RawIncoming
|
|
|
|
ch, captured = _make_channel()
|
|
ch._stt_enabled = False
|
|
|
|
raw = RawIncoming(
|
|
sender_id="123",
|
|
chat_id="456",
|
|
text="",
|
|
media_files=["voice.ogg"],
|
|
timestamp=datetime.now(),
|
|
)
|
|
|
|
mock_transcribe = AsyncMock()
|
|
|
|
async def _run():
|
|
with patch("EvoScientist.stt.transcribe_file", mock_transcribe):
|
|
await ch._enqueue_raw(raw)
|
|
|
|
run_async(_run())
|
|
mock_transcribe.assert_not_called()
|
|
assert captured[0].text == ""
|
|
|
|
|
|
def test_enqueue_raw_stt_appends_to_existing_text():
|
|
"""Transcript is prepended before any existing caption text."""
|
|
from EvoScientist.channels.base import RawIncoming
|
|
|
|
ch, captured = _make_channel()
|
|
ch._stt_enabled = True
|
|
ch._stt_language = "auto"
|
|
ch._stt_model = ""
|
|
ch._stt_device = "cpu"
|
|
ch._stt_compute_type = "int8"
|
|
|
|
raw = RawIncoming(
|
|
sender_id="123",
|
|
chat_id="456",
|
|
text="caption text",
|
|
media_files=["voice.ogg"],
|
|
timestamp=datetime.now(),
|
|
)
|
|
|
|
async def _run():
|
|
with (
|
|
patch(
|
|
"EvoScientist.stt.transcribe_file",
|
|
new=AsyncMock(return_value="hello world"),
|
|
),
|
|
patch("EvoScientist.stt.is_audio_file", return_value=True),
|
|
):
|
|
await ch._enqueue_raw(raw)
|
|
|
|
run_async(_run())
|
|
assert captured[0].text == "hello world\ncaption text"
|
|
|
|
|
|
# ── Manual integration test (run by hand) ────────────────────────────
|
|
#
|
|
# 1. Install deps:
|
|
# uv pip install 'EvoScientist[stt]'
|
|
#
|
|
# 2. Enable STT:
|
|
# EvoSci config set stt_enabled true
|
|
# EvoSci config set stt_language zh # zh / en / auto
|
|
#
|
|
# 3. Transcribe a local audio file directly:
|
|
# python -c "
|
|
# import asyncio
|
|
# from EvoScientist.stt import transcribe_file
|
|
# print(asyncio.run(transcribe_file('sample.ogg', language='zh')))
|
|
# "
|
|
#
|
|
# 4. End-to-end via Telegram:
|
|
# EvoSci serve
|
|
# → send a voice message → bot receives transcribed text
|