Testing Data Pipelines and CI/CD for Data

Pipeline code fails in two ways. The code can be wrong: a join that duplicates rows, a filter that drops a region, a time zone bug. Or the data can be wrong: a source that changed format, a partner file with half the usual rows. Tests catch the first kind before deployment; data checks catch the second kind at run time. You need both, and CI/CD is what makes sure the first kind always runs.

The layers

Layer	What it tests	Runs	Typical tools
Unit tests	One transformation function or model on small, hand-written inputs	Every commit, in seconds	pytest, dbt unit tests
Integration tests	Real SQL on a real engine, reading and writing tables, connectors, end-to-end on a tiny dataset	Every pull request	pytest with DuckDB, PostgreSQL in a container, a CI schema in the warehouse
Data tests	Properties of the actual data: unique keys, not null, relationships, freshness	Every pipeline run, and in CI on the built tables	dbt tests, Great Expectations, Soda
Contract tests	A producer’s output still matches the agreed schema	Producer’s CI	Schema registry compatibility checks, contract validators
Regression diffs	The change alters only what it was meant to alter	Pull requests that change logic	EXCEPT queries, diff tools, dbt audit packages
Orchestration tests	DAGs import, have no cycles, have the expected settings	Every commit	Airflow DagBag tests (Airflow testing)

Unit tests are cheap and precise; integration tests catch the SQL and type problems unit tests cannot see; data tests protect production from inputs nobody anticipated. A pipeline with only data tests finds its own bugs in production; a pipeline with only unit tests is surprised by every upstream change.

Unit tests for transformations

Structure code so the transformation is a function of plain inputs (see reusable pipeline code). Then test the behaviour that breaks pipelines, not just the happy path:

from datetime import datetime, timezone

def clean_orders(rows: list[dict]) -> list[dict]:
    """Normalise raw orders: UTC timestamps, amounts in cents, latest version per order_id."""
    latest: dict[int, dict] = {}
    for r in rows:
        ts = datetime.fromisoformat(r["updated_at"]).astimezone(timezone.utc)
        order = {"order_id": int(r["order_id"]),
                 "amount_cents": round(float(r["amount"]) * 100),
                 "updated_at": ts}
        current = latest.get(order["order_id"])
        if current is None or order["updated_at"] > current["updated_at"]:
            latest[order["order_id"]] = order
    return sorted(latest.values(), key=lambda o: o["order_id"])

def test_empty_input_gives_empty_output():
    assert clean_orders([]) == []

def test_keeps_latest_version_per_key():
    rows = [{"order_id": "1", "amount": "10", "updated_at": "2026-10-05T10:00:00+00:00"},
            {"order_id": "1", "amount": "12", "updated_at": "2026-10-05T11:00:00+00:00"},
            {"order_id": "1", "amount": "11", "updated_at": "2026-10-05T10:30:00+00:00"}]
    assert [o["amount_cents"] for o in clean_orders(rows)] == [1200]

def test_converts_offsets_to_utc():
    rows = [{"order_id": "2", "amount": "5", "updated_at": "2026-10-05T01:30:00+02:00"}]
    assert clean_orders(rows)[0]["updated_at"] == datetime(2026, 10, 4, 23, 30, tzinfo=timezone.utc)

def test_rounds_float_amounts_to_cents():
    rows = [{"order_id": "3", "amount": "19.99", "updated_at": "2026-10-05T00:00:00+00:00"}]
    assert clean_orders(rows)[0]["amount_cents"] == 1999   # int(19.99 * 100) would give 1998

for test in (test_empty_input_gives_empty_output, test_keeps_latest_version_per_key,
             test_converts_offsets_to_utc, test_rounds_float_amounts_to_cents):
    test()
print("4 passed")

4 passed

With pytest installed, save the tests in tests/test_clean_orders.py and run pytest -q; pytest discovers functions named test_* and reports each one. The edge cases above are the ones that most often reach production: empty inputs, duplicate and out-of-order versions, time zones, and floating-point amounts.

Edge cases worth a test in almost every pipeline: an empty input, a duplicate key, a null in a required field, an out-of-order update, a timestamp exactly on the interval boundary, a daylight-saving change, a very large value, and non-ASCII text.

