PTSRaiAgent/tests/test_llm_stt.py

"""Speech-to-text providers (#116, #121): OpenRouter and SaluteSpeech request shape, retries, silence, WAV parsing,
provider choice and the «no foreign STT outside the demo» guard."""

import base64
import io
import json
import time
import wave

import httpx
import pytest
from pydantic import ValidationError

from app.config import Settings
from app.llm import stt as stt_mod
from app.llm.stt import (
    SALUTE_OAUTH_URL,
    SALUTE_RECOGNIZE_URL,
    InvalidAudio,
    MockSTT,
    OpenRouterSTT,
    SaluteSTT,
    SttError,
    SttNoSpeech,
    get_stt,
    load_texts,
    wav_seconds,
)


def make_wav(seconds: float, rate: int = 16_000) -> bytes:
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(rate)
        w.writeframes(b"\x00\x00" * int(rate * seconds))
    return buf.getvalue()


def _settings(**kw) -> Settings:
    base = {"openrouter_api_key": "test", "stt_provider": "openrouter", "stt_timeout_s": 5.0}
    return Settings(_env_file=None, **{**base, **kw})


def _fake(responses: list[httpx.Response]) -> tuple[OpenRouterSTT, list[dict]]:
    sent: list[dict] = []
    replies = iter(responses)

    def handler(request: httpx.Request) -> httpx.Response:
        sent.append(json.loads(request.content))
        return next(replies)

    stt = OpenRouterSTT(_settings(), client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
    return stt, sent


def _ok(text: str) -> httpx.Response:
    return httpx.Response(200, json={"choices": [{"message": {"content": text}}]})


def test_wav_seconds() -> None:
    assert wav_seconds(make_wav(2.5)) == pytest.approx(2.5)
    with pytest.raises(InvalidAudio):
        wav_seconds(b"OggS" + b"\x00" * 100)
    with pytest.raises(InvalidAudio):
        wav_seconds(b"")


async def test_openrouter_sends_audio_and_returns_transcript() -> None:
    audio = make_wav(1)
    stt, sent = _fake([_ok("«Сегодня плохо спал.»")])
    assert await stt.transcribe(audio) == "Сегодня плохо спал."
    body = sent[0]
    assert body["model"] == "google/gemini-2.5-flash" and body["temperature"] == 0
    assert body["reasoning"] == {"enabled": False}
    part = body["messages"][1]["content"][1]
    assert part["type"] == "input_audio" and part["input_audio"]["format"] == "wav"
    assert base64.b64decode(part["input_audio"]["data"]) == audio


async def test_silence_marker_is_no_speech() -> None:
    stt, _ = _fake([_ok(load_texts().silence_marker.upper())])
    with pytest.raises(SttNoSpeech):
        await stt.transcribe(make_wav(1))


async def test_retries_then_succeeds(monkeypatch) -> None:
    monkeypatch.setattr("app.llm.stt.asyncio.sleep", _no_sleep)
    stt, sent = _fake([httpx.Response(503), _ok("текст")])
    assert await stt.transcribe(make_wav(1)) == "текст"
    assert len(sent) == 2


async def test_http_error_raises(monkeypatch) -> None:
    monkeypatch.setattr("app.llm.stt.asyncio.sleep", _no_sleep)
    stt, _ = _fake([httpx.Response(401, json={"error": {"message": "bad key"}})])
    with pytest.raises(SttError):
        await stt.transcribe(make_wav(1))


async def _no_sleep(_: float) -> None:
    return None


async def test_mock_is_offline() -> None:
    assert await MockSTT().transcribe(b"") == load_texts().mock_transcript


def test_provider_choice() -> None:
    assert isinstance(get_stt(_settings(stt_provider="auto")), OpenRouterSTT)
    assert isinstance(get_stt(_settings(stt_provider="auto", openrouter_api_key="")), MockSTT)
    assert get_stt(_settings(stt_provider="openrouter", openrouter_api_key="")) is None
    assert get_stt(_settings(stt_provider="off")) is None
    # SaluteSpeech wins `auto` whenever its credentials are set.
    assert isinstance(get_stt(_settings(stt_provider="auto", salute_speech_credentials="c")), SaluteSTT)
    assert isinstance(get_stt(_settings(stt_provider="salute", salute_speech_credentials="c")), SaluteSTT)
    assert get_stt(_settings(stt_provider="salute")) is None


def test_outside_demo_no_foreign_stt() -> None:
    """#121: with real patients (DEMO_CLOCK=false) OpenRouter refuses to start, `auto` never picks it or mock."""
    with pytest.raises(ValidationError, match="DEMO_CLOCK=false"):
        _settings(stt_provider="openrouter", demo_clock=False)
    assert get_stt(_settings(stt_provider="auto", demo_clock=False)) is None
    salute = get_stt(_settings(stt_provider="auto", demo_clock=False, salute_speech_credentials="c"))
    assert isinstance(salute, SaluteSTT)


def test_salute_ca_must_exist(tmp_path) -> None:
    with pytest.raises(ValidationError, match="SALUTE_SPEECH_CA"):
        _settings(salute_speech_ca=tmp_path / "missing.crt")


class _SaluteFake:
    """Fake Sber: OAuth + recognize. `replies` are recognize responses in order."""

    def __init__(self, replies: list[httpx.Response]) -> None:
        self.replies = iter(replies)
        self.oauth_calls = 0
        self.recognize: list[httpx.Request] = []

    def __call__(self, request: httpx.Request) -> httpx.Response:
        if str(request.url) == SALUTE_OAUTH_URL:
            self.oauth_calls += 1
            assert request.headers["Authorization"] == "Basic cred-1"
            assert request.headers["RqUID"] and b"scope=SALUTE_SPEECH_PERS" in request.content
            exp_ms = int((time.time() + 1800) * 1000)
            return httpx.Response(200, json={"access_token": f"tok-{self.oauth_calls}", "expires_at": exp_ms})
        assert str(request.url).startswith(SALUTE_RECOGNIZE_URL)
        self.recognize.append(request)
        return next(self.replies)


def _salute(replies: list[httpx.Response]) -> tuple[SaluteSTT, _SaluteFake]:
    stt_mod._salute_token.cache_clear()
    fake = _SaluteFake(replies)
    settings = _settings(stt_provider="salute", salute_speech_credentials="cred-1")
    return SaluteSTT(settings, client=httpx.AsyncClient(transport=httpx.MockTransport(fake))), fake


def _said(*parts: str) -> httpx.Response:
    return httpx.Response(200, json={"status": 200, "result": list(parts), "emotions": []})


async def test_salute_sends_raw_pcm_and_joins_result() -> None:
    stt, fake = _salute([_said("сегодня плохо спал", "", "хочу обсудить")])
    wav = make_wav(1.0)
    assert await stt.transcribe(wav) == "сегодня плохо спал хочу обсудить"
    req = fake.recognize[0]
    assert req.headers["Content-Type"] == "audio/x-pcm;bit=16;rate=16000"
    assert req.headers["Authorization"] == "Bearer tok-1"
    assert req.url.params["language"] == "ru-RU"
    assert req.content == wav[44:]  # the WAV header is stripped: PCM only


async def test_salute_token_cached_and_refreshed_on_401() -> None:
    stt, fake = _salute([_said("раз"), _said("два"), httpx.Response(401), _said("три")])
    for _ in range(2):
        await stt.transcribe(make_wav(0.5))
    assert fake.oauth_calls == 1  # one token for both requests
    assert await stt.transcribe(make_wav(0.5)) == "три"
    assert fake.oauth_calls == 2 and fake.recognize[-1].headers["Authorization"] == "Bearer tok-2"


async def test_salute_empty_result_is_no_speech() -> None:
    stt, _ = _salute([_said()])
    with pytest.raises(SttNoSpeech):
        await stt.transcribe(make_wav(0.5))


async def test_salute_errors(monkeypatch) -> None:
    monkeypatch.setattr("app.llm.stt.asyncio.sleep", _no_sleep)
    stt, _ = _salute([httpx.Response(503), httpx.Response(503), httpx.Response(503)])
    with pytest.raises(SttError):
        await stt.transcribe(make_wav(0.5))
    stt, _ = _salute([httpx.Response(400, json={"status": 400, "message": "bad audio"})])
    with pytest.raises(SttError):
        await stt.transcribe(make_wav(0.5))


async def test_salute_rejects_non_pcm_mono() -> None:
    stt, fake = _salute([])
    buf = io.BytesIO()
    with wave.open(buf, "wb") as w:
        w.setnchannels(2)
        w.setsampwidth(2)
        w.setframerate(16_000)
        w.writeframes(b"\x00\x00" * 3200)
    with pytest.raises(InvalidAudio):
        await stt.transcribe(buf.getvalue())
    assert fake.oauth_calls == 0