"""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.""" from app import worker class TestProgressWriter: def test_writes_the_first_value_immediately(self, monkeypatch): writes = [] monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f)) worker._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(worker, "_set_job", lambda job_id, **f: writes.append(f)) write = worker._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(worker, "_set_job", lambda job_id, **f: writes.append(f)) write = worker._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(worker, "_set_job", lambda job_id, **f: writes.append(f)) write = worker._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] class TestCancelPoll: def test_asks_once_then_reuses_the_answer(self, monkeypatch): calls = [] monkeypatch.setattr(worker, "_job_status", lambda job_id: calls.append(job_id) or "running") cancelled = worker._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(worker, "_job_status", status) cancelled = worker._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(worker, "_job_status", lambda job_id: state["status"]) cancelled = worker._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