Integration Testing Codec Contracts

This page sits beneath CI/CD and testing for media pipelines in the broader Pipeline Automation & Batch Processing layer, and it solves one specific, high-cost failure: a code change that still produces a playable file but quietly violates the delivery spec — 44.1 kHz where the contract demands 48 kHz, an mp4 with the wrong major brand, or audio that drifted 2 LUFS hot. Every Python unit test passes, the encode exits zero, and the regression only surfaces when a listener’s player chokes or a distributor rejects the feed.

Problem Framing

A codec contract is the set of properties every asset leaving a stage must have: exact audio codec and sample rate, container format and brand, channel layout, and an integrated loudness within a tight band. Nothing in ordinary CI enforces it. Consider a one-line change that “cleans up” a filtergraph by dropping an explicit aresample:

- af = "loudnorm=I=-16:TP=-1.5:LRA=11,aresample=48000,aformat=channel_layouts=stereo"
+ af = "loudnorm=I=-16:TP=-1.5:LRA=11,aformat=channel_layouts=stereo"

The encode still succeeds. The file still plays on the developer’s laptop. But the output now inherits the source’s 44.1 kHz sample rate, and the loudnorm-only chain measures slightly differently. A probe of the shipped asset reveals the break only after the fact:

$ ffprobe -v error -show_entries stream=codec_name,sample_rate,channels out.m4a
codec_name=aac
sample_rate=44100        # contract requires 48000
channels=2
$ ffmpeg -i out.m4a -af loudnorm=print_format=summary -f null -
Input Integrated:  -14.1 LUFS   # contract requires -16.0 ±0.5

By the time this reaches a distributor’s conformance check or a listener’s device, it has already been packaged and pushed. An integration test that runs the real transform on a fixture and asserts the output against the contract turns this into a red build on the pull request instead. Crucially, it validates the output, not the arguments — it does not care how the file was produced, only that what came out conforms.

Solution Architecture

The suite has four moving parts and one guarantee: it never trusts intent, only the probed bytes of the output file. First, the contract is a versioned schema, checked into the repo. Second, a pytest fixture runs the real, pinned FFmpeg transform on a small sample. Third, ffprobe emits the output’s stream and format metadata as JSON, validated against the schema. Fourth, a separate loudness assertion re-measures integrated loudness because that is a signal property ffprobe cannot report.

Because the loudness check re-runs FFmpeg’s own loudnorm in measure mode, it validates the perceptual result the way a distributor’s conformance tool will, closing the one gap ffprobe leaves open.

Implementation

1. Declare the contract as a versioned schema

Keep the contract as data, not scattered assertions, so a spec change is a reviewable diff and every test reads from one source of truth.

# python 3.10+   jsonschema==4.22.0
CODEC_CONTRACT = {
    "schema_version": 3,
    "audio": {
        "codec_name": "aac",
        "sample_rate": 48000,
        "channels": 2,
        "channel_layout": "stereo",
    },
    "container": {
        "format_name_contains": "mp4",     # ffprobe reports "mov,mp4,m4a,3gp,3g2,mj2"
        "major_brand": "M4A ",             # from format tags
    },
    "loudness": {
        "target_i": -16.0,
        "tolerance_lufs": 0.5,             # ±0.5 LUFS, matching delivery SLA
    },
}

The loudness target and tolerance deliberately mirror the delivery band enforced by audio codec normalization workflows, so this test fails on the same drift a downstream conformance check would reject.

2. Run the real transform and probe the output

The fixture runs the pinned FFmpeg — the exact production build — then probes the result. ffprobe -show_streams -show_format returns everything the contract needs except loudness.

# python 3.10+   pytest==8.2.2
import json, subprocess, pytest

pytestmark = pytest.mark.integration

def _run(cmd: list[str]) -> str:
    return subprocess.run(cmd, capture_output=True, text=True, check=True).stdout

@pytest.fixture
def transcoded(tmp_path):
    src = tmp_path / "src.wav"
    # synthesize a deterministic 1s 44.1kHz mono source to force real work
    _run(["ffmpeg", "-y", "-f", "lavfi",
          "-i", "sine=frequency=220:sample_rate=44100:duration=1",
          "-ac", "1", str(src)])
    dst = tmp_path / "out.m4a"
    # the pipeline transform under test (imported from the worker)
    from worker.transforms import normalize_and_encode
    normalize_and_encode(str(src), str(dst))
    return dst

