voice-transcriptor/tests/test_transcription.py

from pathlib import Path
from types import SimpleNamespace

import pytest

from voice_transcriptor.models import AppSettings
from voice_transcriptor.services.job_manifest import JobManifestRepository
from voice_transcriptor.services.preprocessing import CancellationToken, PreprocessingCancelled
from voice_transcriptor.services.transcription import OpenAITranscriptionClient, PermanentTranscriptionError, RetryPolicy, TranscriptionService, build_prompt, find_resumable_manifest, normalize_language


class Endpoint:
    def __init__(self, responses): self.responses = list(responses); self.calls = []
    def create(self, **kwargs):
        self.calls.append(kwargs); result = self.responses.pop(0)
        if isinstance(result, Exception): raise result
        return SimpleNamespace(text=result)


class StatusFailure(Exception):
    def __init__(self, status_code: int):
        super().__init__(f"sensitive sk-leaked-key status {status_code}")
        self.status_code = status_code; self.response = SimpleNamespace(headers={})


def make_job(tmp_path: Path) -> tuple[JobManifestRepository, Path]:
    job = tmp_path / "job"; (job / "chunks").mkdir(parents=True)
    for index in range(2): (job / "chunks" / f"chunk-{index:05d}.m4a").write_bytes(b"audio")
    repository = JobManifestRepository()
    created = repository.create(job, {"path": str(tmp_path / "source.mp3")}, {"model": "gpt-transcribe"}, [
        {"index": 0, "path": "chunks/chunk-00000.m4a", "source_start_seconds": "0", "source_end_seconds": "10", "duration_seconds": "10"},
        {"index": 1, "path": "chunks/chunk-00001.m4a", "source_start_seconds": "9", "source_end_seconds": "20", "duration_seconds": "11"},
    ])
    return repository, Path(created["manifest_path"])


def test_brazilian_language_and_prompt_preserve_spoken_language() -> None:
    prompt = build_prompt("  AWS, PostgreSQL, Brasília  ")
    assert normalize_language("pt-BR") == "pt"; assert normalize_language("pt_BR") == "pt"; assert normalize_language("en-US") == "en"
    assert "não traduza" in prompt.lower(); assert "números" in prompt.lower(); assert prompt.endswith("AWS, PostgreSQL, Brasília")


def test_adapter_calls_verified_audio_transcriptions_interface(tmp_path: Path) -> None:
    audio = tmp_path / "chunk.m4a"; audio.write_bytes(b"audio")
    endpoint = Endpoint(["Olá, Brasília."]); client = SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))
    text = OpenAITranscriptionClient(client).transcribe(audio, "future-compatible-model", "pt-BR", "vocabulário")
    assert text == "Olá, Brasília."
    call = endpoint.calls[0]
    assert call["model"] == "future-compatible-model"; assert call["language"] == "pt"; assert call["prompt"] == "vocabulário"; assert call["response_format"] == "json"; assert call["file"].closed is True


def test_run_saves_each_chunk_and_resume_skips_completed(tmp_path: Path) -> None:
    repository, manifest_path = make_job(tmp_path); first_endpoint = Endpoint(["Primeiro", "Segundo"])
    service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=first_endpoint))), repository)
    settings = AppSettings("gpt-transcribe", "pt-BR", tmp_path, context_vocabulary="Pix")
    transcript = service.run(manifest_path, settings)
    assert transcript.read_text(encoding="utf-8") == "Primeiro\nSegundo\n"
    manifest = repository.load(manifest_path)
    assert manifest["state"] == "completed"; assert [item["status"] for item in manifest["chunks"]] == ["completed", "completed"]; assert manifest["chunks"][1]["source_start_seconds"] == "9"
    resume_endpoint = Endpoint([])
    TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=resume_endpoint))), repository).run(manifest_path, settings)
    assert resume_endpoint.calls == []


def test_transient_failure_retries_with_exponential_backoff(tmp_path: Path) -> None:
    repository, manifest_path = make_job(tmp_path); endpoint = Endpoint([StatusFailure(429), StatusFailure(503), "Primeiro", "Segundo"]); delays = []
    service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository, retry_policy=RetryPolicy(max_attempts=3, initial_delay_seconds=1, max_delay_seconds=10, jitter_ratio=0), sleep=delays.append)
    service.run(manifest_path, AppSettings("gpt-transcribe", "pt-BR", tmp_path))
    assert delays == [1, 2]; assert len(endpoint.calls) == 4


def test_permanent_api_failure_is_not_retried_and_is_sanitized(tmp_path: Path) -> None:
    repository, manifest_path = make_job(tmp_path); endpoint = Endpoint([StatusFailure(400)])
    service = TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository)
    with pytest.raises(PermanentTranscriptionError) as caught: service.run(manifest_path, AppSettings("bad-model", "pt-BR", tmp_path))
    assert len(endpoint.calls) == 1; assert "sk-leaked-key" not in str(caught.value); assert "sk-leaked-key" not in manifest_path.read_text(encoding="utf-8")


def test_cancelled_job_makes_no_api_request(tmp_path: Path) -> None:
    repository, manifest_path = make_job(tmp_path); endpoint = Endpoint(["unexpected"]); token = CancellationToken(); token.cancel()
    with pytest.raises(PreprocessingCancelled): TranscriptionService(OpenAITranscriptionClient(SimpleNamespace(audio=SimpleNamespace(transcriptions=endpoint))), repository).run(manifest_path, AppSettings("gpt-transcribe", "pt-BR", tmp_path), token=token)
    assert endpoint.calls == []; assert repository.load(manifest_path)["state"] == "cancelled"


def test_find_resumable_manifest_matches_source_and_effective_settings(tmp_path: Path) -> None:
    repository, manifest_path = make_job(tmp_path)
    manifest = repository.load(manifest_path)
    manifest["settings"] = {"model": "gpt-transcribe", "language": "pt-BR", "context_vocabulary": "Pix"}
    repository.save(manifest_path, manifest)
    settings = AppSettings("gpt-transcribe", "pt-BR", tmp_path, context_vocabulary="Pix")

    assert find_resumable_manifest(tmp_path, Path(manifest["source"]["path"]), settings) == manifest_path
    assert find_resumable_manifest(tmp_path, Path(manifest["source"]["path"]), AppSettings("other", "pt-BR", tmp_path, context_vocabulary="Pix")) is None