call-center/tests/test_streaming_asr_sidecar_service.py

import base64
import time
from pathlib import Path

from fastapi.testclient import TestClient

import services.streaming_asr_sidecar_service.app as sidecar_module


def _admin_headers() -> dict[str, str]:
    return {"X-User": "admin", "X-Role": "admin"}


def _pcm_chunk(duration_ms: int = 400, *, amplitude: int = 1000) -> bytes:
    sample_count = int(8000 * (duration_ms / 1000.0))
    return int(amplitude).to_bytes(2, "little", signed=True) * sample_count


class _FakeEngine:
    def __init__(self, *, text: str = "привет") -> None:
        self.text = text
        self.calls: list[tuple[int, int, str | None]] = []

    def transcribe_pcm(
        self,
        pcm_bytes: bytes,
        *,
        sample_rate_hz: int,
        language_hint: str | None = None,
    ) -> sidecar_module.SidecarTranscript:
        self.calls.append((len(pcm_bytes), sample_rate_hz, language_hint))
        return sidecar_module.SidecarTranscript(
            text=self.text,
            language=language_hint or "ru",
            confidence=0.91,
        )


def _reset_sidecar(
    monkeypatch,
    *,
    engine: _FakeEngine | None = None,
    idle_ttl_seconds: float = 45.0,
    supported_languages: set[str] | None = None,
) -> _FakeEngine:
    fake_engine = engine or _FakeEngine()
    monkeypatch.setattr(sidecar_module, "_ENGINE_INSTANCE", None)
    monkeypatch.setattr(sidecar_module, "_ENGINE_FACTORY", lambda: fake_engine)
    monkeypatch.setattr(sidecar_module, "_STREAMS", sidecar_module.StreamStore(idle_ttl_seconds=idle_ttl_seconds))
    monkeypatch.setattr(
        sidecar_module,
        "_supported_languages",
        lambda: set(supported_languages or {"ru"}),
    )
    return fake_engine


def test_streaming_asr_sidecar_stream_lifecycle(monkeypatch):
    engine = _reset_sidecar(monkeypatch)
    client = TestClient(sidecar_module.app)

    open_response = client.post(
        "/internal/asr/streams",
        headers=_admin_headers(),
        json={
            "session_id": "avs_sidecar_1",
            "language_hint": "ru",
            "sample_rate_hz": 8000,
            "encoding": "pcm_s16le",
        },
    )
    assert open_response.status_code == 200
    stream_id = open_response.json()["stream_id"]

    chunk = _pcm_chunk(400)
    push_response = client.post(
        f"/internal/asr/streams/{stream_id}/chunks",
        headers=_admin_headers(),
        json={
            "pcm_b64": base64.b64encode(chunk).decode("ascii"),
            "sample_rate_hz": 8000,
            "encoding": "pcm_s16le",
        },
    )
    assert push_response.status_code == 200
    assert push_response.json()["received_bytes"] == len(chunk)

    partial_one = client.get(
        f"/internal/asr/streams/{stream_id}/partial",
        headers=_admin_headers(),
    )
    assert partial_one.status_code == 200
    first_payload = partial_one.json()
    assert first_payload["text"] == "привет"
    assert first_payload["is_stable"] is False

    second_chunk = _pcm_chunk(80)
    client.post(
        f"/internal/asr/streams/{stream_id}/chunks",
        headers=_admin_headers(),
        json={
            "pcm_b64": base64.b64encode(second_chunk).decode("ascii"),
            "sample_rate_hz": 8000,
            "encoding": "pcm_s16le",
        },
    )
    partial_two = client.get(
        f"/internal/asr/streams/{stream_id}/partial",
        headers=_admin_headers(),
    )
    assert partial_two.status_code == 200
    second_payload = partial_two.json()
    assert second_payload["text"] == "привет"
    assert second_payload["is_stable"] is True

    finalize_response = client.post(
        f"/internal/asr/streams/{stream_id}/finalize",
        headers=_admin_headers(),
    )
    assert finalize_response.status_code == 200
    assert finalize_response.json() == {
        "text": "привет",
        "language": "ru",
        "confidence": 0.91,
    }

    close_response = client.delete(
        f"/internal/asr/streams/{stream_id}",
        headers=_admin_headers(),
    )
    assert close_response.status_code == 200
    assert close_response.json() == {"ok": True, "stream_id": stream_id}
    assert len(engine.calls) >= 3


