209 lines
7.3 KiB
Python
209 lines
7.3 KiB
Python
"""Phase 5 tests: OpenRouter ASR provider unit tests.
|
|
|
|
Covers:
|
|
- Successful transcription via mocked httpx
|
|
- Retry logic on 429, 5xx
|
|
- Error handling for empty response, network errors
|
|
- Language parameter handling (passed / auto omitted)
|
|
"""
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.services.asr_providers import (
|
|
ASRError,
|
|
OpenRouterASRProvider,
|
|
create_asr_provider,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_httpx_client():
|
|
mock_client = AsyncMock(spec=httpx.AsyncClient)
|
|
mock_response = MagicMock(spec=httpx.Response)
|
|
mock_response.json.return_value = {"text": "測試轉錄結果", "usage": {}}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_client.post.return_value = mock_response
|
|
return mock_client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestOpenRouterTranscribe:
|
|
async def test_returns_traditional_chinese(self, mock_httpx_client):
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
provider._client = mock_httpx_client
|
|
|
|
result = await provider.transcribe(b"fake-wav-bytes", language="yue")
|
|
|
|
assert "測" in result or "試" in result or "轉" in result
|
|
|
|
async def test_sends_correct_payload(self, mock_httpx_client):
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
provider._client = mock_httpx_client
|
|
|
|
await provider.transcribe(b"fake-wav-bytes", language="yue")
|
|
|
|
call_args = mock_httpx_client.post.call_args
|
|
assert call_args is not None
|
|
payload = call_args.kwargs["json"]
|
|
assert payload["model"] == "google/gemini-3.1-flash-lite"
|
|
assert "data" in payload["input_audio"]
|
|
assert payload["input_audio"]["format"] == "wav"
|
|
assert payload["language"] == "yue"
|
|
|
|
async def test_auto_language_omitted(self, mock_httpx_client):
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
provider._client = mock_httpx_client
|
|
|
|
await provider.transcribe(b"fake-wav-bytes", language="auto")
|
|
|
|
call_args = mock_httpx_client.post.call_args
|
|
payload = call_args.kwargs["json"]
|
|
assert "language" not in payload
|
|
|
|
async def test_default_language_yue_passed(self, mock_httpx_client):
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
provider._client = mock_httpx_client
|
|
|
|
await provider.transcribe(b"fake-wav-bytes", language="yue")
|
|
|
|
call_args = mock_httpx_client.post.call_args
|
|
payload = call_args.kwargs["json"]
|
|
assert payload.get("language") == "yue"
|
|
|
|
async def test_raises_on_empty_text(self, mock_httpx_client):
|
|
mock_httpx_client.post.return_value.json.return_value = {"text": "", "usage": {}}
|
|
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
provider._client = mock_httpx_client
|
|
|
|
with pytest.raises(ASRError, match="empty transcription"):
|
|
await provider.transcribe(b"fake-wav-bytes", language="yue")
|
|
|
|
async def test_raises_on_http_error(self):
|
|
mock_client = AsyncMock(spec=httpx.AsyncClient)
|
|
mock_response = MagicMock(spec=httpx.Response)
|
|
mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
|
|
"Server error",
|
|
request=MagicMock(),
|
|
response=MagicMock(status_code=500),
|
|
)
|
|
mock_client.post.return_value = mock_response
|
|
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
provider._client = mock_client
|
|
|
|
with pytest.raises(ASRError, match="STT request failed"):
|
|
await provider.transcribe(b"fake-wav-bytes", language="yue")
|
|
|
|
async def test_raises_on_network_error(self):
|
|
mock_client = AsyncMock(spec=httpx.AsyncClient)
|
|
mock_client.post.side_effect = httpx.ConnectError("Connection refused")
|
|
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
provider._client = mock_client
|
|
|
|
with pytest.raises(ASRError, match="STT request failed"):
|
|
await provider.transcribe(b"fake-wav-bytes", language="yue")
|
|
|
|
|
|
class TestSttUrlConstruction:
|
|
def test_appends_audio_transcriptions(self):
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
assert provider._stt_url == "https://openrouter.ai/api/v1/audio/transcriptions"
|
|
|
|
def test_handles_trailing_slash(self):
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1/",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
assert provider._stt_url == "https://openrouter.ai/api/v1/audio/transcriptions"
|
|
|
|
|
|
class TestCloseClient:
|
|
@pytest.mark.asyncio
|
|
async def test_close_cleans_up_client(self):
|
|
mock_client = AsyncMock(spec=httpx.AsyncClient)
|
|
provider = OpenRouterASRProvider(
|
|
api_key="sk-test",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="google/gemini-3.1-flash-lite",
|
|
)
|
|
provider._client = mock_client
|
|
|
|
await provider.close()
|
|
mock_client.aclose.assert_awaited_once()
|
|
assert provider._client is None
|
|
|
|
|
|
class TestCreateAsrProvider:
|
|
def test_creates_dashscope(self, monkeypatch):
|
|
settings = MagicMock()
|
|
settings.asr_provider = "dashscope"
|
|
settings.dashscope_api_key = "sk-test"
|
|
settings.asr_model_name = "qwen3-asr-flash"
|
|
|
|
from app.services.asr_providers import DashScopeASRProvider
|
|
provider = create_asr_provider(settings)
|
|
assert isinstance(provider, DashScopeASRProvider)
|
|
|
|
def test_creates_openrouter(self, monkeypatch):
|
|
settings = MagicMock()
|
|
settings.asr_provider = "openrouter"
|
|
settings.openrouter_api_key = "sk-or-test"
|
|
settings.llm_base_url = "https://openrouter.ai/api/v1"
|
|
settings.asr_openrouter_model = "google/gemini-3.1-flash-lite"
|
|
|
|
provider = create_asr_provider(settings)
|
|
assert isinstance(provider, OpenRouterASRProvider)
|
|
|
|
def test_missing_openrouter_key_raises(self, monkeypatch):
|
|
settings = MagicMock()
|
|
settings.asr_provider = "openrouter"
|
|
settings.openrouter_api_key = ""
|
|
|
|
with pytest.raises(ASRError, match="OPENROUTER_API_KEY"):
|
|
create_asr_provider(settings)
|
|
|
|
def test_unknown_provider_raises(self, monkeypatch):
|
|
settings = MagicMock()
|
|
settings.asr_provider = "unknown"
|
|
|
|
with pytest.raises(ValueError, match="Unknown ASR provider"):
|
|
create_asr_provider(settings)
|