def probe(path) -> dict:
    out = _run(["ffprobe", "-v", "error", "-show_streams",
                "-show_format", "-of", "json", str(path)])
    return json.loads(out)

Using a 44.1 kHz mono source guarantees the transform must actually resample and up-mix — a fixture that already matches the contract would let a broken resample chain pass. The ffprobe invocation follows the official ffprobe documentation.

3. Assert codec, container, and sample rate

Map the probe output onto the schema and assert each field explicitly, with messages that name the exact mismatch so a failure is self-explaining in the CI log.

def test_audio_stream_matches_contract(transcoded):
    meta = probe(transcoded)
    audio = next(s for s in meta["streams"] if s["codec_type"] == "audio")
    c = CODEC_CONTRACT["audio"]
    assert audio["codec_name"] == c["codec_name"], \
        f"codec {audio['codec_name']} != {c['codec_name']}"
    assert int(audio["sample_rate"]) == c["sample_rate"], \
        f"sample_rate {audio['sample_rate']} != {c['sample_rate']}"
    assert audio["channels"] == c["channels"]
    assert audio.get("channel_layout") == c["channel_layout"]

def test_container_matches_contract(transcoded):
    fmt = probe(transcoded)["format"]
    c = CODEC_CONTRACT["container"]
    assert c["format_name_contains"] in fmt["format_name"]
    assert fmt.get("tags", {}).get("major_brand", "").strip() == c["major_brand"].strip()

4. Assert integrated loudness within tolerance

ffprobe cannot report perceived loudness, so re-measure with a loudnorm analysis pass and parse its JSON. This is the assertion that catches a dropped or reordered loudness filter.

def measure_lufs(path) -> float:
    proc = subprocess.run(
        ["ffmpeg", "-hide_banner", "-i", str(path),
         "-af", "loudnorm=print_format=json", "-f", "null", "-"],
        capture_output=True, text=True,
    )
    blob = proc.stderr[proc.stderr.rindex("{"): proc.stderr.rindex("}") + 1]
    return float(json.loads(blob)["input_i"])

def test_loudness_within_tolerance(transcoded):
    c = CODEC_CONTRACT["loudness"]
    measured = measure_lufs(transcoded)
    assert abs(measured - c["target_i"]) <= c["tolerance_lufs"], \
        f"integrated {measured} LUFS outside {c['target_i']}±{c['tolerance_lufs']}"

The loudnorm filter reference in the FFmpeg filters documentation describes the input_i field these assertions read.

Verification

Run the contract suite against the pinned image and confirm it fails on a deliberately broken transform, which is the only proof the test has teeth.

# 1. Green on the correct pipeline
pytest tests/contract -q -m integration
# expected: 4 passed

# 2. Prove it fails when the contract is violated — force 44.1kHz output
FORCE_BROKEN_SR=1 pytest tests/contract/test_audio_stream_matches_contract.py -q
# expected: FAILED ... sample_rate 44100 != 48000

# 3. Sanity-check the probe directly on a produced artifact
ffprobe -v error -show_entries stream=codec_name,sample_rate,channels \
  -show_entries format=format_name:format_tags=major_brand -of json out.m4a

A healthy run is four green assertions on the correct pipeline and an immediate, well-labeled failure when a resample or loudness filter is removed. If step 2 passes, the test is inert — check that it reads real probe output and not cached or stubbed metadata.

Failure Modes & Edge Cases

For output regressions that are not about a fixed contract field but about drift from a known-good reference — a subtly different bitrate ladder or picture — pair this with golden-file testing for transcode output, which compares against a stored reference instead of a declared spec.

FAQ

Related

• Up to the parent guide: CI/CD and testing for media pipelines
• Sibling technique on reference drift: golden-file testing for transcode output
• The loudness target this contract enforces: audio codec normalization workflows
• Pipeline overview: Pipeline Automation & Batch Processing