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.
113 lines
4.3 KiB
Python
113 lines
4.3 KiB
Python
"""후처리 단위 테스트 — glossary/rules/confidence/span-aware."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from luke_scribe.config import Settings
|
|
from luke_scribe.engine.base import TranscriptionOptions
|
|
from luke_scribe.postprocess.confidence import confidence_flags, segment_confidence
|
|
from luke_scribe.postprocess.glossary import apply_glossary
|
|
from luke_scribe.postprocess.llm import EgressGuard
|
|
from luke_scribe.postprocess.pipeline import run_postprocess
|
|
from luke_scribe.postprocess.rules import apply_rules
|
|
from luke_scribe.results.models import Segment
|
|
|
|
|
|
def _segments(texts: list[str], logprobs: list[float] | None = None) -> list[Segment]:
|
|
out = []
|
|
t = 0.0
|
|
for i, text in enumerate(texts):
|
|
lp = logprobs[i] if logprobs else -0.2
|
|
out.append(
|
|
Segment(index=i, start=t, end=t + 1.0, text=text, avg_logprob=lp, no_speech_prob=0.01)
|
|
)
|
|
t += 1.0
|
|
return out
|
|
|
|
|
|
class TestGlossary:
|
|
def test_restores_romanized(self):
|
|
segs = _segments(["브이엘엘엠 서버를 배포했습니다"])
|
|
out = apply_glossary(segs, {"브이엘엘엠": "vLLM"})
|
|
assert out["corrections"][0]["to"] == "vLLM"
|
|
assert "vLLM" in out["segments"][0].text
|
|
|
|
def test_no_double_match(self):
|
|
segs = _segments(["API API"])
|
|
out = apply_glossary(segs, {"API": "API"})
|
|
assert len(out["corrections"]) == 0 # 이미 표준 표기, 교정 없음
|
|
|
|
|
|
class TestRules:
|
|
def test_vllm_spaced(self):
|
|
segs = _segments(["v l l m 추론 서버"])
|
|
out = apply_rules(segs)
|
|
assert "vLLM" in out["segments"][0].text
|
|
|
|
def test_whitespace_collapse(self):
|
|
segs = _segments(["오늘 API 서버"])
|
|
out = apply_rules(segs)
|
|
assert out["segments"][0].text == "오늘 API 서버"
|
|
|
|
|
|
class TestConfidence:
|
|
def test_low_logprob_flagged(self):
|
|
segs = _segments(["불확실한 전사"], logprobs=[-2.0])
|
|
assert segs[0].index in confidence_flags(segs)
|
|
|
|
def test_high_confidence_not_flagged(self):
|
|
segs = _segments(["확실한 전사"], logprobs=[-0.1])
|
|
assert confidence_flags(segs) == []
|
|
|
|
def test_segment_confidence_range(self):
|
|
assert (
|
|
segment_confidence(Segment(index=0, start=0, end=1, text="x", avg_logprob=-0.5)) == 0.5
|
|
)
|
|
|
|
|
|
class TestPipeline:
|
|
def test_rules_mode_applies(self):
|
|
segs = _segments(["v l l m 서버"])
|
|
settings = Settings(_env_file=None, post_mode="rules", post_enabled=True)
|
|
out = run_postprocess(segs, TranscriptionOptions(), settings)
|
|
assert any("vLLM" in s.text for s in out["segments"])
|
|
assert out["meta"]["mode"] == "rules"
|
|
|
|
def test_none_mode_unchanged(self):
|
|
segs = _segments(["v l l m 서버"])
|
|
settings = Settings(_env_file=None, post_mode="none", post_enabled=False)
|
|
out = run_postprocess(segs, TranscriptionOptions(), settings)
|
|
assert out["segments"][0].text == "v l l m 서버"
|
|
|
|
def test_span_aware_keeps_timestamps(self):
|
|
# 교정 후에도 세그먼트 타임라인(start/end)은 보존
|
|
segs = _segments(["브이엘엘엠 서버"])
|
|
before = [(s.start, s.end) for s in segs]
|
|
out = apply_glossary(segs, {"브이엘엘엠": "vLLM"})
|
|
after = [(s.start, s.end) for s in out["segments"]]
|
|
assert before == after
|
|
|
|
|
|
class TestEgressGuard:
|
|
def test_private_ip_rejected(self):
|
|
guard = EgressGuard({"llm": "http://192.168.0.1:8080/v1"})
|
|
assert guard.resolve("llm") is None
|
|
|
|
def test_http_scheme_rejected(self):
|
|
guard = EgressGuard({"llm": "http://example.com/v1"})
|
|
assert guard.resolve("llm") is None
|
|
|
|
def test_https_public_allowed(self, monkeypatch):
|
|
# 오프라인 테스트 환경에서 DNS 해석을 공용 IP로 모킹
|
|
monkeypatch.setattr(
|
|
"luke_scribe.postprocess.llm.socket.gethostbyname", lambda host: "93.184.216.34"
|
|
)
|
|
guard = EgressGuard({"llm": "https://api.example.com/v1"})
|
|
assert guard.resolve("llm") is not None
|
|
|
|
def test_private_hostname_resolved_rejected(self, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"luke_scribe.postprocess.llm.socket.gethostbyname", lambda host: "10.0.0.5"
|
|
)
|
|
guard = EgressGuard({"llm": "https://internal.example.com/v1"})
|
|
assert guard.resolve("llm") is None
|