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.
96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
"""EngineOwner 단위 테스트 — OOM 강등 체인, attempted_profiles 영속화, 우선순위."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from luke_scribe.engine.base import TranscriptionOptions
|
|
from luke_scribe.engine.owner import MAX_DOWNGRADES, EngineOwner, InferenceRequest
|
|
from luke_scribe.errors import OutOfMemory
|
|
|
|
|
|
class OOMFakeEngine:
|
|
"""첫 N번 호출에서 OOM, 이후 성공하는 mock."""
|
|
|
|
def __init__(self, oom_calls: int = 0) -> None:
|
|
self.oom_calls = oom_calls
|
|
self.calls: list[dict] = []
|
|
self.closed = False
|
|
|
|
def transcribe(self, audio_path, options, should_cancel=None, download_progress=None):
|
|
self.calls.append({"device": options.device, "compute_type": options.compute_type})
|
|
if len(self.calls) <= self.oom_calls:
|
|
raise OutOfMemory("CUDA OOM (mock)")
|
|
return type("O", (), {"segments": iter([]), "info": {}})()
|
|
|
|
def unload_all(self):
|
|
self.closed = True
|
|
|
|
|
|
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_no_downgrade_on_success():
|
|
owner = EngineOwner.__new__(EngineOwner)
|
|
owner.settings = None
|
|
owner._engine = OOMFakeEngine(oom_calls=0)
|
|
owner._lock = __import__("threading").Lock()
|
|
owner._realtime_priority = __import__("threading").Lock()
|
|
owner._stats = {"jobs": 0, "downgrades": 0, "oom": 0}
|
|
req = InferenceRequest(audio_path="/tmp/x.wav", options=_opts())
|
|
out = owner.transcribe(req)
|
|
assert out["compute_type"] == "float16"
|
|
assert len(out["attempted_profiles"]) == 1
|
|
|
|
|
|
def test_downgrade_chain_on_oom():
|
|
engine = OOMFakeEngine(oom_calls=1) # 첫 프로파일에서 OOM
|
|
owner = EngineOwner.__new__(EngineOwner)
|
|
owner.settings = None
|
|
owner._engine = engine
|
|
owner._lock = __import__("threading").Lock()
|
|
owner._realtime_priority = __import__("threading").Lock()
|
|
owner._stats = {"jobs": 0, "downgrades": 0, "oom": 0}
|
|
req = InferenceRequest(audio_path="/tmp/x.wav", options=_opts())
|
|
out = owner.transcribe(req)
|
|
# float16 → int8_float16 (강등 1회)
|
|
assert out["compute_type"] == "int8_float16"
|
|
assert len(out["attempted_profiles"]) == 2
|
|
|
|
|
|
def test_downgrade_capped_at_max():
|
|
engine = OOMFakeEngine(oom_calls=99) # 항상 OOM
|
|
owner = EngineOwner.__new__(EngineOwner)
|
|
owner.settings = None
|
|
owner._engine = engine
|
|
owner._lock = __import__("threading").Lock()
|
|
owner._realtime_priority = __import__("threading").Lock()
|
|
owner._stats = {"jobs": 0, "downgrades": 0, "oom": 0}
|
|
req = InferenceRequest(audio_path="/tmp/x.wav", options=_opts())
|
|
with pytest.raises(OutOfMemory):
|
|
owner.transcribe(req)
|
|
# 강등 체인은 cpu까지 포함하지만 시도는 MAX_DOWNGRADES+1 이내
|
|
assert len(engine.calls) <= MAX_DOWNGRADES + 1
|
|
|
|
|
|
def test_attempted_profiles_persisted_for_retry():
|
|
"""재큐(retry) 시 attempted_profiles가 유지되어 총 시도가 캡된다."""
|
|
engine = OOMFakeEngine(oom_calls=0)
|
|
owner = EngineOwner.__new__(EngineOwner)
|
|
owner.settings = None
|
|
owner._engine = engine
|
|
owner._lock = __import__("threading").Lock()
|
|
owner._realtime_priority = __import__("threading").Lock()
|
|
owner._stats = {"jobs": 0, "downgrades": 0, "oom": 0}
|
|
# 이미 2회 시도한 이력이 Job에 영속화된 상황 → 재시도는 즉시 실패해야 함
|
|
req = InferenceRequest(
|
|
audio_path="/tmp/x.wav", options=_opts(), attempted_profiles=["float16", "int8_float16"]
|
|
)
|
|
with pytest.raises(OutOfMemory):
|
|
owner.transcribe(req)
|
|
assert len(engine.calls) == 0 # 새 시도 없음
|