lukehemmin
7327145d7a
Batch+realtime transcription API: faster-whisper engine w/ EngineOwner (single GPU owner, OOM downgrade chain, persisted attempted_profiles), hardware-adaptive device manager (T0-T3 VRAM tiers), Redis/in-proc job queue w/ leases + crash recovery, postprocess (glossary/rules/LLM w/ egress guard), privacy-first result store (UUID keys, source deleted after transcribe), retention sweeper, API-key auth (HMAC digests, scopes, job ownership), WebSocket realtime lane (LocalAgreement), CLI (detect/transcribe/bench/serve/key), Docker, benchmark runner. 127 mock-based tests pass; ruff clean. Includes verification checklist and autoplan review notes.
86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
"""포맷 렌더링 (json/txt/srt/vtt) + 결과 스키마 계약 테스트."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from luke_scribe.results.formats import render
|
|
from luke_scribe.results.models import TranscriptResult
|
|
|
|
|
|
@pytest.fixture
|
|
def result(transcript_result: dict) -> TranscriptResult:
|
|
return TranscriptResult.model_validate(transcript_result)
|
|
|
|
|
|
class TestFormats:
|
|
def test_json_roundtrip(self, result: TranscriptResult):
|
|
out = render(result, "json")
|
|
data = json.loads(out)
|
|
assert data["schema_version"] == "1.0"
|
|
assert data["segments"][0]["text"].startswith("오늘")
|
|
assert data["text"] # 계약: text = segment 텍스트 연결
|
|
|
|
def test_txt(self, result: TranscriptResult):
|
|
out = render(result, "txt")
|
|
assert "오늘 API 서버에서" in out
|
|
assert out.endswith("\n")
|
|
|
|
def test_srt(self, result: TranscriptResult):
|
|
out = render(result, "srt")
|
|
assert "1" in out
|
|
assert "00:00:00,000 --> 00:00:02,500" in out
|
|
# 세그먼트 순번 + 타임스탬프 + 텍스트
|
|
assert "오늘 API 서버에서" in out
|
|
|
|
def test_vtt(self, result: TranscriptResult):
|
|
out = render(result, "vtt")
|
|
assert out.startswith("WEBVTT")
|
|
assert "00:00:00.000 --> 00:00:02.500" in out
|
|
|
|
def test_unsupported_format(self, result: TranscriptResult):
|
|
with pytest.raises(ValueError):
|
|
render(result, "xml") # type: ignore[arg-type]
|
|
|
|
def test_empty_segments_srt(self):
|
|
r = TranscriptResult(status="completed", segments=[], text="")
|
|
out = render(r, "srt")
|
|
assert out == ""
|
|
|
|
|
|
class TestSchemaContract:
|
|
def test_status_completed_no_error(self, transcript_result: dict):
|
|
result = TranscriptResult.model_validate(transcript_result)
|
|
assert result.error is None
|
|
|
|
def test_failed_gets_default_error(self):
|
|
r = TranscriptResult(status="failed", text="")
|
|
assert r.error is not None
|
|
assert r.error.code == "transcription_failed"
|
|
|
|
def test_cancelled_gets_default_error(self):
|
|
r = TranscriptResult(status="cancelled", text="")
|
|
assert r.error is not None
|
|
assert r.error.code == "cancelled"
|
|
|
|
def test_completed_error_rejected(self):
|
|
with pytest.raises(ValueError):
|
|
TranscriptResult(status="completed", error={"code": "x", "message": "y"}, text="")
|
|
|
|
def test_segment_index_must_be_nonneg(self):
|
|
with pytest.raises(ValueError):
|
|
TranscriptResult(
|
|
status="completed", segments=[{"index": -1, "start": 0, "end": 1, "text": "x"}]
|
|
)
|
|
|
|
def test_duration_gt_zero(self):
|
|
with pytest.raises(ValueError):
|
|
TranscriptResult(
|
|
status="completed",
|
|
normalized_audio={"duration_sec": 0},
|
|
segments=[],
|
|
text="",
|
|
)
|