"""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 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"]