77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
"""Phase 4 integration tests: System Audio and Mic capture WebSocket endpoint.
|
|
|
|
Covers:
|
|
- WebSocket handshake with UUID-format video_id (no actual video file)
|
|
- source=system-audio connection accepted with language param
|
|
- source=mic connection accepted with language param
|
|
- Config toggles disable both system-audio and mic features
|
|
|
|
Uses FastAPI TestClient with real router. Only external DashScope ASR is
|
|
implicitly avoided (client disconnects before proxy call completes).
|
|
"""
|
|
import uuid
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def phase4_integration_app(monkeypatch):
|
|
monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test-key")
|
|
monkeypatch.setenv("SYSTEM_AUDIO_ENABLED", "true")
|
|
monkeypatch.setenv("MIC_ENABLED", "true")
|
|
from app.core.config import get_settings
|
|
from app.routers.ws_asr import router
|
|
get_settings.cache_clear()
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return app
|
|
|
|
|
|
def test_websocket_accepts_uuid_video_id(phase4_integration_app):
|
|
"""WebSocket connects with a proper UUID video_id; no actual video file needed."""
|
|
video_uuid = str(uuid.uuid4())
|
|
client = TestClient(phase4_integration_app)
|
|
with client.websocket_connect(f"/ws/asr/{video_uuid}") as ws:
|
|
pass
|
|
|
|
|
|
def test_websocket_accepts_system_audio_source(phase4_integration_app):
|
|
"""WebSocket with source=system-audio and language=en connects for audio processing."""
|
|
client = TestClient(phase4_integration_app)
|
|
with client.websocket_connect(
|
|
"/ws/asr/integ-test-vid?source=system-audio&language=en"
|
|
) as ws:
|
|
pass
|
|
|
|
|
|
def test_websocket_accepts_mic_source(phase4_integration_app):
|
|
"""WebSocket with source=mic and language=zh connects successfully."""
|
|
client = TestClient(phase4_integration_app)
|
|
with client.websocket_connect(
|
|
"/ws/asr/integ-test-vid?source=mic&language=zh"
|
|
) as ws:
|
|
pass
|
|
|
|
|
|
def test_config_toggles_disable_features(monkeypatch):
|
|
"""When both toggles disabled, system-audio and mic sources return error messages."""
|
|
monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test-key")
|
|
monkeypatch.setenv("SYSTEM_AUDIO_ENABLED", "false")
|
|
monkeypatch.setenv("MIC_ENABLED", "false")
|
|
from app.core.config import get_settings
|
|
from app.routers.ws_asr import router
|
|
get_settings.cache_clear()
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
client = TestClient(app)
|
|
|
|
with client.websocket_connect("/ws/asr/vid-1?source=system-audio") as ws:
|
|
data = ws.receive_json()
|
|
assert "disabled" in data.get("error", "").lower()
|
|
|
|
with client.websocket_connect("/ws/asr/vid-2?source=mic") as ws:
|
|
data = ws.receive_json()
|
|
assert "disabled" in data.get("error", "").lower()
|