Integration tests on a real engine

SQL is code too, and much of it lives in strings that unit tests never execute. Run it on a real engine with a tiny dataset. DuckDB runs in-process with no server, which makes it a convenient engine for fast tests of portable SQL; for engine-specific SQL, run PostgreSQL in a container or use a dedicated CI schema in your warehouse.

import duckdb

DAILY_REVENUE_SQL = """
    CREATE OR REPLACE TABLE daily_revenue AS
    SELECT CAST(o.ordered_at AS DATE) AS order_date,
           c.country,
           SUM(o.amount) AS revenue,
           COUNT(*)      AS orders
    FROM orders AS o
    JOIN customers AS c USING (customer_id)
    GROUP BY ALL
"""

def build(con):
    con.execute(DAILY_REVENUE_SQL)

def make_fixture():
    con = duckdb.connect()
    con.execute("CREATE TABLE customers (customer_id INT, country TEXT)")
    con.execute("INSERT INTO customers VALUES (1, 'DE'), (2, 'FR')")
    con.execute("CREATE TABLE orders (order_id INT, customer_id INT, amount DECIMAL(10,2), ordered_at TIMESTAMP)")
    con.execute("""INSERT INTO orders VALUES
        (10, 1, 50.00, '2026-10-05 09:00'), (11, 1, 20.00, '2026-10-05 18:00'),
        (12, 2, 30.00, '2026-10-05 12:00')""")
    return con

def test_revenue_per_country_and_day():
    con = make_fixture()
    build(con)
    rows = con.execute("SELECT * FROM daily_revenue ORDER BY country").fetchall()
    assert [(str(d), c, float(r), n) for d, c, r, n in rows] == [
        ("2026-10-05", "DE", 70.0, 2), ("2026-10-05", "FR", 30.0, 1)]

def test_duplicate_dimension_row_would_fan_out():
    con = make_fixture()
    con.execute("INSERT INTO customers VALUES (1, 'DE')")          # a duplicate in the dimension
    build(con)
    total = con.execute("SELECT SUM(revenue) FROM daily_revenue").fetchone()[0]
    assert float(total) == 170.0                                   # 70 counted twice: the join fanned out

def test_build_is_idempotent():
    con = make_fixture()
    build(con)
    first = con.execute("SELECT * FROM daily_revenue ORDER BY ALL").fetchall()
    build(con)
    assert con.execute("SELECT * FROM daily_revenue ORDER BY ALL").fetchall() == first

for test in (test_revenue_per_country_and_day, test_duplicate_dimension_row_would_fan_out, test_build_is_idempotent):
    test()
print("3 passed")

3 passed

The second test documents a real risk rather than hiding it: a duplicate customer silently doubles revenue. The fix is not in this model but in a uniqueness data test on customers that blocks the run, and the test makes that dependency visible to the next person who edits the query. The third test is the run-twice idempotency test from the idempotency lesson, now part of the suite.

Keep fixtures small and readable: a handful of rows per case, each one there for a reason. Large copies of production data make tests slow, hard to understand and a privacy risk.

Data tests

Data tests run against the real tables, on every pipeline run and in CI against the tables built from the pull request’s code: unique and not-null keys, accepted values, relationships, freshness and volume. They are covered in data quality checks and contracts. The important distinction for CI: a data test failing in CI on fixture data points to a code problem; a data test failing in production on real data points to a data problem.

dbt also has unit tests (since dbt 1.8): you give a model mock input rows and expected output rows in YAML, and dbt runs the model’s SQL against them without touching real data. They fill the same role as the DuckDB tests above for dbt models.

Regression diffs

Before merging a logic change, compare the table built by the new code with the current production table. A symmetric difference shows every row that would change:

CREATE TABLE prod_daily_revenue (order_date date, country text, revenue numeric(12,2));
CREATE TABLE ci_daily_revenue   (order_date date, country text, revenue numeric(12,2));
INSERT INTO prod_daily_revenue VALUES ('2026-10-04', 'DE', 100.00), ('2026-10-04', 'FR', 40.00), ('2026-10-05', 'DE', 70.00);
INSERT INTO ci_daily_revenue   VALUES ('2026-10-04', 'DE', 100.00), ('2026-10-04', 'FR', 42.50), ('2026-10-05', 'DE', 70.00),
                                      ('2026-10-05', 'FR', 30.00);

