"""ResultStore — UUID 키, 경로 트래버설 차단, 원자적 쓰기, 보관 TTL 테스트.""" from __future__ import annotations import os import time import uuid import pytest from luke_scribe.errors import OutputWriteError from luke_scribe.results.models import TranscriptResult from luke_scribe.results.retention import RetentionSweeper from luke_scribe.results.store import ResultStore @pytest.fixture def store(tmp_path) -> ResultStore: return ResultStore(str(tmp_path / "results")) @pytest.fixture def result(transcript_result: dict) -> TranscriptResult: return TranscriptResult.model_validate(transcript_result) class TestResultStore: def test_write_read_roundtrip(self, store: ResultStore, result: TranscriptResult): job_id = str(uuid.uuid4()) store.write_result(job_id, result) read = store.read_result(job_id) assert read is not None assert read.text == result.text assert read.segments[0].text == result.segments[0].text def test_missing_result_returns_none(self, store: ResultStore): assert store.read_result(str(uuid.uuid4())) is None def test_path_traversal_rejected(self, store: ResultStore): with pytest.raises(ValueError): store._job_dir("../../etc/passwd") with pytest.raises(ValueError): store._job_dir("..") def test_source_path_sanitized(self, store: ResultStore): job_id = str(uuid.uuid4()) p = store.source_path_for(job_id, "../../../evil.mp3") assert ".." not in str(p) assert p.name == "evil.mp3" def test_symlink_job_dir_rejected(self, store: ResultStore, result: TranscriptResult): job_id = str(uuid.uuid4()) real = store.root / f"real-{uuid.uuid4().hex}" real.mkdir(parents=True) link = store.root / job_id os.symlink(real, link) with pytest.raises(ValueError): store.write_result(job_id, result) def test_atomic_write_no_partial_json(self, store: ResultStore, result: TranscriptResult): job_id = str(uuid.uuid4()) target = store.write_result(job_id, result) # 임시 파일이 남지 않아야 한다 leftovers = [p for p in target.parent.iterdir() if p.name.startswith(".result-")] assert leftovers == [] def test_delete_job(self, store: ResultStore, result: TranscriptResult): job_id = str(uuid.uuid4()) store.write_result(job_id, result) store.delete_job(job_id) assert store.read_result(job_id) is None def test_iter_job_dirs_includes_all(self, store: ResultStore, result: TranscriptResult): done_id = str(uuid.uuid4()) store.write_result(done_id, result) empty_id = str(uuid.uuid4()) store._job_dir(empty_id).mkdir(parents=True) entries = dict(store.iter_job_dirs()) assert done_id in entries assert empty_id in entries def test_delete_derived_keeps_result_and_meta( self, store: ResultStore, result: TranscriptResult ): job_id = str(uuid.uuid4()) src = store.source_path_for(job_id, "meeting.mp3") src.write_bytes(b"audio") store.write_result(job_id, result) store.write_source_metadata(job_id, {"status": "completed", "completed_at": 1.0}) store.delete_derived(job_id) assert not src.exists() # 원본 오디오 삭제 (privacy-first) assert store.read_result(job_id) is not None # 결과 보존 assert (store._job_dir(job_id) / "meta.json").exists() class TestRetentionSweeper: def _write_terminal( self, store: ResultStore, result: TranscriptResult, completed_at: float ) -> str: job_id = str(uuid.uuid4()) store.write_result(job_id, result) store.write_source_metadata(job_id, {"completed_at": completed_at}) return job_id def test_sweeps_stale_terminal(self, store: ResultStore, result: TranscriptResult): now = time.time() old = self._write_terminal(store, result, completed_at=now - 8 * 86400) fresh = self._write_terminal(store, result, completed_at=now - 3600) sweeper = RetentionSweeper(store, retention_days=7, now=now) removed = sweeper.sweep() assert old in removed assert fresh not in removed assert store.read_result(old) is None assert store.read_result(fresh) is not None def test_iso_timestamp_parsed(self, store: ResultStore, result: TranscriptResult): """ISO 8601 타임스탬프도 처리 (워커는 float, 다른 경로는 ISO 가능).""" from datetime import UTC, datetime now = time.time() iso_old = datetime.fromtimestamp(now - 8 * 86400, tz=UTC).isoformat() job_id = str(uuid.uuid4()) store.write_result(job_id, result) store.write_source_metadata(job_id, {"completed_at": iso_old}) removed = RetentionSweeper(store, retention_days=7, now=now).sweep() assert job_id in removed def test_mtime_fallback(self, store: ResultStore, result: TranscriptResult): """메타가 없으면 result.json mtime 기준 폴백.""" job_id = str(uuid.uuid4()) p = store.write_result(job_id, result) old = time.time() - 8 * 86400 os.utime(p, (old, old)) removed = RetentionSweeper(store, retention_days=7, now=time.time()).sweep() assert job_id in removed def test_retention_disabled(self, store: ResultStore, result: TranscriptResult): job_id = self._write_terminal(store, result, completed_at=time.time() - 8 * 86400) sweeper = RetentionSweeper(store, retention_days=0, now=time.time()) assert sweeper.sweep() == [] assert store.read_result(job_id) is not None def test_sweeps_failed_job_with_meta_only(self, store: ResultStore, result: TranscriptResult): """Eng 리뷰: failed/cancelled job(meta.json만)도 보관 정리 대상.""" now = time.time() failed_id = str(uuid.uuid4()) store._job_dir(failed_id).mkdir(parents=True) store.write_source_metadata( failed_id, {"status": "failed", "error_code": "worker_crash", "failed_at": now - 8 * 86400}, ) removed = RetentionSweeper(store, retention_days=7, now=now).sweep() assert failed_id in removed assert not store._job_dir(failed_id).exists() def test_processing_job_not_swept(self, store: ResultStore, result: TranscriptResult): """queued/processing(터미널 타임스탬프 없음)은 절대 삭제 안 함.""" now = time.time() queued_id = str(uuid.uuid4()) store._job_dir(queued_id).mkdir(parents=True) store.write_source_metadata(queued_id, {"status": "queued"}) removed = RetentionSweeper(store, retention_days=7, now=now).sweep() assert queued_id not in removed assert store._job_dir(queued_id).exists() class TestAtomicFileWriter: def test_write_replaces(self, tmp_path): from luke_scribe.results.store import AtomicFileWriter target = tmp_path / "out.txt" AtomicFileWriter.write(target, "hello") assert target.read_text() == "hello" AtomicFileWriter.write(target, "world") assert target.read_text() == "world" def test_write_failure_raises_output_error(self, tmp_path): from luke_scribe.results.store import AtomicFileWriter # 부모 경로가 파일이면 mkdir 실패 → OutputWriteError blocker = tmp_path / "blocker" blocker.write_text("not a dir") with pytest.raises(OutputWriteError): AtomicFileWriter.write(blocker / "x.txt", "data")