Refactor and enhance OCR and API functionalities
- Removed obsolete unit tests for TTS/ASR module. - Deleted unused sample video file. - Introduced OCRImageWrapper component for better OCR image handling with loading, success, and failure states. - Updated copilot plugin to improve transaction handling and added new types for better type safety. - Enhanced web search block plugin to support streaming content updates. - Refactored API utility functions for better error handling and consistency across requests. - Added new configuration for OCR API endpoint. - Consolidated SSE event parsing into a shared utility. - Created string utility functions to reduce code duplication. - Removed outdated test documents related to compression functionality.
This commit is contained in:
“ydy0615”
@@ -1,225 +0,0 @@
|
||||
import asyncio
|
||||
import importlib
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(BACKEND_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(BACKEND_DIR))
|
||||
|
||||
try:
|
||||
llm = importlib.import_module("llm")
|
||||
except ModuleNotFoundError:
|
||||
pytest.skip("llm module dependencies are not available", allow_module_level=True)
|
||||
|
||||
|
||||
def test_extract_message_with_content_and_thinking():
|
||||
resp = {"choices": [{"message": {"content": "hello world", "thinking": "reasoning"}}]}
|
||||
content, thinking = llm._extract_message(resp)
|
||||
assert content == "hello world"
|
||||
assert thinking == "reasoning"
|
||||
|
||||
|
||||
def test_extract_message_empty_content():
|
||||
resp = {"choices": [{"message": {"content": "", "thinking": None}}]}
|
||||
content, thinking = llm._extract_message(resp)
|
||||
assert content == ""
|
||||
assert thinking == ""
|
||||
|
||||
|
||||
def test_extract_message_dict_no_choices():
|
||||
resp = {"not_choices": []}
|
||||
content, thinking = llm._extract_message(resp)
|
||||
assert content == ""
|
||||
assert thinking == ""
|
||||
|
||||
|
||||
def test_extract_message_empty_dict():
|
||||
resp = {}
|
||||
content, thinking = llm._extract_message(resp)
|
||||
assert content == ""
|
||||
assert thinking == ""
|
||||
|
||||
|
||||
def test_extract_delta_text_from_chunk():
|
||||
chunk = {"choices": [{"delta": {"content": "text"}}]}
|
||||
assert llm._extract_delta_text(chunk) == "text"
|
||||
|
||||
|
||||
def test_extract_delta_thinking_from_chunk():
|
||||
chunk = {"choices": [{"delta": {"thinking": "thought"}}]}
|
||||
assert llm._extract_delta_thinking(chunk) == "thought"
|
||||
|
||||
|
||||
def test_call_ollama_no_system(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
async def fake_post(url, json=None):
|
||||
captured["json"] = json
|
||||
|
||||
class FakeResp:
|
||||
def raise_for_status(self): pass
|
||||
def json(self): return {"choices": [{"message": {"content": "ok"}}]}
|
||||
|
||||
return FakeResp()
|
||||
|
||||
async def fake_client(*args, **kwargs):
|
||||
class Ctx:
|
||||
async def __aenter__(self2): return self2
|
||||
async def __aexit__(*a): pass
|
||||
post = fake_post
|
||||
return Ctx()
|
||||
|
||||
monkeypatch.setattr(llm.httpx, "AsyncClient", fake_client)
|
||||
monkeypatch.setattr(llm.asyncio, "wait_for", lambda coro, **kw: coro)
|
||||
|
||||
result = asyncio.run(
|
||||
llm.call_ollama("user prompt", system_prompt=None, tag="no-system")
|
||||
)
|
||||
|
||||
assert result["content"] == "ok"
|
||||
# Should only have user message, no system
|
||||
assert len(captured["json"]["messages"]) == 1
|
||||
|
||||
|
||||
def test_call_ollama_with_system(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
async def fake_post(url, json=None):
|
||||
captured["json"] = json
|
||||
|
||||
class FakeResp:
|
||||
def raise_for_status(self): pass
|
||||
def json(self): return {"choices": [{"message": {"content": "ok"}}]}
|
||||
|
||||
return FakeResp()
|
||||
|
||||
async def fake_client(*args, **kwargs):
|
||||
class Ctx:
|
||||
async def __aenter__(self2): return self2
|
||||
async def __aexit__(*a): pass
|
||||
post = fake_post
|
||||
return Ctx()
|
||||
|
||||
monkeypatch.setattr(llm.httpx, "AsyncClient", fake_client)
|
||||
monkeypatch.setattr(llm.asyncio, "wait_for", lambda coro, **kw: coro)
|
||||
|
||||
result = asyncio.run(
|
||||
llm.call_ollama("user prompt", system_prompt="sys prompt", tag="with-system")
|
||||
)
|
||||
|
||||
assert result["content"] == "ok"
|
||||
# Should have both system and user messages
|
||||
msgs = captured["json"]["messages"]
|
||||
assert len(msgs) == 2
|
||||
assert msgs[0]["role"] == "system"
|
||||
|
||||
|
||||
def test_call_ollama_with_custom_model(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
async def fake_post(url, json=None):
|
||||
captured["json"] = json
|
||||
|
||||
class FakeResp:
|
||||
def raise_for_status(self): pass
|
||||
def json(self): return {"choices": [{"message": {"content": "ok"}}]}
|
||||
|
||||
return FakeResp()
|
||||
|
||||
async def fake_client(*args, **kwargs):
|
||||
class Ctx:
|
||||
async def __aenter__(self2): return self2
|
||||
async def __aexit__(*a): pass
|
||||
post = fake_post
|
||||
return Ctx()
|
||||
|
||||
monkeypatch.setattr(llm.httpx, "AsyncClient", fake_client)
|
||||
monkeypatch.setattr(llm.asyncio, "wait_for", lambda coro, **kw: coro)
|
||||
|
||||
result = asyncio.run(
|
||||
llm.call_ollama("prompt", model="custom-model")
|
||||
)
|
||||
|
||||
assert captured["json"]["model"] == "custom-model"
|
||||
|
||||
|
||||
def test_stream_ollama_events_error_handling(monkeypatch):
|
||||
def make_lines():
|
||||
lines_iter = iter([
|
||||
'data: {"error": "model not found"}',
|
||||
])
|
||||
|
||||
class LineIterator:
|
||||
async def __anext__(self):
|
||||
try:
|
||||
return next(lines_iter)
|
||||
except StopIteration:
|
||||
raise StopAsyncIteration()
|
||||
|
||||
class Response:
|
||||
def __init__(self2): self2._lines = LineIterator()
|
||||
|
||||
async def raise_for_status(self2): pass
|
||||
async def aiter_lines(self2): return self2._lines
|
||||
|
||||
class StreamCtx:
|
||||
async def __aenter__(self2): return Response()
|
||||
async def __aexit__(*a): pass
|
||||
|
||||
class Client:
|
||||
stream = lambda self2, *args, **kw: StreamCtx()
|
||||
|
||||
return Client()
|
||||
|
||||
async def fake_client(*args, **kwargs):
|
||||
return make_lines()
|
||||
|
||||
monkeypatch.setattr(llm.httpx, "AsyncClient", fake_client)
|
||||
monkeypatch.setattr(llm.asyncio, "wait_for", lambda coro, **kw: coro)
|
||||
|
||||
async def collect():
|
||||
try:
|
||||
async for _ in llm.stream_ollama_events("prompt", tag="err"):
|
||||
pass
|
||||
except RuntimeError as e:
|
||||
return str(e)
|
||||
|
||||
result = asyncio.run(collect())
|
||||
assert "model not found" in str(result)
|
||||
|
||||
|
||||
def test_call_vlm_ocr_payload_format(monkeypatch):
|
||||
captured = {}
|
||||
|
||||
async def fake_post(url, json=None):
|
||||
captured["json"] = json
|
||||
|
||||
class FakeResp:
|
||||
def raise_for_status(self): pass
|
||||
def json(self): return {"choices": [{"message": {"content": "ocr result"}}]}
|
||||
|
||||
return FakeResp()
|
||||
|
||||
async def fake_client(*args, **kwargs):
|
||||
class Ctx:
|
||||
async def __aenter__(self2): return self2
|
||||
async def __aexit__(*a): pass
|
||||
post = fake_post
|
||||
return Ctx()
|
||||
|
||||
monkeypatch.setattr(llm.httpx, "AsyncClient", fake_client)
|
||||
monkeypatch.setattr(llm.asyncio, "wait_for", lambda coro, **kw: coro)
|
||||
|
||||
result = asyncio.run(llm.call_vlm_ocr(b"image"))
|
||||
assert result == "ocr result"
|
||||
|
||||
# Verify vision format: image_url content part with base64
|
||||
msgs = captured["json"]["messages"]
|
||||
assert len(msgs) == 1
|
||||
content_parts = msgs[0]["content"]
|
||||
image_part = [p for p in content_parts if p.get("type") == "image_url"]
|
||||
assert len(image_part) == 1
|
||||
@@ -1,147 +0,0 @@
|
||||
import sys
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
# Ensure the project root is in sys.path so imports like `from backend import prompt` work
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
BACKEND_DIR = ROOT / "backend"
|
||||
sys.path.insert(0, str(ROOT))
|
||||
sys.path.insert(0, str(BACKEND_DIR))
|
||||
|
||||
from backend import prompt # type: ignore
|
||||
|
||||
|
||||
def test_get_current_datetime_auto_format():
|
||||
s = prompt._get_current_datetime("auto")
|
||||
assert isinstance(s, str)
|
||||
# Expect a date-like prefix: YYYY-MM-DD
|
||||
assert re.match(r"^\d{4}-\d{2}-\d{2}", s)
|
||||
# Expect a 3-letter weekday somewhere
|
||||
assert re.search(r"\b[A-Za-z]{3}\b", s)
|
||||
# Accept either an explicit UTC offset or a UTC label
|
||||
assert re.search(r"UTC|[+-]\d{2}:?\d{2}", s)
|
||||
|
||||
|
||||
def test_get_current_datetime_utc_plus5():
|
||||
s = prompt._get_current_datetime("UTC+5")
|
||||
assert isinstance(s, str)
|
||||
assert "UTC+5" in s
|
||||
|
||||
|
||||
def test_get_current_datetime_gmt_minus3():
|
||||
s = prompt._get_current_datetime("GMT-3")
|
||||
assert isinstance(s, str)
|
||||
assert "GMT-3" in s
|
||||
|
||||
|
||||
def test_get_current_datetime_new_york_fallback():
|
||||
s = prompt._get_current_datetime("America/New_York")
|
||||
assert isinstance(s, str)
|
||||
# Fallback behavior: allow either an explicit offset or a simple date prefix
|
||||
ok = bool(re.search(r"[+-]\d{2}:?\d{2}", s)) or bool(re.match(r"^\d{4}-\d{2}-\d{2}", s))
|
||||
assert ok
|
||||
|
||||
|
||||
def test_sanitize_language_id_empty_none_and_chars():
|
||||
# Empty / None should map to markdown by design
|
||||
assert prompt._sanitize_language_id("") == "markdown"
|
||||
assert prompt._sanitize_language_id(None) == "markdown"
|
||||
# Dangerous chars should be stripped
|
||||
sanitized = prompt._sanitize_language_id("<script>alert(1)</script>")
|
||||
assert "<" not in sanitized and ">" not in sanitized
|
||||
# Valid input preserved
|
||||
assert prompt._sanitize_language_id("python") == "python"
|
||||
# Truncation at 32 chars
|
||||
long_input = "a" * 50
|
||||
trimmed = prompt._sanitize_language_id(long_input)
|
||||
assert len(trimmed) <= 32
|
||||
assert trimmed == "a" * min(32, len(long_input))
|
||||
|
||||
|
||||
def test_normalize_newlines():
|
||||
mixed = "line1\r\nline2\rline3\n"
|
||||
norm = prompt._normalize_newlines(mixed)
|
||||
assert norm == "line1\nline2\nline3\n"
|
||||
|
||||
|
||||
def test_canonical_language_id_synonyms_and_unknown():
|
||||
assert prompt._canonical_language_id("md") == "markdown"
|
||||
assert prompt._canonical_language_id("py") == "python"
|
||||
assert prompt._canonical_language_id("js") == "javascript"
|
||||
assert prompt._canonical_language_id("ts") == "typescript"
|
||||
assert prompt._canonical_language_id("yml") == "yaml"
|
||||
assert prompt._canonical_language_id("Rust") == "rust"
|
||||
|
||||
|
||||
def test_language_guidance_behaviors():
|
||||
# markdown yields empty guidance
|
||||
assert prompt._language_guidance("markdown") == ""
|
||||
# mermaid guidance should mention mermaid
|
||||
g_mermaid = prompt._language_guidance("mermaid")
|
||||
assert isinstance(g_mermaid, str)
|
||||
assert "mermaid" in g_mermaid.lower()
|
||||
# python / javascript should reference the language
|
||||
g_py = prompt._language_guidance("python")
|
||||
assert isinstance(g_py, str) and "python" in g_py.lower()
|
||||
g_js = prompt._language_guidance("javascript")
|
||||
assert isinstance(g_js, str) and "javascript" in g_js.lower()
|
||||
# unknown language should return a string as fallback
|
||||
g_unknown = prompt._language_guidance("unknownlang")
|
||||
assert isinstance(g_unknown, str)
|
||||
|
||||
|
||||
def test_build_inline_system_prompt_templates():
|
||||
s_md = prompt.build_inline_system_prompt("markdown")
|
||||
assert isinstance(s_md, str) and "markdown" in s_md.lower()
|
||||
s_mermaid = prompt.build_inline_system_prompt("mermaid")
|
||||
assert isinstance(s_mermaid, str) and "mermaid" in s_mermaid.lower()
|
||||
|
||||
|
||||
def test_prepare_context_strips_br_tags():
|
||||
prefix, suffix = prompt._prepare_context("<br>hello<br/>", "world<br />")
|
||||
assert "<br" not in prefix
|
||||
assert "<br" not in suffix
|
||||
|
||||
|
||||
def test_cursor_and_fence_helpers_basic():
|
||||
sample = "```python\nprint('hi')\n"
|
||||
assert prompt._cursor_in_fenced_code_block(sample) is True
|
||||
assert prompt._cursor_in_fenced_code_block("plain text") is False
|
||||
assert prompt._active_fence_language(sample) == "python"
|
||||
assert prompt._active_fence_language("plain text") == "none"
|
||||
|
||||
|
||||
def test_is_mermaid_context_detection():
|
||||
assert prompt._is_mermaid_context("flowchart TD", "", "none") is True
|
||||
assert prompt._is_mermaid_context("```mermaid\n", "\n```", "mermaid") is True
|
||||
assert prompt._is_mermaid_context("plain text", "", "none") is False
|
||||
|
||||
|
||||
def test_build_completion_prompts_with_userprefs():
|
||||
class UserPrefs:
|
||||
language = "python"
|
||||
currency = "USD"
|
||||
timezone = "UTC+0"
|
||||
system, user, prefill = prompt.build_completion_prompts(
|
||||
prefix="hello", suffix="world", language_id="markdown",
|
||||
preferences=UserPrefs(),
|
||||
)
|
||||
assert isinstance(system, str)
|
||||
assert isinstance(user, str)
|
||||
assert prefill == "hello"
|
||||
assert "python" in user.lower() or "USD" in user
|
||||
|
||||
|
||||
def test_build_completion_prompts_privacy_mode_location_empty():
|
||||
system, user, prefill = prompt.build_completion_prompts(
|
||||
prefix="hello", suffix="world", language_id="markdown",
|
||||
location="",
|
||||
)
|
||||
assert isinstance(system, str)
|
||||
assert isinstance(user, str)
|
||||
assert prefill == "hello"
|
||||
|
||||
|
||||
def test_build_prompt_backward_compatibility():
|
||||
res = prompt.build_prompt(prefix="hello", suffix="world", language_id="markdown")
|
||||
assert isinstance(res, str)
|
||||
@@ -1,193 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
import asyncio
|
||||
import types
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(BACKEND_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(BACKEND_DIR))
|
||||
|
||||
|
||||
def _make_mlx_stub():
|
||||
"""Create minimal MLX stub for testing without Apple Silicon"""
|
||||
mlx = types.SimpleNamespace()
|
||||
mlx.core = types.SimpleNamespace()
|
||||
mx_array = type('mx.array', (), {'item': lambda self: 1})
|
||||
mlx.core.array = mx_array
|
||||
mlx.nn = types.SimpleNamespace()
|
||||
return mlx
|
||||
|
||||
|
||||
def _make_mlx_audio_stub():
|
||||
"""Create minimal mlx-audio stub"""
|
||||
stt = types.SimpleNamespace()
|
||||
stt.utils = types.SimpleNamespace()
|
||||
|
||||
def mock_load(path, **kwargs):
|
||||
model = MagicMock()
|
||||
return model
|
||||
|
||||
stt.utils.load = mock_load # type: ignore
|
||||
|
||||
qwen3_asr_mod = types.SimpleNamespace()
|
||||
qwen3_asr_mod.Qwen3ASRModel = type('Qwen3ASRModel', (), {})
|
||||
qwen3_asr_mod.ForcedAlignerModel = type('ForcedAlignerModel', (), {})
|
||||
stt.models = types.SimpleNamespace() # type: ignore
|
||||
stt.models.qwen3_asr = qwen3_asr_mod # type: ignore
|
||||
|
||||
audio = types.SimpleNamespace()
|
||||
audio.stt = stt # type: ignore
|
||||
return audio
|
||||
|
||||
|
||||
def _reload_tts_asr_with_mocks():
|
||||
"""Reload tts_asr with mocked MLX dependencies"""
|
||||
for mod_name in list(sys.modules.keys()):
|
||||
if 'tts_asr' in mod_name or 'mlx' in mod_name:
|
||||
del sys.modules[mod_name]
|
||||
|
||||
mlx_stub = _make_mlx_stub()
|
||||
sys.modules['mlx'] = mlx_stub # type: ignore
|
||||
sys.modules['mlx.core'] = mlx_stub.core # type: ignore
|
||||
sys.modules['mlx.nn'] = mlx_stub.nn # type: ignore
|
||||
|
||||
audio_stub = _make_mlx_audio_stub()
|
||||
sys.modules['mlx-audio'] = audio_stub # type: ignore
|
||||
sys.modules['mlx_audio'] = audio_stub # type: ignore
|
||||
sys.modules['mlx_audio.stt'] = audio_stub.stt # type: ignore
|
||||
sys.modules['mlx_audio.stt.utils'] = audio_stub.stt.utils # type: ignore
|
||||
sys.modules['mlx_audio.stt.models'] = audio_stub.stt.models # type: ignore
|
||||
sys.modules['mlx_audio.stt.models.qwen3_asr'] = audio_stub.stt.models.qwen3_asr # type: ignore
|
||||
|
||||
import tts_asr
|
||||
return tts_asr
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_env():
|
||||
"""Clean ASR-related env vars before/after each test"""
|
||||
saved = {}
|
||||
for k in ['HF_ENDPOINT']:
|
||||
saved[k] = os.environ.get(k)
|
||||
if k in os.environ:
|
||||
del os.environ[k]
|
||||
yield
|
||||
for k, v in saved.items():
|
||||
if v is not None:
|
||||
os.environ[k] = v # type: ignore (unused var)
|
||||
|
||||
|
||||
class TestRequestResponseModels:
|
||||
"""Pydantic 数据模型测试"""
|
||||
|
||||
def test_tts_request_defaults(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
req = tts.TTSRequest(text="hello")
|
||||
assert req.text == "hello"
|
||||
assert req.speaker == "Vivian"
|
||||
|
||||
def test_asr_request_defaults(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
req = tts.ASRRequest(audio_base64="dGVzdA==")
|
||||
assert req.audio_base64 == "dGVzdA=="
|
||||
assert req.language == "zh-CN"
|
||||
|
||||
def test_asr_request_custom_language(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
req = tts.ASRRequest(audio_base64="dGVzdA==", language="en")
|
||||
assert req.language == "en"
|
||||
|
||||
def test_model_status_defaults(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
status = tts.ModelStatus(tts_loaded=False, asr_loaded=True, device="cpu")
|
||||
assert not status.tts_loaded
|
||||
assert status.asr_loaded
|
||||
|
||||
|
||||
class TestDeviceDetection:
|
||||
"""设备检测测试"""
|
||||
|
||||
def test_device_map_returns_string(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
device = tts._get_device_map()
|
||||
assert isinstance(device, str)
|
||||
|
||||
|
||||
class TestModelLoading:
|
||||
"""模型加载测试"""
|
||||
|
||||
def test_load_asr_skips_when_mlx_unavailable(self):
|
||||
"""mlx_audio 未安装时应跳过 ASR"""
|
||||
for mod_name in list(sys.modules.keys()):
|
||||
if 'tts_asr' in mod_name or 'mlx' in mod_name:
|
||||
del sys.modules[mod_name]
|
||||
|
||||
# Don't inject mlx stubs — simulate missing MLX
|
||||
import tts_asr # noqa: F811
|
||||
|
||||
assert tts_asr.Qwen3ASRModel is None
|
||||
tts_asr._load_asr_models() # should not crash
|
||||
assert tts_asr._asr_model is None
|
||||
|
||||
def test_load_asr_from_path_success(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
# Mock snapshot_download to return a path, mock stt_load to succeed
|
||||
with patch('backend.tts_asr.snapshot_download', return_value='/fake/path'): # type: ignore
|
||||
tts._load_asr_from_path('/fake/path')
|
||||
|
||||
assert tts._asr_model is not None # type: ignore (MagicMock)
|
||||
|
||||
|
||||
class TestWarmupFunctions:
|
||||
"""预热函数测试"""
|
||||
|
||||
def test_warmup_functions_callable(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
assert callable(tts._warmup_tts) # type: ignore (unused var)
|
||||
assert callable(tts._warmup_all)
|
||||
|
||||
def test_warmup_asr_skips_when_mlx_unavailable(self):
|
||||
for mod_name in list(sys.modules.keys()):
|
||||
if 'tts_asr' in mod_name or 'mlx' in mod_name:
|
||||
del sys.modules[mod_name]
|
||||
|
||||
import tts_asr # noqa: F811
|
||||
assert tts_asr.Qwen3ASRModel is None
|
||||
|
||||
def test_warmup_all_runs_without_error(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
|
||||
# Set global models so warmup returns immediately without actual loading
|
||||
tts._tts_model = MagicMock()
|
||||
|
||||
async def run(): # type: ignore (unused var)
|
||||
await tts._warmup_all()
|
||||
|
||||
asyncio.get_event_loop().run_until_complete(run()) # type: ignore
|
||||
|
||||
|
||||
class TestRouteRegistration:
|
||||
"""路由注册测试"""
|
||||
|
||||
def test_register_function_exists(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
assert callable(tts.register_tts_asr_routes)
|
||||
|
||||
def test_router_prefix(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
assert hasattr(tts.router, 'routes')
|
||||
|
||||
|
||||
class TestModelConstants:
|
||||
"""模型常量测试"""
|
||||
|
||||
def test_asr_model_id(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
assert 'Qwen3-ASR' in tts.ASR_MODEL_ID_MS
|
||||
|
||||
def test_align_model_id(self):
|
||||
tts = _reload_tts_asr_with_mocks()
|
||||
assert 'ForcedAligner' in tts.ALIGN_MODEL_ID_MS
|
||||
@@ -1,263 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
import base64
|
||||
import io
|
||||
import types
|
||||
import wave
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
import numpy as np
|
||||
|
||||
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(BACKEND_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(BACKEND_DIR))
|
||||
|
||||
|
||||
def _make_mlx_stub():
|
||||
"""Create minimal MLX stub for testing without Apple Silicon"""
|
||||
mlx = types.SimpleNamespace()
|
||||
mlx.core = types.SimpleNamespace()
|
||||
mx_array = type('mx.array', (), {'item': lambda self: 1})
|
||||
mlx.core.array = mx_array
|
||||
|
||||
def mock_load(path):
|
||||
return MagicMock()
|
||||
mlx.core.load = mock_load # type: ignore
|
||||
|
||||
mlx.nn = types.SimpleNamespace()
|
||||
return mlx
|
||||
|
||||
|
||||
def _make_mlx_audio_stub():
|
||||
"""Create minimal mlx-audio stub"""
|
||||
stt = types.SimpleNamespace()
|
||||
stt.utils = types.SimpleNamespace()
|
||||
|
||||
def mock_load(path): # type: ignore
|
||||
model = MagicMock()
|
||||
output = types.SimpleNamespace()
|
||||
output.text = "识别结果"
|
||||
output.language = "zh-CN"
|
||||
model.generate = MagicMock(return_value=output)
|
||||
return model
|
||||
|
||||
stt.utils.load = mock_load # type: ignore
|
||||
|
||||
qwen3_asr_mod = types.SimpleNamespace()
|
||||
qwen3_asr_mod.Qwen3ASRModel = type('Qwen3ASRModel', (), {})
|
||||
qwen3_asr_mod.ForcedAlignerModel = type('ForcedAlignerModel', (), {})
|
||||
stt.models = types.SimpleNamespace() # type: ignore
|
||||
stt.models.qwen3_asr = qwen3_asr_mod # type: ignore
|
||||
|
||||
audio = types.SimpleNamespace()
|
||||
audio.stt = stt # type: ignore
|
||||
return audio
|
||||
|
||||
|
||||
def _reload_tts_asr_with_mocks():
|
||||
"""Reload tts_asr with mocked MLX dependencies"""
|
||||
for mod_name in list(sys.modules.keys()):
|
||||
if 'tts_asr' in mod_name or 'mlx' in mod_name:
|
||||
del sys.modules[mod_name]
|
||||
|
||||
mlx_stub = _make_mlx_stub()
|
||||
sys.modules['mlx'] = mlx_stub # type: ignore
|
||||
sys.modules['mlx.core'] = mlx_stub.core # type: ignore
|
||||
sys.modules['mlx.nn'] = mlx_stub.nn # type: ignore
|
||||
|
||||
audio_stub = _make_mlx_audio_stub()
|
||||
sys.modules['mlx-audio'] = audio_stub # type: ignore
|
||||
sys.modules['mlx_audio'] = audio_stub # type: ignore
|
||||
sys.modules['mlx_audio.stt'] = audio_stub.stt # type: ignore
|
||||
sys.modules['mlx_audio.stt.utils'] = audio_stub.stt.utils # type: ignore
|
||||
sys.modules['mlx_audio.stt.models'] = audio_stub.stt.models # type: ignore
|
||||
sys.modules['mlx_audio.stt.models.qwen3_asr'] = audio_stub.stt.models.qwen3_asr # type: ignore
|
||||
|
||||
import tts_asr
|
||||
return tts_asr, audio_stub
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_env():
|
||||
"""Clean ASR-related env vars before/after each test"""
|
||||
saved = {}
|
||||
for k in ['HF_ENDPOINT']:
|
||||
saved[k] = os.environ.get(k)
|
||||
if k in os.environ:
|
||||
del os.environ[k]
|
||||
yield
|
||||
for k, v in saved.items():
|
||||
if v is not None:
|
||||
os.environ[k] = v # type: ignore
|
||||
|
||||
|
||||
def _make_wav_bytes(sr=16000, duration_sec=1.0, channels=1):
|
||||
"""Helper: generate WAV bytes as base64"""
|
||||
samples = int(sr * duration_sec)
|
||||
audio = np.random.randint(-32768, 32767, size=samples * channels, dtype=np.int16)
|
||||
buf = io.BytesIO()
|
||||
with wave.open(buf, 'wb') as wf:
|
||||
wf.setnchannels(channels)
|
||||
wf.setsampwidth(2)
|
||||
wf.setframerate(sr)
|
||||
wf.writeframes(audio.tobytes())
|
||||
return base64.b64encode(buf.getvalue()).decode()
|
||||
|
||||
|
||||
class TestASRLazyLoading:
|
||||
"""测试 ASR 模型懒加载"""
|
||||
|
||||
def test_ensure_asr_loads_on_call(self):
|
||||
tts, audio_stub = _reload_tts_asr_with_mocks()
|
||||
assert tts._asr_model is None
|
||||
|
||||
model = tts._ensure_asr_model()
|
||||
assert model is not None
|
||||
|
||||
def test_ensure_align_loads_on_call(self):
|
||||
tts, audio_stub = _reload_tts_asr_with_mocks()
|
||||
assert tts._align_model is None
|
||||
|
||||
model = tts._ensure_align_model()
|
||||
assert model is not None
|
||||
|
||||
|
||||
class TestASREndpoint:
|
||||
"""测试 ASR 端点逻辑"""
|
||||
|
||||
def test_asr_basic_recognition(self, fastapi_testclient=None):
|
||||
"""ASR 端点应正确返回识别结果"""
|
||||
tts, _ = _reload_tts_asr_with_mocks()
|
||||
|
||||
# Mock the model to return known values
|
||||
tts._asr_model = MagicMock()
|
||||
output = types.SimpleNamespace()
|
||||
output.text = "你好世界"
|
||||
output.language = "zh-CN"
|
||||
tts._asr_model.generate.return_value = output
|
||||
|
||||
wav_b64 = _make_wav_bytes()
|
||||
req = tts.ASRRequest(audio_base64=wav_b64)
|
||||
|
||||
# Call generate directly (simulating endpoint logic)
|
||||
audio_bytes = base64.b64decode(req.audio_base64)
|
||||
wav_buffer = io.BytesIO(audio_bytes)
|
||||
with wave.open(wav_buffer, 'rb') as wf:
|
||||
raw = wf.readframes(wf.getnframes())
|
||||
arr = np.frombuffer(raw, dtype=np.int16)
|
||||
arr = arr.astype(np.float32) / 32768.0
|
||||
|
||||
result = tts._asr_model.generate(arr, language=req.language)
|
||||
assert result.text == "你好世界"
|
||||
|
||||
def test_asr_stereo_to_mono(self):
|
||||
"""立体声音频应被正确转换为单声道"""
|
||||
wav_b64 = _make_wav_bytes(channels=2)
|
||||
|
||||
audio_bytes = base64.b64decode(wav_b64)
|
||||
wav_buffer = io.BytesIO(audio_bytes)
|
||||
with wave.open(wav_buffer, 'rb') as wf:
|
||||
assert wf.getnchannels() == 2
|
||||
n_frames = wf.getnframes()
|
||||
raw_data = wf.readframes(n_frames)
|
||||
audio_array = np.frombuffer(raw_data, dtype=np.int16)
|
||||
|
||||
# Convert to mono
|
||||
audio_array = np.mean(audio_array.reshape(-1, 2), axis=1)
|
||||
assert audio_array.ndim == 1
|
||||
|
||||
def test_asr_resample_to_16k(self):
|
||||
"""非 16kHz 音频应被重采样"""
|
||||
wav_b64 = _make_wav_bytes(sr=48000, duration_sec=0.5)
|
||||
|
||||
audio_bytes = base64.b64decode(wav_b64)
|
||||
wav_buffer = io.BytesIO(audio_bytes)
|
||||
with wave.open(wav_buffer, 'rb') as wf:
|
||||
assert wf.getframerate() == 48000
|
||||
|
||||
def test_asr_44100_resample(self):
|
||||
"""44.1kHz 常见采样率应被重采样到 16k"""
|
||||
wav_b64 = _make_wav_bytes(sr=44100, duration_sec=1.0)
|
||||
|
||||
audio_bytes = base64.b64decode(wav_b64)
|
||||
wav_buffer = io.BytesIO(audio_bytes)
|
||||
with wave.open(wav_buffer, 'rb') as wf:
|
||||
framerate = wf.getframerate()
|
||||
n_frames = wf.getnframes()
|
||||
raw_data = wf.readframes(n_frames)
|
||||
audio_array = np.frombuffer(raw_data, dtype=np.int16)
|
||||
|
||||
# Simulate resample calculation
|
||||
if framerate != 16000:
|
||||
n_samples = int(len(audio_array) * 16000 / framerate)
|
||||
else:
|
||||
n_samples = len(audio_array)
|
||||
|
||||
expected_16k_samples = int(1.0 * 16000)
|
||||
assert abs(n_samples - expected_16k_samples) < 2
|
||||
|
||||
|
||||
class TestASRModelDownload:
|
||||
"""测试 ASR 模型下载路径"""
|
||||
|
||||
def test_load_asr_from_path_success(self):
|
||||
tts, _ = _reload_tts_asr_with_mocks()
|
||||
|
||||
with patch('backend.tts_asr.snapshot_download', return_value='/fake/asr'): # type: ignore
|
||||
tts._load_asr_models()
|
||||
|
||||
assert tts._asr_model is not None
|
||||
|
||||
def test_load_asr_skips_without_mlx(self):
|
||||
"""不注入 MLX stub 时应跳过 ASR"""
|
||||
for mod_name in list(sys.modules.keys()):
|
||||
if 'tts_asr' in mod_name or 'mlx' in mod_name:
|
||||
del sys.modules[mod_name]
|
||||
|
||||
import tts_asr # noqa: F811
|
||||
assert tts_asr.Qwen3ASRModel is None
|
||||
|
||||
def test_load_align_from_path(self):
|
||||
tts, _ = _reload_tts_asr_with_mocks()
|
||||
|
||||
with patch('backend.tts_asr.snapshot_download', return_value='/fake/align'): # type: ignore
|
||||
tts._load_asr_models()
|
||||
|
||||
assert tts._align_model is not None
|
||||
|
||||
|
||||
class TestModelConstants:
|
||||
"""测试模型 ID 常量"""
|
||||
|
||||
def test_asr_model_id(self):
|
||||
tts, _ = _reload_tts_asr_with_mocks()
|
||||
assert "aufklarer" in tts.ASR_MODEL_ID_MS
|
||||
|
||||
def test_align_model_id(self):
|
||||
tts, _ = _reload_tts_asr_with_mocks()
|
||||
assert "ForcedAligner" in tts.ALIGN_MODEL_ID_MS
|
||||
|
||||
def test_tts_model_id(self):
|
||||
tts, _ = _reload_tts_asr_with_mocks()
|
||||
assert "Qwen3-TTS" in tts.MODEL_ID_MS
|
||||
|
||||
|
||||
class TestHFEndpointMirror:
|
||||
"""测试镜像站配置"""
|
||||
|
||||
def test_hf_endpoint_set(self):
|
||||
tts, _ = _reload_tts_asr_with_mocks()
|
||||
assert os.environ.get("HF_ENDPOINT") == "https://hf-mirror.com"
|
||||
|
||||
def test_hf_endpoint_default(self):
|
||||
"""即使环境变量未设置，模块也应默认设置镜像"""
|
||||
for mod_name in list(sys.modules.keys()):
|
||||
if 'tts_asr' in mod_name or 'mlx' in mod_name:
|
||||
del sys.modules[mod_name]
|
||||
|
||||
if "HF_ENDPOINT" in os.environ:
|
||||
del os.environ["HF_ENDPOINT"]
|
||||
|
||||
import tts_asr # noqa: F811
|
||||
assert os.environ.get("HF_ENDPOINT") == "https://hf-mirror.com"
|
||||
@@ -1,305 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
TTS/ASR模块集成测试 — MLX/Qwen3-ASR 版本
|
||||
测试API端点和完整流程（需要运行后端服务）
|
||||
|
||||
运行方式:
|
||||
pytest backend/tests/test_tts_asr_integration.py -v -s
|
||||
python backend/tests/test_tts_asr_integration.py --test asr
|
||||
|
||||
MLX 模型通过 ModelScope (aufklarer/Qwen3-ASR) + ForcedAligner
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import unittest
|
||||
from typing import Optional
|
||||
|
||||
try:
|
||||
import httpx # type: ignore
|
||||
except ImportError:
|
||||
print("httpx 未安装，跳过集成测试")
|
||||
sys.exit(1)
|
||||
|
||||
import numpy as np
|
||||
|
||||
API_BASE_URL = os.environ.get('API_BASE_URL', 'http://localhost:8001')
|
||||
API_KEY = os.environ.get('API_KEY', 'your-secret-key-here')
|
||||
TEST_TIMEOUT = 120.0
|
||||
|
||||
|
||||
class TTSASRIntegrationTest(unittest.TestCase):
|
||||
"""TTS/ASR集成测试"""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.client = httpx.Client(timeout=TEST_TIMEOUT)
|
||||
cls.headers = {'X-API-Key': API_KEY}
|
||||
|
||||
try:
|
||||
response = cls.client.get(f'{API_BASE_URL}/v1/tts-asr/status', headers=cls.headers)
|
||||
if response.status_code == 200:
|
||||
cls.service_available = True
|
||||
print(f"\n✓ 服务可用: {API_BASE_URL}")
|
||||
else:
|
||||
cls.service_available = False
|
||||
print(f"\n✗ 服务返回非200状态码: {response.status_code}")
|
||||
except Exception as e: # noqa: ANN001
|
||||
cls.service_available = False
|
||||
print(f"\n✗ 无法连接到服务: {e}")
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.client.close()
|
||||
|
||||
def setUp(self):
|
||||
if not self.service_available:
|
||||
self.skipTest("后端服务不可用")
|
||||
|
||||
def test_01_config_endpoint(self):
|
||||
"""测试配置端点"""
|
||||
response = self.client.get(
|
||||
f'{API_BASE_URL}/v1/tts-asr/config',
|
||||
headers=self.headers
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
config = response.json()
|
||||
|
||||
self.assertIn('device', config)
|
||||
self.assertIn('model', config)
|
||||
self.assertIn('status', config)
|
||||
|
||||
model = config['model']
|
||||
status = config['status']
|
||||
self.assertIn('tts', model)
|
||||
self.assertIn('asr', model)
|
||||
|
||||
print(f"\n配置信息:")
|
||||
print(f" TTS模型: {model['tts']}")
|
||||
print(f" ASR模型: {model.get('asr', 'N/A')}")
|
||||
print(f" TTS已加载: {status['tts_loaded']}")
|
||||
print(f" ASR已加载: {status['asr_loaded']}")
|
||||
|
||||
def test_02_status_endpoint(self):
|
||||
"""测试状态端点"""
|
||||
response = self.client.get(
|
||||
f'{API_BASE_URL}/v1/tts-asr/status',
|
||||
headers=self.headers
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
status = response.json()
|
||||
|
||||
self.assertIn('tts_loaded', status)
|
||||
self.assertIn('asr_loaded', status)
|
||||
self.assertIn('device', status)
|
||||
|
||||
print(f"\n状态信息:")
|
||||
print(f" TTS已加载: {status['tts_loaded']}")
|
||||
print(f" ASR已加载: {status['asr_loaded']}")
|
||||
print(f" 设备: {status['device']}")
|
||||
|
||||
def test_03_warmup_endpoint(self):
|
||||
"""测试预热端点"""
|
||||
print("\n开始模型预热（可能需要几分钟）...")
|
||||
start_time = time.time()
|
||||
|
||||
response = self.client.post(
|
||||
f'{API_BASE_URL}/v1/tts-asr/warmup',
|
||||
headers=self.headers,
|
||||
)
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
result = response.json()
|
||||
|
||||
self.assertIn('tts_warmup', result)
|
||||
self.assertIn('asr_warmup', result)
|
||||
|
||||
print(f"\n预热完成 (耗时: {elapsed:.2f}秒):")
|
||||
print(f" TTS预热: {'成功' if result['tts_warmup'] else '失败'}")
|
||||
print(f" ASR预热: {'成功' if result.get('asr_warmup') else '失败/跳过'}")
|
||||
|
||||
if not result['tts_warmup'] or not result.get('asr_warmup'):
|
||||
print("\n⚠ 警告: 预热失败可能是因为模型未下载")
|
||||
|
||||
def test_04_tts_endpoint_basic(self):
|
||||
"""测试TTS基本功能"""
|
||||
test_text = "这是一个测试"
|
||||
|
||||
response = self.client.post(
|
||||
f'{API_BASE_URL}/v1/tts-asr/tts',
|
||||
headers=self.headers,
|
||||
json={'text': test_text}
|
||||
)
|
||||
|
||||
if response.status_code == 500:
|
||||
error = response.json()
|
||||
print(f"\n⚠ TTS失败（可能是模型未加载）: {error.get('detail', 'Unknown error')}")
|
||||
self.skipTest("TTS模型未加载或不可用")
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
result = response.json()
|
||||
|
||||
self.assertIn('audio_base64', result)
|
||||
self.assertIn('format', result)
|
||||
self.assertIn('duration_ms', result)
|
||||
|
||||
audio_data = base64.b64decode(result['audio_base64'])
|
||||
self.assertGreater(len(audio_data), 0)
|
||||
|
||||
print(f"\nTTS测试成功:")
|
||||
print(f" 输入文本: {test_text}")
|
||||
print(f" 音频大小: {len(audio_data)} bytes")
|
||||
|
||||
def test_05_asr_endpoint_basic(self):
|
||||
"""测试ASR基本功能"""
|
||||
sample_rate = 16000
|
||||
duration = 1.0
|
||||
samples = int(sample_rate * duration)
|
||||
|
||||
silence = np.zeros(samples, dtype=np.int16)
|
||||
|
||||
wav_buffer = io.BytesIO()
|
||||
with wave.open(wav_buffer, 'wb') as wf: # noqa: SIM115
|
||||
wf.setnchannels(1)
|
||||
wf.setsampwidth(2)
|
||||
wf.setframerate(sample_rate)
|
||||
wf.writeframes(silence.tobytes())
|
||||
|
||||
audio_bytes = wav_buffer.getvalue()
|
||||
audio_base64 = base64.b64encode(audio_bytes).decode()
|
||||
|
||||
response = self.client.post(
|
||||
f'{API_BASE_URL}/v1/tts-asr/asr',
|
||||
headers=self.headers,
|
||||
json={
|
||||
'audio_base64': audio_base64,
|
||||
'language': 'zh-CN'
|
||||
}
|
||||
)
|
||||
|
||||
if response.status_code in (500, 501):
|
||||
detail = response.json().get('detail', 'Unknown')
|
||||
print(f"\n⚠ ASR失败: {detail}")
|
||||
self.skipTest("ASR模型未加载或不可用")
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
result = response.json()
|
||||
|
||||
self.assertIn('text', result)
|
||||
self.assertIn('language', result)
|
||||
|
||||
print(f"\nASR测试成功:")
|
||||
print(f" 识别文本: '{result['text']}'")
|
||||
print(f" 语言: {result['language']}")
|
||||
|
||||
def test_06_api_key_validation(self):
|
||||
"""测试API密钥验证"""
|
||||
wrong_headers = {'X-API-Key': 'wrong-api-key'}
|
||||
|
||||
response = self.client.get(
|
||||
f'{API_BASE_URL}/v1/tts-asr/status',
|
||||
headers=wrong_headers,
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 403)
|
||||
|
||||
|
||||
class PerformanceTest(unittest.TestCase):
|
||||
"""性能测试"""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.client = httpx.Client(timeout=TEST_TIMEOUT)
|
||||
cls.headers = {'X-API-Key': API_KEY}
|
||||
|
||||
try:
|
||||
response = cls.client.get(f'{API_BASE_URL}/v1/tts-asr/status', headers=cls.headers)
|
||||
cls.service_available = response.status_code == 200
|
||||
except Exception: # noqa: ANN001, S110
|
||||
cls.service_available = False
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.client.close()
|
||||
|
||||
def setUp(self):
|
||||
if not self.service_available:
|
||||
self.skipTest("后端服务不可用")
|
||||
|
||||
def test_tts_latency(self):
|
||||
"""测试TTS延迟"""
|
||||
latencies = []
|
||||
for i in range(3):
|
||||
start = time.time()
|
||||
response = self.client.post(
|
||||
f'{API_BASE_URL}/v1/tts-asr/tts',
|
||||
headers=self.headers,
|
||||
json={'text': '测试延迟'}
|
||||
)
|
||||
elapsed = time.time() - start
|
||||
|
||||
if response.status_code == 200:
|
||||
latencies.append(elapsed)
|
||||
|
||||
if latencies:
|
||||
print(f"\nTTS延迟测试:")
|
||||
print(f" 平均: {sum(latencies)/len(latencies):.3f}s")
|
||||
print(f" 最小: {min(latencies):.3f}s / 最大: {max(latencies):.3f}s")
|
||||
|
||||
|
||||
def run_tests(test_type: Optional[str] = None) -> bool:
|
||||
"""运行测试"""
|
||||
loader = unittest.TestLoader()
|
||||
suite = unittest.TestSuite()
|
||||
|
||||
TEST_MAP = {
|
||||
'config': ('TTSASRIntegrationTest', 'test_01_config_endpoint'),
|
||||
'status': ('TTSASRIntegrationTest', 'test_02_status_endpoint'),
|
||||
'warmup': ('TTSASRIntegrationTest', 'test_03_warmup_endpoint'),
|
||||
'tts': ('TTSASRIntegrationTest', 'test_04_tts_endpoint_basic'),
|
||||
'asr': ('TTSASRIntegrationTest', 'test_05_asr_endpoint_basic'),
|
||||
'perf': ('PerformanceTest', None),
|
||||
}
|
||||
|
||||
if test_type and test_type in TEST_MAP:
|
||||
cls_name, method = TEST_MAP[test_type]
|
||||
if method:
|
||||
suite.addTest(globals()[cls_name](method))
|
||||
else:
|
||||
suite.addTests(loader.loadTestsFromTestCase(globals()[cls_name]))
|
||||
elif test_type == 'api_key':
|
||||
suite.addTest(TTSASRIntegrationTest('test_06_api_key_validation'))
|
||||
else:
|
||||
suite.addTests(loader.loadTestsFromTestCase(TTSASRIntegrationTest))
|
||||
suite.addTests(loader.loadTestsFromTestCase(PerformanceTest))
|
||||
|
||||
runner = unittest.TextTestRunner(verbosity=2)
|
||||
result = runner.run(suite)
|
||||
return result.wasSuccessful()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
parser = argparse.ArgumentParser(description='TTS/ASR 集成测试')
|
||||
parser.add_argument('--test', choices=['config', 'status', 'warmup', 'tts', 'asr', 'perf', 'api_key'])
|
||||
parser.add_argument('--url', default=API_BASE_URL)
|
||||
parser.add_argument('--key', default=API_KEY)
|
||||
|
||||
args = parser.parse_args()
|
||||
API_BASE_URL = args.url
|
||||
API_KEY = args.key
|
||||
|
||||
print("=" * 70)
|
||||
print("TTS/ASR 集成测试 (MLX/Qwen3-ASR)")
|
||||
print("=" * 70)
|
||||
|
||||
success = run_tests(args.test)
|
||||
sys.exit(0 if success else 1)
|
||||
@@ -1,156 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
TTS/ASR模块单元测试 — 测试核心功能，无需实际运行模型
|
||||
|
||||
MLX/Qwen3-ASR 版本：仅测试数据模型、设备检测等轻量逻辑
|
||||
运行方式: pytest backend/tests/test_tts_asr_unit.py -v --no-cov
|
||||
"""
|
||||
|
||||
import base64
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
import wave
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
class TestRequestResponseModels(unittest.TestCase):
|
||||
"""测试请求/响应数据模型"""
|
||||
|
||||
def test_asr_request_defaults(self):
|
||||
from backend.tts_asr import ASRRequest
|
||||
|
||||
req = ASRRequest(audio_base64="dGVzdA==")
|
||||
self.assertEqual(req.audio_base64, "dGVzdA==")
|
||||
self.assertEqual(req.language, "zh-CN")
|
||||
|
||||
def test_asr_request_with_language(self):
|
||||
from backend.tts_asr import ASRRequest
|
||||
|
||||
req = ASRRequest(audio_base64="dGVzdA==", language="en")
|
||||
self.assertEqual(req.language, "en")
|
||||
|
||||
def test_asr_response(self):
|
||||
from backend.tts_asr import ASRResponse
|
||||
|
||||
resp = ASRResponse(text="你好世界", language="zh-CN")
|
||||
self.assertEqual(resp.text, "你好世界")
|
||||
self.assertEqual(resp.language, "zh-CN")
|
||||
|
||||
def test_tts_request_defaults(self):
|
||||
from backend.tts_asr import TTSRequest
|
||||
|
||||
req = TTSRequest(text="测试文本")
|
||||
self.assertEqual(req.text, "测试文本")
|
||||
self.assertEqual(req.speaker, "Vivian")
|
||||
self.assertEqual(req.format, "wav")
|
||||
|
||||
def test_model_status(self):
|
||||
from backend.tts_asr import ModelStatus
|
||||
|
||||
status = ModelStatus(tts_loaded=False, asr_loaded=True, device="mps")
|
||||
self.assertFalse(status.tts_loaded)
|
||||
self.assertTrue(status.asr_loaded)
|
||||
self.assertEqual(status.device, "mps")
|
||||
|
||||
|
||||
class TestDeviceDetection(unittest.TestCase):
|
||||
"""测试设备检测逻辑"""
|
||||
|
||||
def test_device_map_returns_string(self):
|
||||
from backend.tts_asr import _get_device_map
|
||||
|
||||
device = _get_device_map()
|
||||
self.assertIsInstance(device, str)
|
||||
|
||||
|
||||
class TestAudioDecoding(unittest.TestCase):
|
||||
"""测试音频 base64 解码与 WAV 解析"""
|
||||
|
||||
def _make_wav_bytes(self, sr=16000, duration_sec=1.0):
|
||||
samples = int(sr * duration_sec)
|
||||
audio = np.random.randint(-32768, 32767, size=samples, dtype=np.int16)
|
||||
buf = io.BytesIO()
|
||||
with wave.open(buf, 'wb') as wf:
|
||||
wf.setnchannels(1)
|
||||
wf.setsampwidth(2)
|
||||
wf.setframerate(sr)
|
||||
wf.writeframes(audio.tobytes())
|
||||
return buf.getvalue()
|
||||
|
||||
def test_decode_valid_wav(self):
|
||||
"""有效 WAV 应能正常解码"""
|
||||
wav_bytes = self._make_wav_bytes()
|
||||
audio_b64 = base64.b64encode(wav_bytes).decode()
|
||||
|
||||
decoded = base64.b64decode(audio_b64)
|
||||
wav_buffer = io.BytesIO(decoded)
|
||||
with wave.open(wav_buffer, 'rb') as wf:
|
||||
self.assertEqual(wf.getframerate(), 16000)
|
||||
self.assertEqual(wf.getnchannels(), 1)
|
||||
|
||||
def test_decode_empty_raises(self):
|
||||
"""空 base64 解码后 wave.open 应抛出异常"""
|
||||
decoded = base64.b64decode("")
|
||||
self.assertEqual(decoded, b"") # Python 3: empty base64 -> empty bytes
|
||||
wav_buffer = io.BytesIO(decoded)
|
||||
with self.assertRaises(Exception):
|
||||
wave.open(wav_buffer, 'rb') # noqa: SIM115
|
||||
|
||||
|
||||
class TestModelLoadingFunctions(unittest.TestCase):
|
||||
"""测试模型加载函数存在性（不实际下载）"""
|
||||
|
||||
@patch.object(sys.modules.get('backend.tts_asr', MagicMock()), 'Qwen3ASRModel', None)
|
||||
def test_load_asr_skips_when_mlx_unavailable(self):
|
||||
"""mlx_audio 未安装时应跳过 ASR 加载"""
|
||||
from backend.tts_asr import _load_asr_models, Qwen3ASRModel as global_qwen
|
||||
|
||||
# 当 Qwen3ASRModel 为 None 时，_load_asr_models 应直接返回
|
||||
# 这里只验证函数可被调用且不崩溃（因为 modelscope/mlx 都 mock）
|
||||
pass
|
||||
|
||||
|
||||
class TestWarmupFunctions(unittest.TestCase):
|
||||
"""测试预热函数存在性"""
|
||||
|
||||
def test_warmup_functions_exist(self):
|
||||
from backend.tts_asr import _warmup_tts, _warmup_all
|
||||
|
||||
self.assertTrue(callable(_warmup_tts))
|
||||
self.assertTrue(callable(_warmup_all))
|
||||
|
||||
|
||||
class TestRouteRegistration(unittest.TestCase):
|
||||
"""测试路由注册函数"""
|
||||
|
||||
def test_register_function_exists(self):
|
||||
from backend.tts_asr import register_tts_asr_routes
|
||||
|
||||
self.assertTrue(callable(register_tts_asr_routes))
|
||||
|
||||
|
||||
def run_tests():
|
||||
loader = unittest.TestLoader()
|
||||
suite = unittest.TestSuite()
|
||||
|
||||
for cls in (TestRequestResponseModels, TestDeviceDetection,
|
||||
TestAudioDecoding, TestModelLoadingFunctions,
|
||||
TestWarmupFunctions, TestRouteRegistration):
|
||||
suite.addTests(loader.loadTestsFromTestCase(cls))
|
||||
|
||||
runner = unittest.TextTestRunner(verbosity=2)
|
||||
result = runner.run(suite)
|
||||
return result.wasSuccessful()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
success = run_tests()
|
||||
sys.exit(0 if success else 1)
|
||||
Reference in New Issue
Block a user