VoiceStudio/tests/test_capture_null_segment_end.py

"""
A segment timing the engine could not determine arrives as ``end: None``, and
the REST `/transcribe` response builder used to raise on it.

`_sherpa_result()` (services/asr_backend.py) sets ``duration = None`` when it
cannot derive one from the sample rate, and `OpenAICompatASRBackend`
`_adapt_response()` emits ``end: None`` for every plain-text (`json`/`text`)
response from a server that rejects `verbose_json`. Both reach this endpoint —
sherpa is the first capture engine and the OpenAI-compatible backend is
selectable as the active one.

`round(s.get("end", 0), 2)` does not defend against that: ``.get`` returns the
stored ``None`` rather than the default, because the key is present. The route
answered 500 for a transcript that was otherwise fine, which is the server-side
half of #1904.
"""
import os

import pytest

os.environ.setdefault("OMNIVOICE_MODEL", "test")
os.environ.setdefault("OMNIVOICE_DISABLE_FILE_LOG", "1")

pytestmark = pytest.mark.usefixtures("asr_model_installed")


class _UntimedBackend:
    """What sherpa and the OpenAI-compatible server both hand back when no
    timing is available: text, a start of 0.0, and an honest null end."""
    id = "untimed"

    def transcribe(self, _path, **_kw):
        return {
            "text": "the meeting is at three",
            "segments": [
                {"start": 0.0, "end": None, "text": "the meeting is at three"},
            ],
            "language": "en",
        }


class _PartlyTimedBackend:
    """One timed segment and one the engine gave up on — the duration must come
    from the half that is known, not from the null."""
    id = "partly-timed"

    def transcribe(self, _path, **_kw):
        return {
            "text": "first second",
            "segments": [
                {"start": 0.0, "end": 1.25, "text": "first"},
                {"start": 1.25, "end": None, "text": "second"},
            ],
            "language": "en",
        }


def _client(monkeypatch, backend):
    from fastapi.testclient import TestClient

    monkeypatch.setattr(
        "services.asr_backend.get_capture_asr_backend", lambda **_k: backend())
    monkeypatch.setattr(
        "services.asr_backend.get_active_asr_backend", lambda **_k: backend())
    monkeypatch.setattr(
        "services.asr_backend.load_active_asr_backend", lambda **_k: backend())

    from main import app
    return TestClient(app, client=("127.0.0.1", 50000))


def _post(client, **data):
    return client.post(
        "/transcribe",
        files={"audio": ("a.wav", b"\x00" * 32000, "audio/wav")},
        data=data,
    )


# The route picks its engine from a `mode` form field, not an `accurate`
# flag, so parametrising on `accurate` sent a field the route ignores and ran
# the default fast path twice. Both engines must pass a null end through.
@pytest.mark.parametrize("mode", ["fast", "accurate"])
def test_null_end_is_passed_through_not_rounded(monkeypatch, mode):
    client = _client(monkeypatch, _UntimedBackend)
    r = _post(client, mode=mode)
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["segments"][0]["end"] is None
    assert body["segments"][0]["start"] == 0.0
    assert body["segments"][0]["text"] == "the meeting is at three"
    # Nothing known to measure, so the duration stays 0 rather than becoming null.
    assert body["duration_s"] == 0.0


def test_duration_comes_from_the_timed_segments(monkeypatch):
    client = _client(monkeypatch, _PartlyTimedBackend)
    r = _post(client)
    assert r.status_code == 200, r.text
    body = r.json()
    assert [s["end"] for s in body["segments"]] == [1.25, None]
    assert body["duration_s"] == 1.25


# ── The live-dictation socket's own final-result builder ────────────────────
#
# Every existing capture_ws test stubs `_transcribe_buffer_full` out, so its
# response builder was never exercised. Call it directly: sherpa is the first
# capture engine and its `_sherpa_result` degrades to end=None, so this half is
# reachable through the default dictation path.

def test_ws_full_result_passes_null_end_through(monkeypatch):
    import asyncio

    from api.routers import capture_ws as cw

    monkeypatch.setattr(
        "services.asr_backend.get_capture_asr_backend",
        lambda **_k: _PartlyTimedBackend())

    async def _straight_through(_pool, run, **_kw):
        return run()

    monkeypatch.setattr(
        "services.asr_backend.run_transcribe_guarded", _straight_through)

    result = asyncio.run(
        cw._transcribe_buffer_full([b"\x00" * 32000], pcm_sr=16000))

    assert [s["end"] for s in result["segments"]] == [1.25, None]
    assert result["duration_s"] == 1.25