Validating and testing data pipelines

Fixture files, round-trip and property tests, schema drift detection, a malformed-input suite, and the metrics that catch bad data before a customer does.

Fixtures and round trips

A pipeline bug is usually a parsing bug. Test the boundary — the code that turns bytes into records — with files rather than with in-memory objects, because only files carry the encoding and quoting problems.

tests/fixtures/
  users.valid.json
  users.missing-required.json
  users.nan.json
  legacy.bom.csv
  legacy.crlf.csv
  legacy.tab-delimited.csv
  legacy.single-column.csv
  events.partial-line.ndjson

import json, pathlib, pytest
from mypipe import parse_users

@pytest.mark.parametrize("path", pathlib.Path("tests/fixtures").glob("users.*.json"))
def test_valid_fixtures_parse(path):
    data = json.loads(path.read_text(encoding="utf-8"))
    records = parse_users(data)
    assert all(r["id"] > 0 for r in records)

def test_round_trip_is_lossless(tmp_path):
    original = {"id": 1, "note": "line\nbreak, with comma", "amount": "19.99"}
    out = tmp_path / "out.json"
    out.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8")
    assert json.loads(out.read_text(encoding="utf-8")) == original

Property tests find the cases you would not write

from hypothesis import given, strategies as st

# business keys: letters, digits, spaces, quotes, newlines, commas
text = st.text(alphabet=st.characters(blacklist_categories=("Cs",)), max_size=200)

@given(st.lists(st.dictionaries(st.text(max_size=8), text, max_size=5), max_size=20))
def test_csv_round_trip(rows):
    encoded = to_csv(rows)
    assert list(from_csv(encoded)) == rows

@given(st.dictionaries(st.text(min_size=1, max_size=10), text))
def test_json_round_trip(obj):
    assert json.loads(json.dumps(obj)) == obj

Round-trip properties catch the quoting, escaping and Unicode bugs that hand-written examples miss, because the generator will produce a value containing a quote, a newline and an emoji at the same time.

Schema drift and reject metrics

Signal	What it catches	Where to alert
Schema hash changed	A producer added or renamed a field	Pipeline startup, as a warning
Required-field failure rate	Upstream broke a contract	Per batch; alert above a small threshold
New enum value seen	Producer shipped a state you do not handle	On first occurrence
Row count outside the expected band	Truncated or duplicated input	Per run versus a rolling baseline
Null rate jumped	Field silently dropped upstream	Per field, versus last week

def load(batch, stats):
    good, rejected = [], []
    for i, row in enumerate(batch):
        try:
            good.append(schema.validate(row))
        except ValidationError as e:
            rejected.append({"index": i, "error": str(e)[:200]})
    stats.incr("rows_in", len(batch))
    stats.incr("rows_rejected", len(rejected))
    if rejected:
        # keep the rejects; never drop them silently
        write_ndjson("rejects/dt=2026-09-18/part.ndjson", rejected)
    return good

💡

Dead-letter the rejects instead of dropping them. Quietly discarding rows that fail validation is how a pipeline reports success while losing a tenth of its data, and the loss is usually discovered by an analyst months later.

FAQ

How many fixtures are enough?

Cover the shapes, not the values: valid, missing required, wrong type, extreme number, empty, BOM and CRLF, embedded delimiter, embedded newline, and one truncated record. That set catches most real incidents.

Should validation failures stop the pipeline?

Fail fast on a schema violation you cannot route around, and quarantine individual bad records. A whole batch failing on one malformed row is as harmful as silently discarding it.

Schema definition and validation Line-delimited data: NDJSON and log formats

Last refreshed 2026-09-18.

Validating and testing data pipelines

Fixtures and round trips

Property tests find the cases you would not write

Schema drift and reject metrics

FAQ

Related