(SELECT 'only in prod' AS side, * FROM prod_daily_revenue
 EXCEPT
 SELECT 'only in prod', * FROM ci_daily_revenue)
UNION ALL
(SELECT 'only in ci', * FROM ci_daily_revenue
 EXCEPT
 SELECT 'only in ci', * FROM prod_daily_revenue)
ORDER BY order_date, country, side;

     side     | order_date | country | revenue
--------------+------------+---------+---------
 only in ci   | 2026-10-04 | FR      |   42.50
 only in prod | 2026-10-04 | FR      |   40.00
 only in ci   | 2026-10-05 | FR      |   30.00

The reviewer now asks the right questions: is the FR change for 4 October the intended fix? Why did a new FR row appear for 5 October? For large tables, compare aggregates per partition first (row counts and sums per day) and drill into the partitions that differ. EXCEPT treats rows as sets, so also compare count(*) to catch changes in the number of duplicate rows.

CI/CD for data code

A typical pipeline for a repository holding Python jobs, SQL or dbt models and DAGs:

1. Lint and format: ruff for Python, sqlfluff for SQL, YAML validation. Fast and catches typos before anything runs.
2. Unit tests: pytest and dbt unit tests. Seconds to a few minutes.
3. Build in an isolated environment: create the changed models in a schema named after the pull request (for example ci_pr_482), never in production. With dbt, “slim CI” builds only modified models and their children with --select state:modified+, comparing against the production manifest, and --defer reads unchanged parents from production instead of rebuilding them.
4. Data tests and regression diffs on the CI build.
5. Review: a human reads the code and the diff summary.
6. Deploy on merge: publish DAGs and code to production (a versioned artifact or a bundle), run schema migrations, then let the scheduler pick up the change. Drop the CI schema.

name: data-ci
on: [pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r requirements.txt
      - run: ruff check . && sqlfluff lint models/
      - run: pytest -q tests/
      - name: Build and test changed models in a PR schema
        env:
          DBT_SCHEMA: ci_pr_${{ github.event.number }}
          WAREHOUSE_PASSWORD: ${{ secrets.CI_WAREHOUSE_PASSWORD }}
        run: |
          dbt deps
          dbt build --target ci --select state:modified+ --defer --state prod-artifacts/

Practices that matter more than the tool:

• Separate environments and credentials. CI uses a role that can create and drop CI schemas but cannot write production tables. Secrets come from the CI secret store, never from the repository.
• Make deploys reversible. Version the code artifact so a bad deploy can be rolled back; for table changes, prefer additive migrations (add a column, backfill, switch readers, then drop the old one) over in-place destructive changes.
• Plan backfills as part of the change. A logic change to a historical model usually needs a backfill; say so in the pull request and run it through the normal job (see backfills).
• Keep CI fast. Run the full suite on merge to main or nightly if it is slow, but keep the pull-request checks to a few minutes, or people stop waiting for them.

Common mistakes

1. Testing only the happy path, so duplicates, nulls and time zones are discovered in production.
2. SQL that no test ever executes, because it lives in strings inside Python.
3. Fixtures copied from production: slow, opaque and a privacy risk.
4. CI that builds into the production schema, or shares credentials with production.
5. Relying on data tests alone and finding your own logic bugs after publishing.
6. Merging logic changes without a diff against production, then explaining unexpected numbers to finance.

Interview angle

“How do you test a data pipeline?” separates candidates quickly. Name the layers (unit tests for transforms, integration tests on a real engine, data tests on real data, regression diffs for logic changes), give concrete edge cases (duplicates, nulls, late and out-of-order data, time zones, empty inputs), mention the run-twice idempotency test, and describe a CI pipeline with an isolated schema and a diff step. Practise with how do you test a data pipeline? and CI/CD for data pipelines.

Key takeaway

Unit-test transformations with the edge cases that break pipelines, run SQL against a real engine on tiny fixtures, keep data tests on every run, diff logic changes against production before merging, and ship through CI/CD with isolated environments, least-privilege credentials and reversible deploys.