bambuddy/backend/tests/unit/services/test_timelapse_jobs.py

"""Timelapse editor re-encodes run as background jobs that report progress and can be cancelled."""

import asyncio
import shutil
import stat
import sys
from pathlib import Path

import pytest

from backend.app.services import timelapse_jobs
from backend.app.services.timelapse_processor import TimelapseProcessor, _communicate


@pytest.fixture(autouse=True)
def _no_jobs():
    timelapse_jobs._jobs.clear()
    yield
    timelapse_jobs._jobs.clear()


async def _finished(job: timelapse_jobs.TimelapseJob) -> None:
    assert job.task is not None
    await asyncio.wait_for(asyncio.shield(job.task), 5)


class TestJobs:
    @pytest.mark.asyncio
    async def test_a_finished_job_keeps_its_result(self):
        async def run(job):
            job.set_progress(0.5)
            return "archive/1/out.mp4", "Saved as out.mp4"

        job = timelapse_jobs.start_job(1, run)
        assert job.status == timelapse_jobs.RUNNING
        await _finished(job)

        assert timelapse_jobs.get_job(1) is job
        assert (job.status, job.progress, job.output_path, job.message) == (
            timelapse_jobs.COMPLETED,
            1.0,
            "archive/1/out.mp4",
            "Saved as out.mp4",
        )

    @pytest.mark.asyncio
    async def test_a_failed_job_says_why(self):
        async def run(job):
            raise RuntimeError("Video processing failed")

        job = timelapse_jobs.start_job(1, run)
        await _finished(job)

        assert (job.status, job.error) == (timelapse_jobs.FAILED, "Video processing failed")

    @pytest.mark.asyncio
    async def test_one_job_per_archive_at_a_time(self):
        release = asyncio.Event()

        async def run(job):
            await release.wait()
            return "a", "b"

        job = timelapse_jobs.start_job(1, run)
        with pytest.raises(timelapse_jobs.JobAlreadyRunning):
            timelapse_jobs.start_job(1, run)
        other = timelapse_jobs.start_job(2, run)

        release.set()
        await _finished(job)
        await _finished(other)
        again = timelapse_jobs.start_job(1, run)
        await _finished(again)
        assert again.status == timelapse_jobs.COMPLETED

    @pytest.mark.asyncio
    async def test_cancelling_stops_the_job(self):
        started = asyncio.Event()
        stopped = []

        async def run(job):
            started.set()
            try:
                await asyncio.sleep(30)
            except asyncio.CancelledError:
                stopped.append(True)
                raise
            return "a", "b"

        job = timelapse_jobs.start_job(1, run)
        await started.wait()

        assert await timelapse_jobs.cancel_job(1) is True
        assert job.status == timelapse_jobs.CANCELLED
        assert stopped == [True]
        assert await timelapse_jobs.cancel_job(1) is False

    @pytest.mark.asyncio
    async def test_shutdown_cancels_every_running_job(self):
        async def run(job):
            await asyncio.sleep(30)
            return "a", "b"

        jobs = [timelapse_jobs.start_job(i, run) for i in (1, 2)]
        await asyncio.sleep(0)

        await timelapse_jobs.cancel_all()

        assert [j.status for j in jobs] == [timelapse_jobs.CANCELLED] * 2

    @pytest.mark.asyncio
    async def test_no_job_to_cancel(self):
        assert await timelapse_jobs.cancel_job(7) is False


def _fake_ffmpeg(tmp_path: Path, body: str) -> str:
    """A stand-in ffmpeg: a Python script; the output path is its last argument."""
    script = tmp_path / "ffmpeg"
    script.write_text(f"#!{sys.executable}\nimport sys, time\nout = sys.argv[-1]\n{body}\n")
    script.chmod(script.stat().st_mode | stat.S_IEXEC)
    return str(script)


def _processor(tmp_path: Path, ffmpeg: str) -> TimelapseProcessor:
    processor = TimelapseProcessor.__new__(TimelapseProcessor)
    processor.input_path = tmp_path / "in.mp4"
    processor.ffmpeg = ffmpeg
    processor.ffprobe = ffmpeg
    return processor


class TestProcessor:
    @pytest.mark.asyncio
    async def test_progress_is_the_share_of_the_output_written(self, tmp_path):
        ffmpeg = _fake_ffmpeg(
            tmp_path,
            "open(out, 'wb').write(b'x')\n"
            "for us in ('N/A', '1000000', '2000000'):\n"
            "    print(f'out_time_us={us}', flush=True)\n"
            "    print('progress=continue', flush=True)",
        )
        seen: list[float] = []

        # 8 s of input at double speed: 4 s of output.
        ok = await _processor(tmp_path, ffmpeg).process(
            tmp_path / "out.mp4", trim_start=2, trim_end=10, speed=2.0, on_progress=seen.append
        )

        assert ok is True
        assert seen == [0.25, 0.5]

    @pytest.mark.asyncio
    async def test_cancelling_stops_ffmpeg_and_drops_the_partial_file(self, tmp_path):
        ffmpeg = _fake_ffmpeg(
            tmp_path,
            "open(out, 'wb').write(b'partial')\nprint('out_time_us=1000000', flush=True)\ntime.sleep(60)",
        )
        output = tmp_path / "out.mp4"
        reported = asyncio.Event()

        task = asyncio.create_task(
            _processor(tmp_path, ffmpeg).process(output, trim_end=10, on_progress=lambda _: reported.set())
        )
        await asyncio.wait_for(reported.wait(), 10)
        task.cancel()

        with pytest.raises(asyncio.CancelledError):
            await asyncio.wait_for(task, 5)
        assert not output.exists()

    @pytest.mark.asyncio
    async def test_a_probe_that_hangs_is_killed(self):
        sleep = shutil.which("sleep")
        if sleep is None:
            pytest.skip("no sleep binary")
        process = await asyncio.create_subprocess_exec(
            sleep, "30", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )

        # Killed, not waited out: well inside the 30 s the process would run.
        with pytest.raises(TimeoutError):
            await asyncio.wait_for(_communicate(process, 0.2), 5)

        assert process.returncode is not None


@pytest.mark.asyncio
async def test_without_a_known_length_it_encodes_without_progress(tmp_path):
    ffmpeg = _fake_ffmpeg(tmp_path, "open(out, 'wb').write(b'x')\nprint('out_time_us=1000000', flush=True)")
    processor = _processor(tmp_path, ffmpeg)
    processor.ffprobe = "/nonexistent/ffprobe"
    seen: list[float] = []

    ok = await processor.process(tmp_path / "out.mp4", on_progress=seen.append)

    assert ok is True
    assert seen == []