luke_scribe/tests/unit/test_postprocess.py

"""후처리 단위 테스트 — 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_vllm_restored_from_blm(self):
        # GPU 실전에서 재현된 오인식: vLLM → BLM
        segs = _segments(["오늘은 BLM 서버를 배포합니다"])
        out = apply_rules(segs)
        assert "vLLM" in out["segments"][0].text
        assert "BLM" not in out["segments"][0].text

    def test_blm_boundary_required(self):
        # 단어 경계가 없으면 교정하지 않는다 (부분 문자열 보호)
        segs = _segments(["sublm 단어"])
        out = apply_rules(segs)
        assert "sublm" in out["segments"][0].text
        assert "vLLM" not in out["segments"][0].text

    def test_blm_case_insensitive(self):
        segs = _segments(["blm 서버"])
        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