lukehemmin
3dfa660503
Full HTML dashboard (dark theme, vanilla JS, no external deps) served at / and /dashboard with five tabs: system status (admin), file upload-to-transcribe with progress/result/downloads, job history with cancel/result modal, realtime mic demo over the existing WebSocket, and API key create/list. Backend: POST/GET /v1/keys (admin; raw key returned once, digest-only storage); KeyStore.list_keys(); EngineOwner.emit_hypothesis is now a real implementation (PCM16 chunk -> WAV -> realtime-lane decode with the single GPU lock) instead of a stub. Notebook: tunnel cell links /dashboard and smoke-checks the HTML. + 6 tests (dashboard public HTML, key create/list/scope, WAV header, emit_hypothesis cleanup). 153 tests pass, ruff clean, JS syntax verified with node --check.
277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""API 통합 테스트 — TestClient 기반 인증/업로드/소유권/큐 흐름.
|
|
|
|
실제 모델·Redis 없이 in-proc 브로커 + mock 결과로 동작한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from luke_scribe.api.app import create_app
|
|
from luke_scribe.config import Settings
|
|
|
|
|
|
@pytest.fixture
|
|
def settings(tmp_path) -> Settings:
|
|
return Settings(
|
|
_env_file=None,
|
|
results_dir=str(tmp_path / "results"),
|
|
api_key_file=str(tmp_path / "api_keys.json"),
|
|
api_keys="key-transcribe,key-admin:admin,transcribe",
|
|
queue_backend="inproc",
|
|
model_cache_dir=None,
|
|
tunnel="none",
|
|
max_queue=10,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(settings: Settings) -> TestClient:
|
|
app = create_app(settings)
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
|
|
class TestHealth:
|
|
def test_health_public(self, client: TestClient):
|
|
r = client.get("/health")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "ok"
|
|
assert body["model_ready"] is False # 모델 캐시 없음 → 프로비저닝 보류
|
|
|
|
|
|
class TestDashboard:
|
|
def test_dashboard_html_public(self, client: TestClient):
|
|
"""대시보드 HTML은 공개 — 인증은 클라이언트에서 API 키 입력."""
|
|
for path in ("/", "/dashboard"):
|
|
r = client.get(path)
|
|
assert r.status_code == 200
|
|
assert "text/html" in r.headers["content-type"]
|
|
assert "luke_scribe" in r.text
|
|
assert "실시간" in r.text # 전체 기능 포함
|
|
|
|
|
|
class TestKeys:
|
|
def test_create_and_list_key(self, client: TestClient):
|
|
headers = {"X-API-Key": "key-admin"}
|
|
r = client.post("/v1/keys", json={"scopes": ["transcribe"]}, headers=headers)
|
|
assert r.status_code == 201, r.text
|
|
body = r.json()
|
|
assert body["key"].startswith("luke-")
|
|
assert body["key_id"].startswith("k-")
|
|
# raw 키는 1회만 노출 — 목록에는 다이제스트 ID만
|
|
r2 = client.get("/v1/keys", headers=headers)
|
|
keys = r2.json()["keys"]
|
|
assert body["key_id"] in [k["id"] for k in keys]
|
|
assert all("key" not in k for k in keys)
|
|
|
|
def test_admin_scope_required(self, client: TestClient):
|
|
headers = {"X-API-Key": "key-transcribe"}
|
|
assert (
|
|
client.post("/v1/keys", json={"scopes": ["transcribe"]}, headers=headers).status_code
|
|
== 403
|
|
)
|
|
assert client.get("/v1/keys", headers=headers).status_code == 403
|
|
|
|
def test_no_auth_401(self, client: TestClient):
|
|
assert client.get("/v1/keys").status_code == 401
|
|
assert client.post("/v1/keys", json={"scopes": []}).status_code == 401
|
|
|
|
|
|
class TestAutoWorker:
|
|
def test_auto_worker_off_by_default(self, client: TestClient):
|
|
"""기본(False)에서는 워커 스레드를 띄우지 않는다 (테스트/프로덕션 안전)."""
|
|
assert client.app.state.inproc_worker is None
|
|
|
|
def test_auto_worker_started_when_enabled(self, tmp_path):
|
|
"""auto_worker=true + inproc → lifespan이 워커 스레드를 시작하고 종료 시 정지."""
|
|
from luke_scribe.api.app import create_app
|
|
from luke_scribe.config import Settings
|
|
|
|
settings = Settings(
|
|
_env_file=None,
|
|
results_dir=str(tmp_path / "results"),
|
|
api_key_file=str(tmp_path / "api_keys.json"),
|
|
queue_backend="inproc",
|
|
model_cache_dir=None,
|
|
tunnel="none",
|
|
auto_worker=True,
|
|
)
|
|
app = create_app(settings)
|
|
with TestClient(app) as c:
|
|
w = c.app.state.inproc_worker
|
|
assert w is not None
|
|
assert not w._stop.is_set()
|
|
# lifespan 종료 → 워커 정지 요청
|
|
assert w._stop.is_set()
|
|
|
|
|
|
class TestAuth:
|
|
def test_missing_key_rejected(self, client: TestClient):
|
|
r = client.get("/v1/jobs")
|
|
assert r.status_code == 401
|
|
|
|
def test_bad_key_rejected(self, client: TestClient):
|
|
r = client.get("/v1/jobs", headers={"X-API-Key": "wrong-key"})
|
|
assert r.status_code == 401
|
|
|
|
def test_admin_scope_enforced(self, client: TestClient):
|
|
r = client.get("/v1/system", headers={"X-API-Key": "key-transcribe"})
|
|
assert r.status_code == 403
|
|
r2 = client.get("/v1/system", headers={"X-API-Key": "key-admin"})
|
|
assert r2.status_code == 200
|
|
body = r2.json()
|
|
assert body["capability_tier"] in ("T0", "T1", "T2", "T3")
|
|
assert body["queue_depth"] == 0
|
|
|
|
|
|
class TestJobs:
|
|
def test_create_get_cancel_flow(self, client: TestClient, settings: Settings):
|
|
headers = {"X-API-Key": "key-transcribe"}
|
|
r = client.post(
|
|
"/v1/jobs",
|
|
files={"file": ("meeting.mp3", b"fake-audio-bytes", "audio/mpeg")},
|
|
data={"options": json.dumps({"language": "ko", "model": "large-v3-turbo"})},
|
|
headers=headers,
|
|
)
|
|
assert r.status_code == 202, r.text
|
|
body = r.json()
|
|
job_id = body["job_id"]
|
|
assert body["status"] == "queued"
|
|
assert body["queue_position"] == 0
|
|
|
|
# 조회
|
|
r = client.get(f"/v1/jobs/{job_id}", headers=headers)
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "queued"
|
|
|
|
# 결과는 아직 없음
|
|
r = client.get(f"/v1/jobs/{job_id}/result", headers=headers)
|
|
assert r.status_code == 409
|
|
|
|
# 취소
|
|
r = client.delete(f"/v1/jobs/{job_id}", headers=headers)
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "cancelled"
|
|
|
|
def test_ownership_enforced(self, client: TestClient):
|
|
"""Eng P14: 다른 키가 만든 job 조회/취소 불가."""
|
|
a = client.post(
|
|
"/v1/jobs",
|
|
files={"file": ("a.mp3", b"x", "audio/mpeg")},
|
|
data={"options": "{}"},
|
|
headers={"X-API-Key": "key-transcribe"},
|
|
)
|
|
job_id = a.json()["job_id"]
|
|
r = client.get(f"/v1/jobs/{job_id}", headers={"X-API-Key": "key-admin"})
|
|
assert r.status_code == 403
|
|
r = client.delete(f"/v1/jobs/{job_id}", headers={"X-API-Key": "key-admin"})
|
|
assert r.status_code == 403
|
|
|
|
def test_unknown_job_404(self, client: TestClient):
|
|
r = client.get(
|
|
"/v1/jobs/00000000-0000-0000-0000-000000000000",
|
|
headers={"X-API-Key": "key-transcribe"},
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
def test_bad_options_422(self, client: TestClient):
|
|
r = client.post(
|
|
"/v1/jobs",
|
|
files={"file": ("a.mp3", b"x", "audio/mpeg")},
|
|
data={"options": "not json {"},
|
|
headers={"X-API-Key": "key-transcribe"},
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
def test_bad_format_422(self, client: TestClient, settings: Settings):
|
|
headers = {"X-API-Key": "key-transcribe"}
|
|
r = client.post(
|
|
"/v1/jobs",
|
|
files={"file": ("a.mp3", b"x", "audio/mpeg")},
|
|
data={"options": "{}"},
|
|
headers=headers,
|
|
)
|
|
job_id = r.json()["job_id"]
|
|
r2 = client.get(f"/v1/jobs/{job_id}/result?format=docx", headers=headers)
|
|
assert r2.status_code == 422
|
|
|
|
def test_oversize_413(self, client: TestClient, settings: Settings):
|
|
settings.max_upload_bytes = 10
|
|
r = client.post(
|
|
"/v1/jobs",
|
|
files={"file": ("big.mp3", b"x" * 100, "audio/mpeg")},
|
|
data={"options": "{}"},
|
|
headers={"X-API-Key": "key-transcribe"},
|
|
)
|
|
assert r.status_code == 413
|
|
|
|
|
|
class TestQueueFull:
|
|
def test_queue_full_429(self, client: TestClient, settings: Settings):
|
|
settings.max_queue = 1
|
|
headers = {"X-API-Key": "key-transcribe"}
|
|
r1 = client.post(
|
|
"/v1/jobs",
|
|
files={"file": ("a.mp3", b"x", "audio/mpeg")},
|
|
data={"options": "{}"},
|
|
headers=headers,
|
|
)
|
|
assert r1.status_code == 202
|
|
r2 = client.post(
|
|
"/v1/jobs",
|
|
files={"file": ("b.mp3", b"x", "audio/mpeg")},
|
|
data={"options": "{}"},
|
|
headers=headers,
|
|
)
|
|
assert r2.status_code == 429
|
|
assert "Retry-After" in r2.headers
|
|
|
|
|
|
class TestResultEndpoint:
|
|
def test_result_after_processing(self, client: TestClient, settings: Settings, tmp_path):
|
|
"""워커가 결과를 저장한 뒤 result 엔드포인트가 JSON/SRT를 반환."""
|
|
from luke_scribe.results.models import TranscriptResult
|
|
from luke_scribe.results.store import ResultStore
|
|
|
|
headers = {"X-API-Key": "key-transcribe"}
|
|
r = client.post(
|
|
"/v1/jobs",
|
|
files={"file": ("a.mp3", b"x", "audio/mpeg")},
|
|
data={"options": "{}"},
|
|
headers=headers,
|
|
)
|
|
job_id = r.json()["job_id"]
|
|
|
|
# 워커 대신 결과를 직접 기록 (모델 없이 mock)
|
|
result = TranscriptResult(
|
|
status="completed",
|
|
source={"name": "a.mp3", "codec": "mp3", "size_bytes": 1},
|
|
normalized_audio={"duration_sec": 2.0},
|
|
execution={"model": "large-v3-turbo", "device": "cpu", "compute_type": "int8"},
|
|
text="테스트 전사 결과",
|
|
segments=[{"index": 0, "start": 0.0, "end": 1.0, "text": "테스트 전사 결과"}],
|
|
)
|
|
from luke_scribe.jobqueue.jobs import JobStatus
|
|
|
|
store = ResultStore(settings.results_dir)
|
|
store.write_result(job_id, result)
|
|
# 상태를 completed로 (워커가 했을 일) — queued→processing→completed
|
|
job = client.app.state.broker.get(job_id)
|
|
job.transition(JobStatus.PROCESSING)
|
|
job.transition(JobStatus.COMPLETED)
|
|
client.app.state.broker.save_meta(job)
|
|
|
|
r_json = client.get(f"/v1/jobs/{job_id}/result?format=json", headers=headers)
|
|
assert r_json.status_code == 200
|
|
assert json.loads(r_json.json()["content"])["text"] == "테스트 전사 결과"
|
|
|
|
r_srt = client.get(f"/v1/jobs/{job_id}/result?format=srt", headers=headers)
|
|
assert r_srt.status_code == 200
|
|
assert "WEBVTT" not in r_srt.json()["content"] # srt 포맷
|
|
assert "--> " in r_srt.json()["content"]
|