From 733824c17777367ac21d21b324789bb964fc7984 Mon Sep 17 00:00:00 2001 From: Woody Date: Tue, 19 May 2026 09:48:37 +0800 Subject: [PATCH] test: add Phase 5 ASR provider and integration tests test_phase5_config.py: 6 tests for ASR_PROVIDER validation and default values. test_phase5_openrouter_provider.py: 14 tests covering OpenRouterSTT transcription, retry logic, error handling, URL construction, cleanup, and factory dispatch. test_phase5_integration.py: 4 tests for full video-to-transcribe flow with both providers (mocked) and per-provider API key validation. Co-authored-by: Sisyphus --- backend/app/test/test_phase5_config.py | 63 ++++++ backend/app/test/test_phase5_integration.py | 125 +++++++++++ .../test/test_phase5_openrouter_provider.py | 208 ++++++++++++++++++ 3 files changed, 396 insertions(+) create mode 100644 backend/app/test/test_phase5_config.py create mode 100644 backend/app/test/test_phase5_integration.py create mode 100644 backend/app/test/test_phase5_openrouter_provider.py diff --git a/backend/app/test/test_phase5_config.py b/backend/app/test/test_phase5_config.py new file mode 100644 index 0000000..bc9cf2a --- /dev/null +++ b/backend/app/test/test_phase5_config.py @@ -0,0 +1,63 @@ +"""Phase 5 tests: ASR configuration validation. + +Covers: +- Valid ASR_PROVIDER values (dashscope, openrouter) load correctly +- Invalid ASR_PROVIDER raises ValueError +- Default values for new Phase 5 settings +""" +import os + +import pytest + + +class TestAsrProviderConfig: + def test_dashscope_is_default(self, monkeypatch, tmp_path): + monkeypatch.setenv("ASR_PROVIDER", "dashscope") + monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test") + from app.core.config import get_settings + get_settings.cache_clear() + s = get_settings() + assert s.asr_provider == "dashscope" + + def test_openrouter_provider_loads(self, monkeypatch, tmp_path): + monkeypatch.setenv("ASR_PROVIDER", "openrouter") + monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test") + monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test") + from app.core.config import get_settings + get_settings.cache_clear() + s = get_settings() + assert s.asr_provider == "openrouter" + + def test_invalid_provider_raises_valueerror(self, monkeypatch, tmp_path): + monkeypatch.setenv("ASR_PROVIDER", "invalid_provider") + monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test") + monkeypatch.setenv("OPENROUTER_API_KEY", "") + from app.core.config import get_settings + get_settings.cache_clear() + with pytest.raises(ValueError, match="Invalid ASR_PROVIDER"): + get_settings() + + def test_openrouter_api_key_defaults_empty(self, monkeypatch, tmp_path): + monkeypatch.setenv("ASR_PROVIDER", "dashscope") + monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test") + monkeypatch.setenv("OPENROUTER_API_KEY", "") + from app.core.config import get_settings + get_settings.cache_clear() + s = get_settings() + assert s.openrouter_api_key == "" + + def test_asr_openrouter_model_default(self, monkeypatch, tmp_path): + monkeypatch.setenv("ASR_PROVIDER", "dashscope") + monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test") + from app.core.config import get_settings + get_settings.cache_clear() + s = get_settings() + assert s.asr_openrouter_model == "google/gemini-3.1-flash-lite" + + def test_openrouter_model_customizable(self, monkeypatch, tmp_path): + monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test") + monkeypatch.setenv("ASR_OPENROUTER_MODEL", "openai/whisper-large-v3") + from app.core.config import get_settings + get_settings.cache_clear() + s = get_settings() + assert s.asr_openrouter_model == "openai/whisper-large-v3" diff --git a/backend/app/test/test_phase5_integration.py b/backend/app/test/test_phase5_integration.py new file mode 100644 index 0000000..bd8db18 --- /dev/null +++ b/backend/app/test/test_phase5_integration.py @@ -0,0 +1,125 @@ +"""Phase 5 tests: Integration — full video → transcribe with provider switching. + +Covers: +- Full transcript with dashscope provider (mocked OpenAI) +- Full transcript with openrouter provider (mocked httpx) +- API key validation per provider +""" +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.routers.video import router + + +@pytest.fixture +def video_client(tmp_path, monkeypatch): + upload_dir = tmp_path / "test_uploads" + upload_dir.mkdir() + monkeypatch.setenv("VIDEO_UPLOAD_DIR", str(upload_dir)) + monkeypatch.setenv("MAX_VIDEO_SIZE_MB", "50") + + from app.core.config import get_settings + get_settings.cache_clear() + app = FastAPI() + app.include_router(router, prefix="/api/v1") + return TestClient(app), upload_dir + + +def _upload_video(client, filename="test.mp4", content=b"\x00" * 1024): + resp = client.post( + "/api/v1/video/upload", + files={"file": (filename, content, "video/mp4")}, + ) + assert resp.status_code == 200 + return resp.json()["video_id"] + + +class TestDashScopeIntegration: + @patch("app.routers.video.VideoService.extract_audio") + @patch("app.services.asr_providers.OpenAI") + def test_transcribe_with_dashscope(self, mock_openai_cls, mock_extract, video_client, monkeypatch): + monkeypatch.setenv("ASR_PROVIDER", "dashscope") + monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-dashscope-test") + from app.core.config import get_settings + get_settings.cache_clear() + + client, upload_dir = video_client + video_id = _upload_video(client) + + fake_wav = upload_dir / "extracted.wav" + fake_wav.write_bytes(b"RIFF" + b"\x00" * 100) + mock_extract.return_value = fake_wav + + mock_resp = MagicMock() + mock_resp.choices = [MagicMock()] + mock_resp.choices[0].message.content = "测试转录結果" + mock_openai_instance = MagicMock() + mock_openai_instance.chat.completions.create.return_value = mock_resp + mock_openai_cls.return_value = mock_openai_instance + + resp = client.post(f"/api/v1/video/{video_id}/transcribe") + assert resp.status_code == 200 + assert "text" in resp.json() + assert "測" in resp.json()["text"] or "轉" in resp.json()["text"] + + +class TestOpenRouterIntegration: + @patch("app.routers.video.VideoService.extract_audio") + @patch("app.services.asr_providers.httpx.AsyncClient") + def test_transcribe_with_openrouter(self, mock_httpx_cls, mock_extract, video_client, monkeypatch): + monkeypatch.setenv("ASR_PROVIDER", "openrouter") + monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test") + from app.core.config import get_settings + get_settings.cache_clear() + + client, upload_dir = video_client + video_id = _upload_video(client) + + fake_wav = upload_dir / "extracted.wav" + fake_wav.write_bytes(b"RIFF" + b"\x00" * 100) + mock_extract.return_value = fake_wav + + mock_response = MagicMock(spec=httpx.Response) + mock_response.json.return_value = {"text": "測試轉錄結果", "usage": {}} + mock_response.raise_for_status = MagicMock() + + mock_http_client = AsyncMock() + mock_http_client.post.return_value = mock_response + mock_httpx_cls.return_value = mock_http_client + + resp = client.post(f"/api/v1/video/{video_id}/transcribe") + assert resp.status_code == 200 + assert "text" in resp.json() + assert "轉" in resp.json()["text"] or "錄" in resp.json()["text"] + + +class TestApiKeyValidation: + def test_missing_dashscope_key_returns_500(self, video_client, monkeypatch): + monkeypatch.setenv("ASR_PROVIDER", "dashscope") + monkeypatch.setenv("DASHSCOPE_API_KEY", "") + from app.core.config import get_settings + get_settings.cache_clear() + + client, upload_dir = video_client + video_id = _upload_video(client) + + resp = client.post(f"/api/v1/video/{video_id}/transcribe") + assert resp.status_code == 500 + assert "DASHSCOPE_API_KEY" in resp.json()["detail"] + + def test_missing_openrouter_key_returns_500(self, video_client, monkeypatch): + monkeypatch.setenv("ASR_PROVIDER", "openrouter") + monkeypatch.setenv("OPENROUTER_API_KEY", "") + from app.core.config import get_settings + get_settings.cache_clear() + + client, upload_dir = video_client + video_id = _upload_video(client) + + resp = client.post(f"/api/v1/video/{video_id}/transcribe") + assert resp.status_code == 500 + assert "OPENROUTER_API_KEY" in resp.json()["detail"] diff --git a/backend/app/test/test_phase5_openrouter_provider.py b/backend/app/test/test_phase5_openrouter_provider.py new file mode 100644 index 0000000..45fcf17 --- /dev/null +++ b/backend/app/test/test_phase5_openrouter_provider.py @@ -0,0 +1,208 @@ +"""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)