from __future__ import annotations import time 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_grpc_streaming_provider_returns_partial_and_final(): calls: list[dict] = [] class _Message: def __init__(self, **kwargs) -> None: self.__dict__.update(kwargs) def HasField(self, name: str) -> bool: return getattr(self, name, None) is not None class _RawAudio(_Message): LINEAR16_PCM = 1 class _TextNormalizationOptions(_Message): TEXT_NORMALIZATION_ENABLED = 1 PHONE_FORMATTING_MODE_DISABLED = 1 class _LanguageRestrictionOptions(_Message): WHITELIST = 1 class _RecognitionModelOptions(_Message): REAL_TIME = 1 class _DefaultEouClassifier(_Message): HIGH = 2 class _FakeSttPb2: RawAudio = _RawAudio AudioFormatOptions = _Message TextNormalizationOptions = _TextNormalizationOptions LanguageRestrictionOptions = _LanguageRestrictionOptions RecognitionModelOptions = _RecognitionModelOptions DefaultEouClassifier = _DefaultEouClassifier EouClassifierOptions = _Message StreamingOptions = _Message StreamingRequest = _Message AudioChunk = _Message Eou = _Message class _FakeChannel: def close(self) -> None: calls.append({"method": "close"}) class _FakeGrpc: @staticmethod def ssl_channel_credentials(): return "ssl" @staticmethod def secure_channel(target: str, credentials): calls.append({"method": "secure_channel", "target": target, "credentials": credentials}) return _FakeChannel() class _RecognizerStub: def __init__(self, channel) -> None: calls.append({"method": "stub_init", "channel": channel}) def RecognizeStreaming(self, request_iterator, *, metadata, timeout): calls.append({"method": "recognize", "metadata": metadata, "timeout": timeout}) for request in request_iterator: calls.append({"method": "request", "request": request}) if getattr(request, "chunk", None) is not None: yield _Message( partial=_Message( alternatives=[ _Message(text="need schedule", confidence=0.7, languages=[]), ] ) ) continue if getattr(request, "eou", None) is not None: yield _Message( final_refinement=_Message( normalized_text=_Message( alternatives=[ _Message(text="need schedule in almaty", confidence=0.91, languages=[]), ] ) ) ) return class _FakeGrpcPb2: RecognizerStub = _RecognizerStub provider = asr_module.YandexSpeechKitGrpcStreamingASRProvider( api_key="asr-key", grpc_target="stt.test:443", timeout_seconds=3, grpc_module=_FakeGrpc, stt_pb2_module=_FakeSttPb2, stt_service_pb2_grpc_module=_FakeGrpcPb2, ) stream_id = provider.open_stream("session-1", language_hint="ru") provider.push_pcm(stream_id, b"\x10\x00\x20\x00") partial = None for _ in range(20): partial = provider.poll_partial(stream_id) if partial is not None: break time.sleep(0.02) assert partial is not None assert partial.text == "need schedule" assert partial.is_final is False final = provider.finalize(stream_id) provider.close_stream(stream_id) assert final.text == "need schedule in almaty" recognize_call = next(call for call in calls if call.get("method") == "recognize") assert ("authorization", "Api-Key asr-key") in recognize_call["metadata"] requests = [call["request"] for call in calls if call.get("method") == "request"] assert getattr(requests[0], "session_options", None) is not None assert getattr(requests[1], "chunk", None).data == b"\x10\x00\x20\x00" assert getattr(requests[-1], "eou", None) is not None 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.YandexSpeechKitGrpcStreamingASRProvider, ) assert isinstance( asr_module.build_streaming_asr_provider("yandex_speechkit_buffered"), asr_module.YandexSpeechKitBufferedStreamingASRProvider, )