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.
107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
"""Job 상태 머신 + in-proc 브로커 단위 테스트."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from luke_scribe.errors import JobAlreadyTerminal, QueueFull
|
|
from luke_scribe.jobqueue.broker import InProcBroker
|
|
from luke_scribe.jobqueue.jobs import Job, JobStatus
|
|
|
|
|
|
class TestJobStateMachine:
|
|
def test_legal_transitions(self):
|
|
job = Job()
|
|
assert job.status == JobStatus.QUEUED
|
|
job.transition(JobStatus.PROCESSING)
|
|
assert job.status == JobStatus.PROCESSING
|
|
job.transition(JobStatus.COMPLETED)
|
|
assert job.status == JobStatus.COMPLETED
|
|
|
|
def test_illegal_transition_rejected(self):
|
|
job = Job()
|
|
job.transition(JobStatus.PROCESSING)
|
|
job.transition(JobStatus.COMPLETED)
|
|
with pytest.raises(JobAlreadyTerminal):
|
|
job.transition(JobStatus.PROCESSING) # terminal에서 되돌아갈 수 없음
|
|
|
|
def test_queued_cancel(self):
|
|
job = Job()
|
|
job.transition(JobStatus.CANCELLED)
|
|
assert job.status == JobStatus.CANCELLED
|
|
assert job.cancelled_at is not None
|
|
|
|
def test_lease_expiry(self):
|
|
job = Job()
|
|
job.refresh_lease(60.0, now=1000.0)
|
|
assert not job.lease_expired(now=1050.0)
|
|
assert job.lease_expired(now=1100.0)
|
|
|
|
def test_attempted_profiles_persisted(self):
|
|
job = Job(attempted_profiles=["float16", "int8_float16"])
|
|
assert job.to_meta()["attempted_profiles"] == ["float16", "int8_float16"]
|
|
restored = Job.from_meta(job.to_meta())
|
|
assert restored.attempted_profiles == ["float16", "int8_float16"]
|
|
|
|
|
|
class TestInProcBroker:
|
|
def test_enqueue_claim_complete(self):
|
|
broker = InProcBroker()
|
|
job = Job()
|
|
broker.enqueue(job)
|
|
claimed = broker.claim_next("w1", 60.0)
|
|
assert claimed is not None
|
|
assert claimed.status == JobStatus.PROCESSING
|
|
broker.transition(claimed, JobStatus.COMPLETED)
|
|
assert broker.get(job.id).status == JobStatus.COMPLETED
|
|
|
|
def test_queue_position(self):
|
|
broker = InProcBroker()
|
|
a, b = Job(), Job()
|
|
broker.enqueue(a)
|
|
broker.enqueue(b)
|
|
assert a.queue_position == 0
|
|
assert b.queue_position == 1
|
|
|
|
def test_queue_full(self):
|
|
broker = InProcBroker()
|
|
broker.settings.max_queue = 2
|
|
broker.enqueue(Job())
|
|
broker.enqueue(Job())
|
|
with pytest.raises(QueueFull):
|
|
broker.enqueue(Job())
|
|
|
|
def test_cancel_queued(self):
|
|
broker = InProcBroker()
|
|
job = Job()
|
|
broker.enqueue(job)
|
|
cancelled = broker.cancel(job.id)
|
|
assert cancelled.status == JobStatus.CANCELLED
|
|
assert broker.queue_depth() == 0
|
|
|
|
def test_cancel_processing_sets_flag(self):
|
|
broker = InProcBroker()
|
|
job = Job()
|
|
broker.enqueue(job)
|
|
broker.claim_next("w1", 60.0)
|
|
cancelled = broker.cancel(job.id)
|
|
assert cancelled.cancel_requested is True
|
|
assert cancelled.status == JobStatus.PROCESSING # 워커가 세그먼트 경계에서 종료
|
|
|
|
def test_reconcile_stale_recovers_crash(self):
|
|
"""Eng 리뷰 P11: 워커 크래시 → stale processing → failed 복구."""
|
|
broker = InProcBroker()
|
|
job = Job()
|
|
broker.enqueue(job)
|
|
broker.claim_next("w1", 60.0) # processing + lease 60s
|
|
# lease 만료 시뮬레이션
|
|
job2 = broker.get(job.id)
|
|
job2.lease_expires_at = time.time() - 1
|
|
broker.save_meta(job2)
|
|
reclaimed = broker.reconcile_stale(60.0)
|
|
assert job.id in reclaimed
|
|
assert broker.get(job.id).status == JobStatus.FAILED
|
|
assert broker.get(job.id).error_code == "worker_crash"
|