siftlode/backend/tests/test_run_ffmpeg.py

"""run_ffmpeg's process plumbing — the part that can hang a worker thread forever.

No ffmpeg here: a fake Popen stands in, so the reader threads, the 1s poll loop, the stall watchdog
and the bounded exit wait are all exercised on any machine. These are regression tests for a real
deadlock: stderr was opened and never read, so a chatty ffmpeg filled the ~64KB pipe, blocked on its
next write and never exited, and `proc.wait()` never returned."""
import subprocess
import time

import pytest

from app.downloads import edit


class _FakeStream:
    """A pipe. `block=True` = the process is alive but silent (a wedged ffmpeg): iteration parks
    until the stream is closed, which is what terminate()/kill() does to a real pipe."""

    def __init__(self, lines=(), block=False):
        self._lines = list(lines)
        self._block = block
        self.closed = False
        self.consumed = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self._lines:
            self.consumed += 1
            return self._lines.pop(0)
        while self._block and not self.closed:
            time.sleep(0.02)
        raise StopIteration

    def close(self):
        self.closed = True


class _FakeProc:
    pid = 4242

    def __init__(self, stdout, stderr, returncode=0, wait_timeouts=0):
        self.stdout = stdout
        self.stderr = stderr
        self._rc = returncode
        # How many of the first `wait()` calls raise TimeoutExpired — a process ignoring SIGTERM,
        # or one that lingers after closing its pipes.
        self._wait_timeouts = wait_timeouts
        self.terminated = False
        self.killed = False
        self.waits = 0

    def poll(self):
        # Popen.poll(): None while the process is alive. Our fake stays alive until it is stopped —
        # `_stop` uses this to skip a process that has already gone.
        return self._rc if self.terminated else None

    def terminate(self):
        self.terminated = True
        # A terminated process's pipes hit EOF — that is how the reader threads finish.
        self.stdout.close()
        self.stderr.close()

    def kill(self):
        self.killed = True
        self.terminate()

    def wait(self, timeout=None):
        self.waits += 1
        if self._wait_timeouts > 0:
            self._wait_timeouts -= 1
            raise subprocess.TimeoutExpired("ffmpeg", timeout or 0)
        return self._rc


@pytest.fixture
def fake_popen(monkeypatch):
    """Install a fake Popen and hand the test the process it will produce."""

    def install(proc):
        monkeypatch.setattr(edit.subprocess, "Popen", lambda *a, **k: proc)
        return proc

    return install


def test_progress_is_reported_and_a_clean_exit_returns(fake_popen):
    proc = fake_popen(
        _FakeProc(_FakeStream(["out_time_us=5000000\n", "progress=continue\n"]), _FakeStream())
    )
    seen = []
    edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=seen.append, should_cancel=lambda: False)
    assert seen == [50.0]  # 5s of 10s
    assert not proc.terminated


def test_progress_is_capped_below_100(fake_popen):
    fake_popen(_FakeProc(_FakeStream(["out_time_us=99000000\n"]), _FakeStream()))
    seen = []
    edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=seen.append, should_cancel=lambda: False)
    assert seen == [99.0]  # never reports done before the process actually exits


def test_a_nonzero_exit_raises_with_the_stderr_tail(fake_popen):
    # The tail must come from the drained stderr — reading the pipe only AFTER the exit was the
    # original deadlock.
    proc = fake_popen(
        _FakeProc(_FakeStream([]), _FakeStream(["Invalid data found\n"]), returncode=2)
    )
    with pytest.raises(RuntimeError, match="Invalid data found"):
        edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: False)
    assert not proc.terminated  # it exited on its own; nothing to kill


def test_cancel_is_honoured_while_the_process_is_silent(fake_popen):
    # The pre-fix loop only checked cancel when a progress line arrived, so a quiet ffmpeg was
    # uncancellable. The 1s poll tick is what makes this terminate.
    proc = fake_popen(_FakeProc(_FakeStream(block=True), _FakeStream(block=True)))
    started = time.monotonic()
    with pytest.raises(edit.EditAborted):
        edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: True)
    assert proc.terminated
    assert time.monotonic() - started < 5  # not "whenever ffmpeg next speaks"


def test_a_wedged_process_is_killed_by_the_stall_watchdog(fake_popen):
    proc = fake_popen(_FakeProc(_FakeStream(block=True), _FakeStream(block=True)))
    with pytest.raises(RuntimeError, match="stalled"):
        edit.run_ffmpeg(
            ["ffmpeg"],
            total_s=10,
            on_progress=lambda _p: None,
            should_cancel=lambda: False,
            stall_timeout_s=0.5,
        )
    assert proc.terminated  # the worker thread is freed, not parked forever


def test_a_callback_exception_still_stops_the_process(fake_popen):
    # `on_progress` is a database write in production. A failure THERE used to unwind past a live
    # ffmpeg — orphaning it, still encoding, after a 10s join on its pipes.
    proc = fake_popen(_FakeProc(_FakeStream(["out_time_us=1000000\n"]), _FakeStream(block=True)))

    def boom(_pct):
        raise RuntimeError("database is down")

    with pytest.raises(RuntimeError, match="database is down"):
        edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=boom, should_cancel=lambda: False)
    assert proc.terminated


def test_a_process_that_ignores_sigterm_is_killed(fake_popen):
    # `_stop`'s escalation: terminate, wait 5s, then SIGKILL. Without it a cancel could park the
    # worker thread on a process that simply refuses to go.
    proc = fake_popen(
        _FakeProc(_FakeStream(block=True), _FakeStream(block=True), wait_timeouts=1)
    )
    with pytest.raises(edit.EditAborted):
        edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: True)
    assert proc.terminated and proc.killed


def test_a_process_lingering_after_eof_is_stopped_and_reported(fake_popen):
    # Pipes closed but the process still around: the bounded wait must give up rather than block
    # the worker thread on `proc.wait()` forever (the old code waited with no timeout at all).
    proc = fake_popen(_FakeProc(_FakeStream([]), _FakeStream(), wait_timeouts=1))
    with pytest.raises(RuntimeError, match="did not exit"):
        edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: False)
    assert proc.terminated
    # Two waits: the bounded grace wait that timed out, then _stop's own — never an unbounded one.
    assert proc.waits == 2


def test_stderr_is_drained_while_the_process_runs(fake_popen):
    # The deadlock case: a chatty ffmpeg that exits 0. EVERY stderr line must be consumed — with a
    # real pipe, an unread one fills at ~64KB and blocks ffmpeg's next write forever.
    err = _FakeStream([f"warning {i}\n" for i in range(500)])
    fake_popen(_FakeProc(_FakeStream(["out_time_us=1000000\n"]), err))
    edit.run_ffmpeg(["ffmpeg"], total_s=10, on_progress=lambda _p: None, should_cancel=lambda: False)
    assert err.consumed == 500
    assert err.closed