wehub-resource-sync
89 lines
2.5 KiB
Python
89 lines
2.5 KiB
Python
"""Tests for Gradium STT plugin configuration and behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.plugin("gradium")
|
|
|
|
|
|
def test_stt_default_language():
|
|
"""STT defaults to English."""
|
|
from livekit.plugins.gradium import STT
|
|
|
|
stt = STT(api_key="test-key")
|
|
|
|
assert stt._opts.language == "en"
|
|
|
|
|
|
def test_stt_custom_language():
|
|
"""STT accepts custom language."""
|
|
from livekit.plugins.gradium import STT
|
|
|
|
stt = STT(api_key="test-key", language="fr")
|
|
|
|
assert stt._opts.language == "fr"
|
|
|
|
|
|
def test_stream_language_overrides_stream_options():
|
|
"""stream language overrides the copied stream config."""
|
|
from livekit.plugins.gradium import STT
|
|
|
|
stt = STT(api_key="test-key", language="fr", http_session=MagicMock())
|
|
|
|
def _fake_create_task(coro, *args, **kwargs):
|
|
coro.close()
|
|
task = MagicMock()
|
|
return task
|
|
|
|
with patch("livekit.agents.stt.stt.asyncio.create_task", side_effect=_fake_create_task):
|
|
stream = stt.stream(language="de")
|
|
|
|
assert stream._opts.language == "de"
|
|
assert stt._opts.language == "fr"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_speech_stream_sends_language_in_json_config():
|
|
"""SpeechStream sends language in setup json_config."""
|
|
from livekit.agents import DEFAULT_API_CONNECT_OPTIONS, LanguageCode
|
|
from livekit.plugins.gradium import STT
|
|
from livekit.plugins.gradium.stt import SpeechStream, STTOptions
|
|
|
|
sent_messages: list[str] = []
|
|
|
|
class FakeWebSocket:
|
|
async def send_str(self, message: str) -> None:
|
|
sent_messages.append(message)
|
|
|
|
class FakeSession:
|
|
async def ws_connect(self, url, headers):
|
|
return FakeWebSocket()
|
|
|
|
stt = STT(api_key="test-key")
|
|
opts = STTOptions(language=LanguageCode("pt"), temperature=0.3)
|
|
|
|
def _fake_create_task(coro, *args, **kwargs):
|
|
coro.close()
|
|
task = MagicMock()
|
|
return task
|
|
|
|
with patch("livekit.agents.stt.stt.asyncio.create_task", side_effect=_fake_create_task):
|
|
stream = SpeechStream(
|
|
stt=stt,
|
|
opts=opts,
|
|
conn_options=DEFAULT_API_CONNECT_OPTIONS,
|
|
api_key="test-key",
|
|
model_endpoint="wss://api.gradium.ai/api/speech/asr",
|
|
model_name="default",
|
|
http_session=FakeSession(),
|
|
)
|
|
|
|
await stream._connect_ws()
|
|
|
|
setup_msg = json.loads(sent_messages[0])
|
|
assert setup_msg["json_config"] == {"language": "pt", "temp": 0.3}
|