"""AudioIngestor — ffprobe 검증, 상한(크기/길이), 취소·임시파일 정리 테스트. ffprobe/ffmpeg 바이너리 없이 subprocess를 mock해서 동작을 검증한다. """ from __future__ import annotations import json import subprocess from pathlib import Path import pytest from luke_scribe.audio.ingest import AudioIngestor from luke_scribe.config import Settings from luke_scribe.errors import ( AudioProbeFailed, CancelledError, InvalidInput, UnsupportedInputEnvelope, ) @pytest.fixture def settings(tmp_path) -> Settings: return Settings( _env_file=None, max_duration_sec=14400, max_upload_bytes=2 * 1024 * 1024 * 1024, ) def _fake_probe_json(*, duration: str = "120.0", has_audio: bool = True) -> str: streams = ( [{"codec_type": "audio", "codec_name": "mp3"}] if has_audio else [{"codec_type": "video"}] ) return json.dumps({"format": {"duration": duration}, "streams": streams}) class FakeSubprocess: """ffprobe 호출에 대한 확정적 mock.""" def __init__( self, *, probe_json: str | None = None, probe_rc: int = 0, probe_err: str = "" ) -> None: self.probe_json = probe_json self.probe_rc = probe_rc self.probe_err = probe_err self.calls: list[list[str]] = [] def run(self, cmd, **kwargs): self.calls.append(cmd) if "show_entries" in cmd: return subprocess.CompletedProcess( cmd, self.probe_rc, stdout=self.probe_json or "", stderr=self.probe_err ) return subprocess.CompletedProcess(cmd, 0, stdout=self.probe_json or "", stderr="") class FakeStream: """Popen stdout/stderr 용 — 즉시 EOF.""" def read(self, n: int = -1) -> bytes: return b"" def test_probe_success(settings: Settings, tmp_path, monkeypatch): fake = FakeSubprocess(probe_json=_fake_probe_json(duration="120.0")) monkeypatch.setattr("luke_scribe.audio.ingest.subprocess", fake) src = tmp_path / "a.mp3" src.write_bytes(b"x" * 100) ing = AudioIngestor(settings) probe = ing.probe(src) assert probe.duration_sec == 120.0 assert probe.codec == "mp3" def test_probe_missing_file(settings: Settings, tmp_path): ing = AudioIngestor(settings) with pytest.raises(InvalidInput): ing.probe(tmp_path / "nope.mp3") def test_probe_size_limit(settings: Settings, tmp_path, monkeypatch): settings.max_upload_bytes = 100 src = tmp_path / "big.mp3" src.write_bytes(b"x" * 200) ing = AudioIngestor(settings) with pytest.raises(UnsupportedInputEnvelope): ing.probe(src) def test_probe_duration_limit(settings: Settings, tmp_path, monkeypatch): fake = FakeSubprocess(probe_json=_fake_probe_json(duration="999999")) monkeypatch.setattr("luke_scribe.audio.ingest.subprocess", fake) src = tmp_path / "long.mp3" src.write_bytes(b"x" * 10) ing = AudioIngestor(settings) with pytest.raises(UnsupportedInputEnvelope): ing.probe(src) def test_probe_no_audio_stream(settings: Settings, tmp_path, monkeypatch): fake = FakeSubprocess(probe_json=_fake_probe_json(has_audio=False)) monkeypatch.setattr("luke_scribe.audio.ingest.subprocess", fake) src = tmp_path / "video.mp4" src.write_bytes(b"x" * 10) ing = AudioIngestor(settings) with pytest.raises(AudioProbeFailed): ing.probe(src) def test_probe_ffprobe_missing(settings: Settings, tmp_path, monkeypatch): class NoFfprobe: def run(self, cmd, **kwargs): raise FileNotFoundError("ffprobe") monkeypatch.setattr("luke_scribe.audio.ingest.subprocess", NoFfprobe()) src = tmp_path / "a.mp3" src.write_bytes(b"x") ing = AudioIngestor(settings) with pytest.raises(AudioProbeFailed): ing.probe(src) def test_probe_bad_output(settings: Settings, tmp_path, monkeypatch): fake = FakeSubprocess(probe_json="not-json{") monkeypatch.setattr("luke_scribe.audio.ingest.subprocess", fake) src = tmp_path / "a.mp3" src.write_bytes(b"x") ing = AudioIngestor(settings) with pytest.raises(AudioProbeFailed): ing.probe(src) class TestIngestCancellation: def test_cancel_before_ingest(self, settings: Settings, tmp_path): src = tmp_path / "a.mp3" src.write_bytes(b"x") ing = AudioIngestor(settings) with pytest.raises(CancelledError): ing.ingest(src, should_cancel=lambda: True) def test_temp_dir_cleaned_on_cancel(self, settings: Settings, tmp_path, monkeypatch): """취소/실패 시 임시 디렉터리가 반드시 정리된다 (Eng P1).""" src = tmp_path / "a.mp3" src.write_bytes(b"x") class CancellingPopen: """무한 인코딩 중인 ffmpeg — wait 폴링에서 계속 TimeoutExpired.""" def __init__(self, cmd, **kwargs): self.stdout = FakeStream() self.stderr = FakeStream() self.pid = 99999999 # 존재하지 않는 pid → _kill_group의 ProcessLookupError 경로 def wait(self, timeout=None): raise subprocess.TimeoutExpired(cmd=[], timeout=0.5) def kill(self): pass # ffprobe는 성공, ffmpeg Popen은 취소 폴링 루프에 진입 fake = FakeSubprocess(probe_json=_fake_probe_json(duration="60.0")) monkeypatch.setattr("luke_scribe.audio.ingest.subprocess.run", fake.run) monkeypatch.setattr("luke_scribe.audio.ingest.subprocess.Popen", CancellingPopen) monkeypatch.setattr( "luke_scribe.audio.ingest.tempfile.mkdtemp", lambda prefix="": str(tmp_path / "ingest-tmp"), ) ing = AudioIngestor(settings) # 초기 검사 2회(프로브 전/후)는 통과, ffmpeg 폴링 루프에서 취소 calls = {"n": 0} def should_cancel(): calls["n"] += 1 return calls["n"] > 2 with pytest.raises(CancelledError): ing.ingest(src, should_cancel=should_cancel) assert not Path(tmp_path / "ingest-tmp").exists() def test_temp_dir_cleaned_on_ffmpeg_failure(self, settings: Settings, tmp_path, monkeypatch): src = tmp_path / "a.mp3" src.write_bytes(b"x") class FailingFfmpeg: def __init__(self, cmd, **kwargs): self.stdout = FakeStream() self.stderr = FakeStream() self.returncode = 1 def wait(self, timeout=None): return 1 fake = FakeSubprocess(probe_json=_fake_probe_json(duration="60.0")) monkeypatch.setattr("luke_scribe.audio.ingest.subprocess.Popen", FailingFfmpeg) monkeypatch.setattr("luke_scribe.audio.ingest.subprocess.run", fake.run) monkeypatch.setattr( "luke_scribe.audio.ingest.tempfile.mkdtemp", lambda prefix="": str(tmp_path / "ingest-tmp2"), ) ing = AudioIngestor(settings) with pytest.raises(AudioProbeFailed): ing.ingest(src) assert not Path(tmp_path / "ingest-tmp2").exists()