wehub-resource-sync
157 lines
5.1 KiB
Python
157 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.plugin("deepgram")
|
|
|
|
|
|
class _FakeWS:
|
|
def __init__(self) -> None:
|
|
self.sent: list[str] = []
|
|
self.closed = False
|
|
|
|
async def send_str(self, data: str) -> None:
|
|
self.sent.append(data)
|
|
|
|
|
|
def _make_flux_stream(*, ws=None, **opts_kwargs):
|
|
# exercise SpeechStreamv2's update logic without starting its connection task
|
|
from livekit.agents.types import NOT_GIVEN
|
|
from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2, STTOptions
|
|
|
|
opts = STTOptions(
|
|
model="flux-general-en",
|
|
sample_rate=16000,
|
|
keyterm=[],
|
|
endpoint_url="wss://api.deepgram.com/v2/listen",
|
|
eot_threshold=opts_kwargs.get("eot_threshold", NOT_GIVEN),
|
|
eager_eot_threshold=opts_kwargs.get("eager_eot_threshold", NOT_GIVEN),
|
|
eot_timeout_ms=opts_kwargs.get("eot_timeout_ms", NOT_GIVEN),
|
|
language_hint=opts_kwargs.get("language_hint", []),
|
|
)
|
|
opts.keyterm = opts_kwargs.get("keyterm", [])
|
|
stream = SimpleNamespace(
|
|
_opts=opts,
|
|
_reconnect_event=asyncio.Event(),
|
|
_reconfigure_atask=None,
|
|
_ws=ws,
|
|
)
|
|
stream._send_configure = SpeechStreamv2._send_configure.__get__(stream)
|
|
return stream
|
|
|
|
|
|
async def test_update_options_uses_stored_language_for_model_validation():
|
|
from livekit.plugins.deepgram import STT
|
|
|
|
stt = STT(api_key="test-key", language="fr")
|
|
stt.update_options(model="nova-2-meeting")
|
|
assert stt._opts.model == "nova-2-general"
|
|
|
|
|
|
async def test_update_options_explicit_language_overrides_stored():
|
|
from livekit.plugins.deepgram import STT
|
|
|
|
stt = STT(api_key="test-key", language="fr")
|
|
stt.update_options(model="nova-2-meeting", language="en-US")
|
|
assert stt._opts.model == "nova-2-meeting"
|
|
|
|
|
|
async def test_update_options_no_language_set_keeps_en_only_model():
|
|
from livekit.plugins.deepgram import STT
|
|
|
|
stt = STT(api_key="test-key")
|
|
stt.update_options(model="nova-2-meeting")
|
|
assert stt._opts.model == "nova-2-meeting"
|
|
|
|
|
|
async def test_flux_live_fields_reconfigure_without_reconnect():
|
|
from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2
|
|
|
|
ws = _FakeWS()
|
|
stream = _make_flux_stream(ws=ws, eot_threshold=0.7)
|
|
SpeechStreamv2.update_options(
|
|
stream, eot_threshold=0.85, keyterm=["LiveKit"], language_hint=["en"]
|
|
)
|
|
|
|
# a Configure send is scheduled in-band, no reconnect
|
|
assert not stream._reconnect_event.is_set()
|
|
assert stream._reconfigure_atask is not None
|
|
await stream._reconfigure_atask
|
|
|
|
assert stream._opts.eot_threshold == 0.85
|
|
assert len(ws.sent) == 1
|
|
assert json.loads(ws.sent[0]) == {
|
|
"type": "Configure",
|
|
"thresholds": {"eot_threshold": 0.85},
|
|
"keyterms": ["LiveKit"],
|
|
"language_hints": ["en"],
|
|
}
|
|
|
|
|
|
async def test_flux_reconnect_fields_skip_inband_configure():
|
|
from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2
|
|
|
|
ws = _FakeWS()
|
|
stream = _make_flux_stream(ws=ws)
|
|
SpeechStreamv2.update_options(stream, model="flux-general-multi", eot_threshold=0.8)
|
|
|
|
# model can't be tuned in-band; a reconnect carries every option instead
|
|
assert stream._reconnect_event.is_set()
|
|
assert stream._reconfigure_atask is None
|
|
assert ws.sent == []
|
|
|
|
|
|
async def test_flux_configure_sends_only_changed_fields():
|
|
from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2
|
|
|
|
# stream already has a threshold and keyterms configured
|
|
ws = _FakeWS()
|
|
stream = _make_flux_stream(ws=ws, eot_threshold=0.7, keyterm=["existing"])
|
|
|
|
# only keyterms change: the Configure delta must omit the unchanged threshold
|
|
SpeechStreamv2.update_options(stream, keyterm=["LiveKit", "Deepgram"])
|
|
await stream._reconfigure_atask
|
|
|
|
assert json.loads(ws.sent[0]) == {"type": "Configure", "keyterms": ["LiveKit", "Deepgram"]}
|
|
|
|
|
|
async def test_flux_configure_thresholds_only_delta():
|
|
from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2
|
|
|
|
ws = _FakeWS()
|
|
stream = _make_flux_stream(ws=ws, keyterm=["existing"])
|
|
|
|
SpeechStreamv2.update_options(stream, eot_timeout_ms=5000)
|
|
await stream._reconfigure_atask
|
|
|
|
assert json.loads(ws.sent[0]) == {"type": "Configure", "thresholds": {"eot_timeout_ms": 5000}}
|
|
|
|
|
|
async def test_flux_configure_sends_are_ordered():
|
|
from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2
|
|
|
|
ws = _FakeWS()
|
|
stream = _make_flux_stream(ws=ws, eot_threshold=0.7)
|
|
|
|
# rapid successive updates chain off each other and reach the server in order
|
|
SpeechStreamv2.update_options(stream, eot_threshold=0.8)
|
|
SpeechStreamv2.update_options(stream, eot_threshold=0.9)
|
|
await stream._reconfigure_atask
|
|
|
|
assert [json.loads(m)["thresholds"]["eot_threshold"] for m in ws.sent] == [0.8, 0.9]
|
|
|
|
|
|
async def test_flux_configure_noop_when_disconnected():
|
|
from livekit.plugins.deepgram.stt_v2 import SpeechStreamv2
|
|
|
|
stream = _make_flux_stream(ws=None)
|
|
# no active connection: the next reconnect carries the latest options instead
|
|
SpeechStreamv2.update_options(stream, keyterm=["LiveKit"])
|
|
await stream._reconfigure_atask
|
|
|
|
assert stream._opts.keyterm == ["LiveKit"]
|