luke_scribe/tests/unit/test_engine_faster_whisper.py

"""FasterWhisperEngine 단위 테스트 — CTranslate2 호출 계약 (device 분리).

faster-whisper/CTranslate2는 ``device`` 문자열로 "cuda"만 허용하며 인덱스는
별도 ``device_index`` 인자로 받는다. DeviceManager가 내려주는 "cuda:N"을
제대로 분리해 전달하는지 mock으로 검증한다 (GPU 환경에서만 재현되는 버그
— Colab 실전 테스트에서 ``unsupported device cuda:0``으로 확인됨).
"""

from __future__ import annotations

import sys
import types

from luke_scribe.engine.base import TranscriptionOptions
from luke_scribe.engine.faster_whisper_engine import FasterWhisperEngine


class FakeWhisperModel:
    """faster_whisper.WhisperModel 대체 — 생성 인자/세그먼트/info를 기록한다."""

    calls: list[dict] = []
    segments: list = []  # transcribe()가 yield할 세그먼트 (기본: namedtuple)
    info: object = None  # 기본: TranscriptionInfo(namedtuple) 흉내

    def __init__(self, *args, **kwargs) -> None:
        self.kwargs = kwargs
        self.__class__.calls.append(kwargs)

    def transcribe(self, audio_path, **kwargs):
        return iter(list(self.__class__.segments)), self.__class__.info


def _install_fake(monkeypatch) -> None:
    mod = types.ModuleType("faster_whisper")
    mod.WhisperModel = FakeWhisperModel
    monkeypatch.setitem(sys.modules, "faster_whisper", mod)
    FakeWhisperModel.calls.clear()
    FakeWhisperModel.segments = []
    FakeWhisperModel.info = None


def _opts(**kw) -> TranscriptionOptions:
    kw.setdefault("model", "large-v3-turbo")
    kw.setdefault("device", "cuda:0")
    kw.setdefault("compute_type", "float16")
    return TranscriptionOptions(**kw)


def test_cuda_device_index_split(monkeypatch):
    """'cuda:0' → device='cuda', device_index=0 (Colab A100 첫 GPU)."""
    _install_fake(monkeypatch)
    FasterWhisperEngine().transcribe("/tmp/x.wav", _opts())
    kwargs = FakeWhisperModel.calls[-1]
    assert kwargs["device"] == "cuda"
    assert kwargs["device_index"] == 0
    assert kwargs["compute_type"] == "float16"


def test_cuda_second_device_index(monkeypatch):
    """'cuda:1' → device='cuda', device_index=1."""
    _install_fake(monkeypatch)
    FasterWhisperEngine().transcribe("/tmp/x.wav", _opts(device="cuda:1"))
    kwargs = FakeWhisperModel.calls[-1]
    assert kwargs["device"] == "cuda"
    assert kwargs["device_index"] == 1


def test_cpu_passthrough(monkeypatch):
    """'cpu' → 그대로 device='cpu', device_index=0."""
    _install_fake(monkeypatch)
    FasterWhisperEngine().transcribe("/tmp/x.wav", _opts(device="cpu", compute_type="int8"))
    kwargs = FakeWhisperModel.calls[-1]
    assert kwargs["device"] == "cpu"
    assert kwargs["device_index"] == 0
    assert kwargs["compute_type"] == "int8"


def test_split_device_helper():
    assert FasterWhisperEngine._split_device("cuda:0") == ("cuda", 0)
    assert FasterWhisperEngine._split_device("cuda:3") == ("cuda", 3)
    assert FasterWhisperEngine._split_device("cpu") == ("cpu", 0)
    assert FasterWhisperEngine._split_device("cuda") == ("cuda", 0)
    # 파싱 불가 인덱스는 그대로 전달 (모델 로드 시 명시적 오류로 fail-explicit)
    assert FasterWhisperEngine._split_device("cuda:xx") == ("cuda:xx", 0)


def test_namedtuple_segments_normalized_to_dicts(monkeypatch):
    """faster-whisper Segment(namedtuple) → dict 정규화 (GPU 실전 버그)."""
    from collections import namedtuple

    _install_fake(monkeypatch)
    Seg = namedtuple(
        "Segment",
        [
            "id",
            "seek",
            "start",
            "end",
            "text",
            "tokens",
            "temperature",
            "avg_logprob",
            "compression_ratio",
            "no_speech_prob",
        ],
    )
    FakeWhisperModel.segments = [
        Seg(
            id=0,
            seek=0,
            start=0.0,
            end=2.5,
            text="오늘 vLLM을 배포합니다.",
            tokens=[1, 2],
            temperature=0.0,
            avg_logprob=-0.2,
            compression_ratio=1.0,
            no_speech_prob=0.01,
        )
    ]
    outcome = FasterWhisperEngine().transcribe(
        "/tmp/x.wav", _opts(device="cpu", compute_type="int8")
    )
    segs = list(outcome.segments)
    assert len(segs) == 1
    seg = segs[0]
    # dict 계약: .get() 사용 가능 (파이프라인이 이걸로 접근)
    assert seg.get("text") == "오늘 vLLM을 배포합니다."
    assert seg.get("end") == 2.5
    assert seg.get("avg_logprob") == -0.2


def test_dict_segments_passthrough(monkeypatch):
    """이미 dict인 세그먼트는 그대로 (mock 계약과 호환)."""
    _install_fake(monkeypatch)
    FakeWhisperModel.segments = [{"index": 0, "start": 0.0, "end": 1.0, "text": "x"}]
    outcome = FasterWhisperEngine().transcribe(
        "/tmp/x.wav", _opts(device="cpu", compute_type="int8")
    )
    segs = list(outcome.segments)
    assert segs == [{"index": 0, "start": 0.0, "end": 1.0, "text": "x"}]


def test_namedtuple_info_normalized_to_dict(monkeypatch):
    """faster-whisper TranscriptionInfo(namedtuple) → dict (GPU 실전 버그)."""
    from collections import namedtuple

    _install_fake(monkeypatch)
    Info = namedtuple(
        "TranscriptionInfo",
        ["language", "language_probability", "duration", "duration_after_vad"],
    )
    FakeWhisperModel.info = Info(
        language="ko", language_probability=0.99, duration=10.464, duration_after_vad=9.088
    )
    outcome = FasterWhisperEngine().transcribe(
        "/tmp/x.wav", _opts(device="cpu", compute_type="int8")
    )
    # dict 계약: .get() 사용 가능 (batch.py가 이걸로 접근)
    assert outcome.info.get("language") == "ko"
    assert outcome.info.get("duration") == 10.464


def test_dict_info_passthrough(monkeypatch):
    """이미 dict인 info는 그대로 (mock 계약과 호환)."""
    _install_fake(monkeypatch)
    FakeWhisperModel.info = {"language": "ko"}
    outcome = FasterWhisperEngine().transcribe(
        "/tmp/x.wav", _opts(device="cpu", compute_type="int8")
    )
    assert outcome.info == {"language": "ko"}