TL;DR
We'll give a video pipeline a real test suite that runs on every pull request, costs nothing, and stores no video files in git. Fixtures are generated with FFmpeg's
lavfisource, outputs are asserted withffprobeJSON snapshots (not byte comparisons), and webhook handlers are tested by replaying recorded payloads. pytest, FFmpeg 8.1 pinned in GitHub Actions, Python 3.12.
What we're building
A tests/ directory for a pipeline that looks like most of them: a probe() step, a build_ffmpeg_args() step that picks a ladder, a transcode() step that shells out, and a webhook handler that advances state when the encoder (local or managed) reports back.
The four tiers, in order of how often they run:
| Tier | Needs FFmpeg? | Needs network? | Runs on |
|---|---|---|---|
| 1. Pure logic (arg builders, ladder policy) | no | no | every commit |
| 2. FFmpeg in the loop on synthetic fixtures | yes | no | every commit |
| 3. Webhook replay from recorded payloads | no | no | every commit |
| 4. One real provider call | yes | yes | nightly, budgeted |
This post builds tiers 2 and 3 in full and shows the skeleton for 1 and 4.
1. Project layout
pipeline/ probe.py # ffprobe wrapper ladder.py # pure: probe result -> list of renditions transcode.py # shells out to ffmpeg webhooks.py # FastAPI handler tests/ conftest.py # fixture factory test_ladder.py # tier 1 test_transcode.py # tier 2 test_webhooks.py # tier 3 fixtures/webhooks/*.json goldens/*.json .github/workflows/ci.yml python -m venv .venv && source .venv/bin/activate pip install pytest pytest-snapshot fastapi httpx 2. The fixture factory (no binaries in git) 🎬
FFmpeg can synthesise a test pattern and a tone from nothing. A two-second 320x240 clip is a few hundred KB and takes well under a second to make.
# tests/conftest.py import subprocess from pathlib import Path import pytest def make_fixture( out: Path, seconds: float = 2, size: str = "320x240", fps: int = 24, audio: bool = True, vcodec: str = "libx264", ) -> Path: cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", f"testsrc2=duration={seconds}:size={size}:rate={fps},format=yuv420p"] if audio: cmd += ["-f", "lavfi", "-i", f"sine=frequency=440:duration={seconds}"] cmd += ["-c:v", vcodec, "-preset", "ultrafast", "-crf", "28"] if audio: cmd += ["-c:a", "aac", "-shortest"] cmd += [str(out)] subprocess.run(cmd, check=True) return out @pytest.fixture def fixture_factory(tmp_path): def _make(name="in.mp4", **kw): return make_fixture(tmp_path / name, **kw) return _make Now every test can ask for exactly the input it needs:
fixture_factory() # 2s, 320x240, 24fps, with audio fixture_factory("vertical.mp4", size="240x426") # portrait source fixture_factory("silent.mp4", audio=False) # no audio stream fixture_factory("hfr.mp4", fps=60) # high frame rate 💡 Tip:
testsrc2produces a pattern with motion and text, which is closer to real content than plain colour bars and makes encoder behaviour (keyframes, bitrate) more realistic.
3. Tier 1: pure logic, no FFmpeg
The refactor that makes everything else possible: keep "decide what to run" separate from "run it".
# pipeline/ladder.py from dataclasses import dataclass @dataclass(frozen=True) class Rendition: height: int video_kbps: int LADDER = [Rendition(1080, 5000), Rendition(720, 2800), Rendition(480, 1400), Rendition(360, 800)] def pick_ladder(width: int, height: int) -> list[Rendition]: """Never upscale. Portrait sources are judged by their short edge.""" short_edge = min(width, height) return [r for r in LADDER if r.height <= short_edge] # tests/test_ladder.py from pipeline.ladder import pick_ladder def test_720p_source_gets_three_rungs(): assert [r.height for r in pick_ladder(1280, 720)] == [720, 480, 360] def test_portrait_source_uses_short_edge(): assert [r.height for r in pick_ladder(1080, 1920)] == [1080, 720, 480, 360] def test_tiny_source_gets_nothing_above_itself(): assert pick_ladder(320, 240) == [] These run in milliseconds and catch a surprising share of real bugs, because ladder logic is where the edge cases live.
4. Tier 2: FFmpeg in the loop, asserted with ffprobe
Encoders are not bit-exact across versions, so never compare output bytes. Probe the output and assert on the properties your pipeline is responsible for.
# pipeline/probe.py import json, subprocess from pathlib import Path def probe(path: Path) -> dict: out = subprocess.run( ["ffprobe", "-v", "error", "-print_format", "json", "-show_streams", "-show_format", str(path)], capture_output=True, text=True, check=True, ).stdout return json.loads(out) STABLE_STREAM_KEYS = ("codec_type", "codec_name", "width", "height", "pix_fmt", "r_frame_rate", "channels", "sample_rate") def stable_view(p: dict) -> dict: """Keep only fields that survive FFmpeg upgrades. Drop bit_rate, size, tags.""" return { "streams": [{k: s.get(k) for k in STABLE_STREAM_KEYS if k in s} for s in p["streams"]], "duration": round(float(p["format"]["duration"]), 1), "nb_streams": p["format"]["nb_streams"], } # pipeline/transcode.py import subprocess from pathlib import Path from .ladder import Rendition def transcode(src: Path, dst: Path, r: Rendition) -> Path: cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", str(src), "-vf", f"scale=-2:{r.height}", "-c:v", "libx264", "-preset", "veryfast", "-b:v", f"{r.video_kbps}k", "-g", "48", "-keyint_min", "48", "-sc_threshold", "0", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", str(dst)] subprocess.run(cmd, check=True) return dst Now the tests. One direct assertion, one golden snapshot.
# tests/test_transcode.py import json from pipeline.ladder import Rendition from pipeline.probe import probe, stable_view from pipeline.transcode import transcode def test_360p_rendition_has_expected_shape(fixture_factory, tmp_path): src = fixture_factory(size="640x360", seconds=2) out = transcode(src, tmp_path / "out.mp4", Rendition(360, 800)) view = stable_view(probe(out)) video = next(s for s in view["streams"] if s["codec_type"] == "video") audio = next(s for s in view["streams"] if s["codec_type"] == "audio") assert video["codec_name"] == "h264" assert (video["width"], video["height"]) == (640, 360) assert video["pix_fmt"] == "yuv420p" assert audio["codec_name"] == "aac" assert view["duration"] == 2.0 def test_360p_rendition_matches_golden(fixture_factory, tmp_path, snapshot): src = fixture_factory(size="640x360", seconds=2) out = transcode(src, tmp_path / "out.mp4", Rendition(360, 800)) snapshot.snapshot_dir = "tests/goldens" snapshot.assert_match(json.dumps(stable_view(probe(out)), indent=2, sort_keys=True), "360p.json") First run creates the golden; later runs diff against it.
pytest tests/test_transcode.py --snapshot-update # once, to create tests/goldens/360p.json pytest tests/test_transcode.py # tests/test_transcode.py .. [100%] # 2 passed in 1.84s ⚠️ Note: if a golden ever includes
bit_rateorsize, it will break on the next FFmpeg upgrade and someone will delete the test.stable_view()exists to stop that.
The keyframe cadence you set with -g 48 is stable across builds and worth asserting too:
def keyframe_indices(path): out = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "frame=key_frame", "-of", "csv=p=0", str(path)], capture_output=True, text=True, check=True).stdout.split() return [i for i, k in enumerate(out) if k == "1"] def test_keyframes_every_48_frames(fixture_factory, tmp_path): src = fixture_factory(size="640x360", seconds=4, fps=24) out = transcode(src, tmp_path / "out.mp4", Rendition(360, 800)) assert keyframe_indices(out) == [0, 48] 5. Tier 3: webhook replay from recorded payloads
The provider is not part of your test suite. Its payloads are. Capture one real payload per event type once (from a dev environment, with a request logger), strip anything sensitive, and commit the JSON.
// tests/fixtures/webhooks/media_ready.json { "type": "media.ready", "id": "evt_01", "data": { "id": "media_abc123", "status": "ready", "playback_url": "https://cdn.example.com/media_abc123/index.m3u8" } } // tests/fixtures/webhooks/media_failed.json { "type": "media.failed", "id": "evt_02", "data": { "id": "media_abc123", "status": "failed", "error": { "message": "unsupported codec" } } } Field names will differ per provider; the shape of the tests does not.
# pipeline/webhooks.py from fastapi import FastAPI, Request app = FastAPI() STATE: dict[str, str] = {} # media_id -> status; swap for your DB SEEN: set[str] = set() # event ids, for idempotency @app.post("/webhooks/video") async def video_webhook(req: Request): evt = await req.json() if evt["id"] in SEEN: return {"ok": True, "duplicate": True} SEEN.add(evt["id"]) media_id = evt["data"]["id"] if evt["type"] == "media.ready": STATE[media_id] = "ready" elif evt["type"] == "media.failed": STATE[media_id] = "failed" return {"ok": True} # tests/test_webhooks.py import json from pathlib import Path import pytest from httpx import ASGITransport, AsyncClient from pipeline import webhooks FIX = Path("tests/fixtures/webhooks") load = lambda name: json.loads((FIX / f"{name}.json").read_text()) @pytest.fixture(autouse=True) def reset_state(): webhooks.STATE.clear(); webhooks.SEEN.clear() @pytest.fixture async def client(): async with AsyncClient(transport=ASGITransport(app=webhooks.app), base_url="http://t") as c: yield c async def test_ready_marks_media_ready(client): r = await client.post("/webhooks/video", json=load("media_ready")) assert r.status_code == 200 assert webhooks.STATE["media_abc123"] == "ready" async def test_duplicate_ready_is_idempotent(client): await client.post("/webhooks/video", json=load("media_ready")) r = await client.post("/webhooks/video", json=load("media_ready")) assert r.json()["duplicate"] is Source：https://dev.to/masonwritescode/test-your-video-pipeline-in-ci-for-free-synthetic-fixtures-ffprobe-goldens-and-webhook-replay-3h9dRelated posts
DeepSeek's Open-Source "Super-Agent" Codes Without Being Asked
All-in-one platform or a stack you assemble? Where do coaches and educators land?
I Wanted to Live on Coruscant. PAT Scared Me. Now My Alexa Lives on Mute.
Detect What You Can Name: PaliGemma 2 Object Detection with Transformers
When Should an LLM Router Stop Being Sticky? llm-d 1.0 Cache-Affinity Routing, Break-Even Math, and a Simulation
