feat: implement full-platform STT API (v2.3 consensus plan)
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.
This commit is contained in:
@@ -0,0 +1,85 @@
|
||||
"""포맷 렌더링 (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="",
|
||||
)
|
||||
Reference in New Issue
Block a user