luke_scribe/tests/unit/test_ingest.py

"""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()