siftlode/backend/tests/test_ffmpeg_callbacks.py

"""The rate limit in front of run_ffmpeg's callbacks.

run_ffmpeg calls `on_progress`/`should_cancel` once per ffmpeg output LINE (`-progress pipe:1`
emits a block of ~12 every half second), and both callbacks are database round-trips: `_set_job`
opens a session and UPDATE+COMMITs, `_job_status` opens one and SELECTs. Unthrottled, a long
re-encode is ~10^5 of each."""

import app.downloads.pipeline as pipeline


class TestProgressWriter:
    def test_writes_the_first_value_immediately(self, monkeypatch):
        writes = []
        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
        pipeline._ffmpeg_progress_writer(7, "processing")(12.4)
        assert writes == [{"progress": 12, "phase": "processing"}]

    def test_a_burst_of_lines_costs_one_write(self, monkeypatch):
        writes = []
        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
        write = pipeline._ffmpeg_progress_writer(7, "processing")
        for pct in range(0, 40):  # what one second of ffmpeg output looks like
            write(pct)
        assert len(writes) == 1

    def test_the_next_second_writes_again(self, monkeypatch):
        writes = []
        clock = {"t": 1000.0}
        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
        write = pipeline._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
        write(10)
        write(11)  # same second → dropped
        clock["t"] += 1.5
        write(12)
        assert [w["progress"] for w in writes] == [10, 12]

    def test_an_unchanged_whole_percent_is_never_rewritten(self, monkeypatch):
        writes = []
        clock = {"t": 1000.0}
        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
        monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
        write = pipeline._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
        write(10.1)
        clock["t"] += 5
        write(10.9)  # still 10% — the bar shows integers
        assert [w["progress"] for w in writes] == [10]

    def test_a_slow_same_percent_stretch_heartbeats_the_watchdog(self, monkeypatch):
        """B-sub-4: a re-encode that sits on one integer percent for minutes must NOT be killed as
        stalled — the writer touches the row (heartbeat) each interval even though it writes no new
        progress, so the stall watchdog keeps seeing a fresh updated_at."""
        writes, touches = [], []
        clock = {"t": 1000.0}
        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
        monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: touches.append(job_id))
        write = pipeline._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
        write(10.1)  # first real write → progress 10
        for _ in range(3):  # three more intervals stuck on 10%
            clock["t"] += 1.5
            write(10.4)
        assert [w["progress"] for w in writes] == [10]  # progress never rewritten
        assert touches == [7, 7, 7]  # …but each interval heartbeats the watchdog

    def test_no_redundant_heartbeat_when_a_progress_write_just_happened(self, monkeypatch):
        """A progress write already refreshes updated_at, so an unchanged-percent call in the SAME
        interval must NOT also heartbeat — that would be a redundant DB write (~2x during an active
        download). The heartbeat only fires once no write/beat touched the row for an interval."""
        writes, touches = [], []
        clock = {"t": 1000.0}
        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
        monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: touches.append(job_id))
        write = pipeline._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
        write(6.0)  # progress write at t=1000 → refreshes updated_at
        clock["t"] += 0.3
        write(
            6.4
        )  # same percent, SAME interval → the write already touched the row, so no heartbeat
        assert [w["progress"] for w in writes] == [6]
        assert touches == []


class TestCancelPoll:
    def test_asks_once_then_reuses_the_answer(self, monkeypatch):
        calls = []
        monkeypatch.setattr(
            pipeline, "_job_status", lambda job_id: calls.append(job_id) or "running"
        )
        cancelled = pipeline._cancel_poll(7)
        for _ in range(40):
            assert cancelled() is False
        assert len(calls) == 1

    def test_a_cancelled_job_is_reported_and_never_queried_again(self, monkeypatch):
        calls = []
        clock = {"t": 1000.0}

        def status(job_id):
            calls.append(job_id)
            return "cancelled"

        monkeypatch.setattr(pipeline, "_job_status", status)
        cancelled = pipeline._cancel_poll(7, now=lambda: clock["t"])
        assert cancelled() is True
        clock["t"] += 60  # cancellation is sticky: no re-query, and no un-cancelling
        assert cancelled() is True
        assert len(calls) == 1

    def test_a_cancel_arriving_later_is_seen_on_the_next_tick(self, monkeypatch):
        clock = {"t": 1000.0}
        state = {"status": "running"}
        monkeypatch.setattr(pipeline, "_job_status", lambda job_id: state["status"])
        cancelled = pipeline._cancel_poll(7, now=lambda: clock["t"])
        assert cancelled() is False
        state["status"] = "paused"
        clock["t"] += 1.0  # the throttle must not delay a cancel by more than its interval
        assert cancelled() is True