def test_streaming_asr_sidecar_partial_uses_recent_audio_tail(monkeypatch):
    engine = _reset_sidecar(monkeypatch)
    monkeypatch.setattr(sidecar_module, "_partial_max_audio_ms", lambda: 1800)
    client = TestClient(sidecar_module.app)

    open_response = client.post(
        "/internal/asr/streams",
        headers=_admin_headers(),
        json={
            "session_id": "avs_sidecar_tail",
            "language_hint": "ru",
            "sample_rate_hz": 8000,
            "encoding": "pcm_s16le",
        },
    )
    assert open_response.status_code == 200
    stream_id = open_response.json()["stream_id"]

    long_chunk = _pcm_chunk(3200)
    push_response = client.post(
        f"/internal/asr/streams/{stream_id}/chunks",
        headers=_admin_headers(),
        json={
            "pcm_b64": base64.b64encode(long_chunk).decode("ascii"),
            "sample_rate_hz": 8000,
            "encoding": "pcm_s16le",
        },
    )
    assert push_response.status_code == 200

    partial_response = client.get(
        f"/internal/asr/streams/{stream_id}/partial",
        headers=_admin_headers(),
    )
    assert partial_response.status_code == 200
    assert engine.calls
    partial_len, sample_rate_hz, language_hint = engine.calls[-1]
    assert sample_rate_hz == 8000
    assert language_hint == "ru"
    assert partial_len < len(long_chunk)
    assert partial_len <= int(1.8 * 8000 * 2)


def test_streaming_asr_sidecar_rejects_unsupported_language(monkeypatch):
    _reset_sidecar(monkeypatch, supported_languages={"ru"})
    client = TestClient(sidecar_module.app)

    response = client.post(
        "/internal/asr/streams",
        headers=_admin_headers(),
        json={
            "session_id": "avs_sidecar_kk",
            "language_hint": "kk",
            "sample_rate_hz": 8000,
            "encoding": "pcm_s16le",
        },
    )

    assert response.status_code == 400
    assert "Unsupported streaming ASR language" in response.json()["detail"]


def test_streaming_asr_sidecar_cleans_up_idle_streams(monkeypatch):
    _reset_sidecar(monkeypatch, idle_ttl_seconds=0.01)
    client = TestClient(sidecar_module.app)

    open_response = client.post(
        "/internal/asr/streams",
        headers=_admin_headers(),
        json={
            "session_id": "avs_sidecar_idle",
            "language_hint": "ru",
            "sample_rate_hz": 8000,
            "encoding": "pcm_s16le",
        },
    )
    assert open_response.status_code == 200
    stream_id = open_response.json()["stream_id"]

    with sidecar_module._STREAMS._lock:
        sidecar_module._STREAMS._streams[stream_id].last_activity_monotonic = time.monotonic() - 1.0

    health_response = client.get("/health")
    assert health_response.status_code == 200

    missing_response = client.get(
        f"/internal/asr/streams/{stream_id}/partial",
        headers=_admin_headers(),
    )
    assert missing_response.status_code == 404


def test_streaming_asr_sidecar_requirements_include_requests_dependency():
    requirements_path = Path(__file__).resolve().parents[1] / "requirements.txt"
    requirements = {
        line.strip()
        for line in requirements_path.read_text(encoding="utf-8").splitlines()
        if line.strip() and not line.lstrip().startswith("#")
    }

    assert any(line.startswith("faster-whisper==") for line in requirements)
    assert any(line.startswith("requests==") for line in requirements)