208 lines
7.9 KiB
Python
208 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
from services.ai_voice_runtime_service.audiosocket import pcm16le_to_wav_bytes
|
|
from services.ai_voice_runtime_service.providers import asr as asr_module
|
|
|
|
|
|
class _DummyResponse:
|
|
def __init__(self, json_payload: dict | None = None) -> None:
|
|
self.content = b"{}"
|
|
self._json_payload = json_payload or {}
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict:
|
|
return self._json_payload
|
|
|
|
|
|
def test_yandex_asr_provider_posts_lpcm_with_tts_credential_fallback(monkeypatch):
|
|
calls: list[dict] = []
|
|
pcm = b"\x01\x00" * 160
|
|
wav_bytes = pcm16le_to_wav_bytes(pcm, sample_rate_hz=8000)
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> "_DummyClient":
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(
|
|
self,
|
|
url: str,
|
|
*,
|
|
headers: dict[str, str],
|
|
params: dict[str, str],
|
|
content: bytes,
|
|
) -> _DummyResponse:
|
|
calls.append(
|
|
{
|
|
"url": url,
|
|
"headers": headers,
|
|
"params": params,
|
|
"content": content,
|
|
"timeout": self.timeout,
|
|
}
|
|
)
|
|
return _DummyResponse({"result": "almaty schedule"})
|
|
|
|
monkeypatch.delenv("AI_VOICE_ASR_YANDEX_API_KEY", raising=False)
|
|
monkeypatch.delenv("AI_VOICE_ASR_YANDEX_IAM_TOKEN", raising=False)
|
|
monkeypatch.setenv("AI_VOICE_TTS_YANDEX_API_KEY", "tts-yandex-key")
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_API_BASE", "https://stt.example.test")
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_FOLDER_ID", "folder-test")
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_TIMEOUT_SECONDS", "5")
|
|
monkeypatch.setattr(asr_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = asr_module.YandexSpeechKitASRProvider()
|
|
result = provider.transcribe(wav_bytes, language_hint="ru")
|
|
|
|
assert result.text == "almaty schedule"
|
|
assert result.language == "ru-RU"
|
|
assert len(calls) == 1
|
|
assert calls[0]["url"] == "https://stt.example.test/speech/v1/stt:recognize"
|
|
assert calls[0]["headers"]["Authorization"] == "Api-Key tts-yandex-key"
|
|
assert calls[0]["headers"]["Content-Type"] == "application/octet-stream"
|
|
assert calls[0]["params"]["lang"] == "ru-RU"
|
|
assert calls[0]["params"]["format"] == "lpcm"
|
|
assert calls[0]["params"]["sampleRateHertz"] == "8000"
|
|
assert calls[0]["params"]["topic"] == "general"
|
|
assert calls[0]["params"]["folderId"] == "folder-test"
|
|
assert calls[0]["content"] == pcm
|
|
assert calls[0]["timeout"] == 5.0
|
|
|
|
|
|
def test_yandex_asr_provider_uses_iam_token(monkeypatch):
|
|
calls: list[dict] = []
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> "_DummyClient":
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(
|
|
self,
|
|
url: str,
|
|
*,
|
|
headers: dict[str, str],
|
|
params: dict[str, str],
|
|
content: bytes,
|
|
) -> _DummyResponse:
|
|
calls.append({"url": url, "headers": headers, "params": params, "content": content})
|
|
return _DummyResponse({"result": "operator"})
|
|
|
|
monkeypatch.delenv("AI_VOICE_ASR_YANDEX_API_KEY", raising=False)
|
|
monkeypatch.delenv("AI_VOICE_TTS_YANDEX_API_KEY", raising=False)
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_IAM_TOKEN", "iam-token")
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_API_BASE", "https://stt.example.test")
|
|
monkeypatch.setattr(asr_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = asr_module.YandexSpeechKitASRProvider()
|
|
result = provider.transcribe(b"\x02\x00" * 80, language_hint="kk")
|
|
|
|
assert result.text == "operator"
|
|
assert result.language == "kk-KZ"
|
|
assert calls[0]["headers"]["Authorization"] == "Bearer iam-token"
|
|
assert calls[0]["params"]["lang"] == "kk-KZ"
|
|
|
|
|
|
def test_yandex_asr_provider_supports_kz_v3_async_rest(monkeypatch):
|
|
calls: list[dict] = []
|
|
|
|
class _DummyClient:
|
|
def __init__(self, *, timeout: float) -> None:
|
|
self.timeout = timeout
|
|
|
|
def __enter__(self) -> "_DummyClient":
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, *, headers: dict[str, str], json: dict) -> _DummyResponse:
|
|
calls.append({"method": "POST", "url": url, "headers": headers, "json": json})
|
|
return _DummyResponse({"id": "operation-1", "done": True})
|
|
|
|
def get(
|
|
self,
|
|
url: str,
|
|
*,
|
|
headers: dict[str, str],
|
|
params: dict[str, str] | None = None,
|
|
) -> _DummyResponse:
|
|
calls.append({"method": "GET", "url": url, "headers": headers, "params": params or {}})
|
|
return _DummyResponse(
|
|
{
|
|
"finalRefinement": {
|
|
"normalizedText": {
|
|
"alternatives": [
|
|
{
|
|
"text": "almaty schedule",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
)
|
|
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_API_KEY", "asr-key")
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_API_BASE", "https://stt.api.ml.yandexcloud.kz")
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_FOLDER_ID", "folder-test")
|
|
monkeypatch.setenv("AI_VOICE_ASR_YANDEX_POLL_INTERVAL_SECONDS", "0.05")
|
|
monkeypatch.setattr(asr_module.httpx, "Client", _DummyClient)
|
|
|
|
provider = asr_module.YandexSpeechKitASRProvider()
|
|
result = provider.transcribe(b"\x03\x00" * 80, language_hint="ru")
|
|
|
|
assert result.text == "almaty schedule"
|
|
assert result.language == "ru-RU"
|
|
assert calls[0]["url"] == "https://stt.api.ml.yandexcloud.kz/stt/v3/recognizeFileAsync"
|
|
assert calls[0]["headers"]["Authorization"] == "Api-Key asr-key"
|
|
assert calls[0]["headers"]["Content-Type"] == "application/json"
|
|
assert calls[0]["headers"]["x-folder-id"] == "folder-test"
|
|
assert calls[0]["json"]["recognitionModel"]["audioFormat"]["rawAudio"]["sampleRateHertz"] == "8000"
|
|
assert calls[0]["json"]["recognitionModel"]["languageRestriction"]["languageCode"] == ["ru-RU"]
|
|
assert calls[1]["url"] == "https://stt.api.ml.yandexcloud.kz/stt/v3/getRecognition"
|
|
assert calls[1]["params"] == {"operationId": "operation-1"}
|
|
|
|
|
|
def test_yandex_buffered_streaming_provider_buffers_pcm_until_finalize():
|
|
calls: list[dict] = []
|
|
|
|
class _FakeYandexASR:
|
|
def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None):
|
|
calls.append({"audio_bytes": audio_bytes, "language_hint": language_hint})
|
|
return asr_module.ASRTranscription(text="need schedule", language=language_hint, confidence=None)
|
|
|
|
provider = asr_module.YandexSpeechKitBufferedStreamingASRProvider(asr_provider=_FakeYandexASR()) # type: ignore[arg-type]
|
|
stream_id = provider.open_stream("session-1", language_hint="ru")
|
|
|
|
provider.push_pcm(stream_id, b"\x10\x00")
|
|
provider.push_pcm(stream_id, b"\x20\x00")
|
|
|
|
assert provider.poll_partial(stream_id) is None
|
|
|
|
result = provider.finalize(stream_id)
|
|
provider.close_stream(stream_id)
|
|
|
|
assert result.text == "need schedule"
|
|
assert calls == [{"audio_bytes": b"\x10\x00\x20\x00", "language_hint": "ru"}]
|
|
|
|
|
|
def test_yandex_asr_builders():
|
|
assert isinstance(asr_module.build_asr_provider("yandex"), asr_module.YandexSpeechKitASRProvider)
|
|
assert isinstance(asr_module.build_asr_provider("speechkit"), asr_module.YandexSpeechKitASRProvider)
|
|
assert isinstance(
|
|
asr_module.build_streaming_asr_provider("yandex_speechkit"),
|
|
asr_module.YandexSpeechKitBufferedStreamingASRProvider,
|
|
)
|