Data Pipeline Testing and CI/CD

Data Engineering & Modern Data Platforms · Lesson 23 · Data Ingestion, ETL and Orchestration

Lesson Overview

The cents bug that forced the backfill in Advanced Airflow Operations was a one-line cast through float. It passed code review because the DAG still imported, the grid still showed green tasks, and nobody had a fixture where 49.90 had to become 4990. Production was the test. Finance was the assertion. The backfill was the rollback plan you use when the test suite does not exist.

This lesson builds the gate in front of tlp_orders_daily. Pure transform tests in pytest lock the rules from Data Cleansing and Transformation: cents, quarantine reasons, dedup winners, SCD change rows, and an idempotent merge against PostgreSQL. Expectation-style checks, including a pinned Great Expectations 1.x example, state what a batch must look like before publish. DAG tests load the Airflow DagBag and prove the graph still has the edges from Apache Airflow Architecture and DAGs. A continuous integration workflow runs those tests on every change. A continuous delivery step promotes the same artifact the tests imported, with a migration and a DAG pause policy that does not invent a second copy of the SQL.

Nothing here requires a private test framework. pytest, PostgreSQL, Airflow's public DagBag, and Great Expectations' documented 1.x objects are enough. When the cleanse moves to Spark in Module 3, the pytest assertions on small rows stay the specification. Spark tests will add partition and shuffle behavior. They should not get a new definition of a cent.

Learning Objectives

By the end of this lesson, you should be able to:

1. Place tests on a pyramid: pure functions, database merges, expectations, DAG structure, and a narrow integration path.
2. Write pytest cases for money parsing, quarantine, dedup conflicts, and SCD change emission.
3. Prove a fact merge is idempotent by running it twice in a transaction you can roll back.
4. Express batch checks as expectations: not null, unique grain, accepted domain, and row-count bounds.
5. Use a pinned Great Expectations 1.x context for one real validation, and keep the revenue rule in pytest as well.
6. Load a DagBag in CI and fail on import errors, missing tasks, or a broken edge.
7. Keep DAG tests from requiring production credentials.
8. Describe a CI pipeline that runs unit tests and DAG import on every pull request.
9. Promote one artifact to staging and production without editing business dates by hand.
10. Explain what not to run in CI: a production backfill, a live dataset event, a test against prod data.
11. Seed fixtures that include the pack-price ambiguity and a closed-period row.
12. Say which tests remain the contract when the worker becomes Spark.

Prerequisites

• Data Cleansing and Transformation: the functions under test.
• Apache Airflow Architecture and DAGs: tlp_orders_daily, TaskFlow, reconcile SQL.
• Advanced Airflow Operations: retries, datasets, pools. The DAG test should notice if publish loses its outlet.
• Data Lineage and Metadata Management: names and the idea that a green task is not a test.
• pytest, and a local PostgreSQL for the merge test. Airflow 2.9 installed in the CI job that imports DAGs.

Detailed Core Concepts

The pyramid for a data pipeline

A pipeline fails in different ways, and one kind of test cannot see all of them.

Most defects in this module are pure-unit defects. The float cast does not need a scheduler to be wrong. Teams that only "test in Airflow" pay minutes per run and still forget the fixture. Teams that only unit-test functions never notice that publish_freshness lost its edge to reconcile.

Run the top of the pyramid on every commit. Run DAG import on every commit. Run one staging DagRun before production promotion, not on every keystroke. Do not run a backfill in CI.

Fixtures are the specification

A fixture is a small input with an expected output, stored in the repo. For TechLearningPro the canonical fixtures include:

• A clean line, 49.90 text to 4990 cents.
• A float that must be rejected if it arrives already coerced, and a string that must not pass through float.
• Two deliveries of the same order_line_id with equal cents and different timestamps: one survivor, the later timestamp.
• Two deliveries with different cents: quarantine CONFLICTING_MEASURE, no survivor.
• The ambiguous pack row from lesson 18: quarantine, not 12 * 58800.
• A learner feed with two countries: two change rows, not one.
• A closed-period correction: the merge does not change the frozen row.

If a fixture is missing, the rule is a story. Put the story in tests/fixtures/ as JSON or CSV so the Python test and a future Spark test can read the same file. Sharing the file is how the specification survives the engine change.

Determinism and time

Tests that call datetime.now() for a business date will flake around midnight and in CI timezones. Pass the timestamp in the fixture. Assert UTC conversions with an offset in the string, as the reseller file has. Freeze "today" only if the code under test is the alarm DAG's comparison function, and pass a clock into that function (def is_late(now, deadline)) so the test does not patch globals.

Idempotence tests are ordinary assertions

Run the merge. Record the row, the hash, and loaded_at. Run the merge again with the same winners. Assert one row, the same loaded_at, and an update count of zero. Then change the cents in staging, run again, and assert one row, new cents, new hash, and a new loaded_at. Then mark the business date frozen and assert the third change is quarantined and the fact row is unchanged. That sequence is the whole idempotency contract. It does not require Airflow. It requires a database transaction. Roll it back or use a schema created for the test, so developers do not share state.

Expectations

An expectation is a named predicate on a batch with a structured result: success, unexpected count, and enough detail to debug. Great Expectations popularized the vocabulary: ExpectColumnValuesToNotBeNull, ExpectColumnValuesToBeUnique, ExpectColumnValuesToBeInSet. The idea is older than the library. You can implement the predicates in pytest and also run them in Great Expectations if you want data docs and a stored suite.

Use expectations for properties that are about the batch as a whole:

• order_line_id not null and unique on the winners table.
• extended_amount_cents equals quantity * unit_price_cents - discount_cents.
• country_code in the ISO set or null when optional.
• Row count above a floor on a business date that the manifest says is non-empty, and below a ceiling that would mean a join fan-out.

Do not use an expectation as a substitute for the reconciliation against the payment provider. Uniqueness can pass on a table that is short by a cent on every row. Keep both. The expectation guards shape. The reconcile guards the money identity you agreed with finance.

Thresholds belong in the test as named constants, not as magic numbers copied into a DAG and a notebook. The 0.5 percent quarantine revenue threshold from lesson 19 should be one constant imported by the task and by the test.

Great Expectations 1.x, pinned

Great Expectations 0.18 and 1.x do not share the same context API. Pin the major version. The snippet in the walkthrough uses the 1.x public surface: gx.get_context(mode="ephemeral"), context.data_sources.add_pandas, an expectation class such as gx.expectations.ExpectColumnValuesToNotBeNull, and a validation against a pandas batch. Confirm the method names against the docs for the patch you install. If they differ, the pytest copies of the same predicates still gate CI. The library is a runner and a report. The rule is the predicate.

Do not let a checkpoint exit code alone decide the build until you have verified, on your version, that a failed expectation returns non-zero. Some integrations have historically exited zero and stored a failed result in JSON. Assert result.success in Python.

DAG tests

airflow.models.DagBag loads a folder and records import errors.

pythonCopy
from airflow.models import DagBag

bag = DagBag(dag_folder="dags", include_examples=False)
assert bag.import_errors == {}

Set environment variables before Airflow imports so CI does not look for a production database:

pythonCopy
os.environ["AIRFLOW_HOME"] = str(tmp_path)
os.environ["AIRFLOW__CORE__LOAD_EXAMPLES"] = "False"
os.environ["AIRFLOW__CORE__UNIT_TEST_MODE"] = "True"
os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = f"sqlite:///{tmp_path}/airflow.db"

On Airflow 2.3 and later the database URI setting lives in the database section as sql_alchemy_conn. Set this before the Airflow import. DAG tests should not use conn_id="warehouse" to connect. They only check that the DAG defines that connection id. A test that needs a real warehouse is an integration test and belongs on staging.

Assert structure:

• tlp_orders_daily exists.
• Task ids include cleanse_orders, load_dim_learner, load_dim_course, load_fact_order_line, reconcile_revenue, publish_freshness.
• load_fact_order_line is downstream of the dimension tasks and the cleanse. In Airflow, task.downstream_task_ids and task.upstream_task_ids are the public sets.
• publish_freshness is downstream of reconcile_revenue.
• catchup is false.
• reconcile_revenue has retries == 0.
• The dataset outlet on publish, if you added it in lesson 21, is present: inspect task.outlets.

These assertions fail the pull request when someone wires all_done incorrectly or drops a task during a refactor. They do not prove the SQL is correct. The database test does that.

What CI runs, and what only staging runs

On every pull request:

• Lint and format check.
• pytest tests/unit for pure functions.
• pytest tests/sql against ephemeral Postgres if the project has Docker in CI; otherwise a documented local job, and CI still runs the pure tests. Prefer ephemeral Postgres so the merge test is not optional.
• pytest tests/dags for DagBag.
• Secret scan: no URIs with passwords, no password= in DAG files.

On merge to the release branch:

• Build an immutable artifact: a git SHA, a container image or a versioned DAG bundle.
• Deploy that artifact to staging.
• Run one staging DagRun for a fixed fixture date, not for "yesterday in prod."
• Check ops.dataset_freshness on staging and the import-error command on the staging scheduler.

On promotion to production:

• Deploy the same SHA.
• Do not re-run unit tests as the only gate and then copy a different folder to prod.
• Unpause or deploy with is_paused_upon_creation left in a known state. A deploy that unpauses a new DAG with catchup=True is the surprise backfill from lesson 20. CI should have failed that DAG already. The deploy script should still refuse a DAG whose catchup is true unless an allow-list says otherwise.
• Migrations of warehouse tables run as a separate, reviewed step before the DAG that depends on them. A DAG deploy that also ALTERs the fact table inside a task on first run is a migration with no review.

CI does not:

• Backfill production.
• Emit a production OpenLineage event that marks a dataset complete.
• Update the production Airflow Dataset.
• Read production replicas "just to test counts" unless that job is a read-only contract check with a stated purpose and a role that cannot write.

Configuration and environments

Staging and production differ by connection, bucket, and catalog namespace, not by a fork of parse_line. The DAG reads conn_id values that resolve differently because the Airflow deployment differs. Tests pass a fake connection id and never resolve it. Feature flags that change the cents formula in production and not in CI will recreate the original bug. If a flag must exist, the test matrix includes both values, or the flag is not allowed to change money.

Spark later

The fixture files and the expected cents, quarantine codes, and survivor ids are the contract. A Spark port reads the same fixtures, runs the same functions or a reviewed port, and asserts the same outputs. DAG tests still check that cleanse_orders is upstream of the fact load after the body becomes a submit. Add a Spark-specific test for "one partition does not drop a key" only when you have a Spark runtime in CI. Do not block the unit gate on a cluster. Big Data Fundamentals is where the runtime enters the course. This lesson is why that runtime has a specification to meet.

Architecture and Design Diagrams

mermaidCopy
flowchart TD
    pr[Pull request] --> unit[pytest pure rules]
    pr --> sql[pytest merge on ephemeral Postgres]
    pr --> expect[expectation predicates]
    pr --> bag[DagBag import and edges]
    unit --> mergeGate[Merge allowed]
    sql --> mergeGate
    expect --> mergeGate
    bag --> mergeGate
    mergeGate --> artifact[Immutable SHA artifact]
    artifact --> staging[Staging Airflow and warehouse]
    staging --> oneRun[One fixture DagRun]
    oneRun --> prod[Production deploy of the same SHA]

Production is not on the pull-request path. The fixture DagRun is staging. The assertions in unit and SQL tests are the ones that would have failed the float cast before it shipped.

mermaidCopy
flowchart LR
    fixtures[tests/fixtures CSV and JSON] --> py[CPython cleanse]
    fixtures --> later[Spark cleanse later]
    py --> same[Same expected cents and reasons]
    later --> same
    dagTest[DagBag] --> edges[Edges and retries and outlets]

Two engines, one fixture. The DAG test stays even after the engine changes, because the graph is still the control plane.

Technical Explanation

Pure tests

pythonCopy
import json
from pathlib import Path

import pytest

from tlp.cleanse import parse_line, Quarantine, dedupe_lines

FIXTURES = Path("tests/fixtures")

def test_price_text_becomes_cents():
    row = json.loads((FIXTURES / "simple_line.json").read_text())
    parsed = parse_line(row, source_row=1)
    assert parsed.unit_price_cents == 4990
    assert parsed.extended_amount_cents == 4990

def test_pack_price_is_quarantined():
    row = json.loads((FIXTURES / "ambiguous_pack.json").read_text())
    parsed = parse_line(row, source_row=3)
    assert isinstance(parsed, Quarantine)
    assert parsed.reason_code == "AMBIGUOUS_PACK_PRICE"

def test_conflicting_money_has_no_survivor():
    rows = json.loads((FIXTURES / "conflict_pair.json").read_text())
    winners, quarantined = dedupe_lines(rows)
    assert winners == []
    assert quarantined[0].reason_code == "CONFLICTING_MEASURE"

def test_learner_feed_keeps_both_countries():
    from tlp.cleanse import learner_changes_from_feed
    changes = learner_changes_from_feed(
        json.loads((FIXTURES / "learner_two_countries.json").read_text())
    )
    assert len(changes) == 2
    assert {c.country_code for c in changes} == {"IN", "US"}

dedupe_lines and learner_changes_from_feed are the data-plane functions. They do not import Airflow. That split from lesson 20 is what makes this file fast.

A test that expects an exception uses pytest.raises. A float input:

pythonCopy
def test_float_money_is_refused():
    with pytest.raises(Exception) as caught:
        parse_line({"unit_price": 49.90, "order_line_id": "L1"}, source_row=1)
    # Prefer a typed ParseError in production code and match the reason code.
    assert "MONEY_WAS_FLOAT" in str(caught.value) or getattr(caught.value, "reason", "") == "MONEY_WAS_FLOAT"

Better: parse_line returns a Quarantine for bad rows and raises only for programmer errors. Then the assertion is reason_code == "MONEY_WAS_FLOAT". Pick one policy and test that policy. Lesson 18 returned Quarantine for parse failures. Match it.

Merge test against PostgreSQL

Use a session-scoped database URL from the environment, defaulting to a local test instance, and a transaction that rolls back. The pattern with pytest and a connection:

pythonCopy
import os
import psycopg

@pytest.fixture
def conn():
    url = os.environ["TEST_DATABASE_URL"]
    with psycopg.connect(url) as connection:
        connection.execute("SET search_path TO test_warehouse, public")
        yield connection
        connection.rollback()

Apply DDL once per session from the same SQL files the DAG uses, into schema test_warehouse. Then:

pythonCopy
def test_merge_twice_is_stable(conn):
    load_winners(conn, "tests/fixtures/simple_line.json")
    merge_fact(conn)
    first = fetch_fact(conn, "L1")
    merge_fact(conn)
    second = fetch_fact(conn, "L1")
    assert first.extended_amount_cents == 4990
    assert second.loaded_at == first.loaded_at
    assert count_facts(conn, "L1") == 1

If loaded_at changes, the ON CONFLICT update ran without the hash predicate. That is a regression test with a precise failure.

Expectation predicates in plain Python

pythonCopy
from dataclasses import dataclass

@dataclass(frozen=True)
class ExpectationResult:
    success: bool
    expectation: str
    unexpected_count: int
    detail: str

def expect_unique(rows: list[dict], column: str) -> ExpectationResult:
    values = [r[column] for r in rows]
    unique = len(values) == len(set(values))
    return ExpectationResult(
        success=unique and all(v is not None for v in values),
        expectation=f"unique_not_null:{column}",
        unexpected_count=len(values) - len(set(values)),
        detail=column,
    )

def expect_cents_identity(rows: list[dict]) -> ExpectationResult:
    bad = [
        r for r in rows
        if r["extended_amount_cents"]
        != r["quantity"] * r["unit_price_cents"] - r["discount_cents"]
    ]
    return ExpectationResult(
        success=not bad,
        expectation="extended_amount_identity",
        unexpected_count=len(bad),
        detail="quantity * unit_price_cents - discount_cents",
    )

These are the gates. They do not depend on a library release. The test fails CI with assert result.success, result.

Great Expectations 1.x beside the predicate

Pin great_expectations to a 1.x release in requirements-dev.txt. The following matches the 1.x flow documented for pandas batches: an ephemeral context, a pandas data source, a dataframe asset, a whole-dataframe batch definition, and ExpectColumnValuesToNotBeNull. Verify names against the version you pin before treating a failure as a product bug.

pythonCopy
import great_expectations as gx
import pandas as pd

def test_gx_order_line_id_not_null():
    frame = pd.DataFrame(
        [{"order_line_id": "L1", "extended_amount_cents": 4990}]
    )
    context = gx.get_context(mode="ephemeral")
    source = context.data_sources.add_pandas(name="orders")
    asset = source.add_dataframe_asset(name="winners")
    batch_definition = asset.add_batch_definition_whole_dataframe(name="full")
    suite = context.suites.add(gx.ExpectationSuite(name="winners"))
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="order_line_id")
    )
    validation = context.validation_definitions.add(
        gx.ValidationDefinition(
            name="winners_validation",
            data=batch_definition,
            suite=suite,
        )
    )
    result = validation.run(batch_parameters={"dataframe": frame})
    assert result.success

If your pinned patch uses a slightly different builder, keep this test in a module that can be skipped with a reason, and do not skip expect_unique. The identity of revenue is the pytest predicate. Great Expectations is how you scale the same idea to more columns and to data docs later.

A production task may call the same suite against staged winners and raise AirflowFailException when success is false, which ties lesson 21 to this gate. The suite in git is the suite in the task. Loading a different suite from a UI that nobody reviews forks the rule.

DagBag test

pythonCopy
import os
from pathlib import Path

def test_orders_dag_structure(tmp_path, monkeypatch):
    monkeypatch.setenv("AIRFLOW_HOME", str(tmp_path))
    monkeypatch.setenv("AIRFLOW__CORE__LOAD_EXAMPLES", "False")
    monkeypatch.setenv("AIRFLOW__CORE__UNIT_TEST_MODE", "True")
    monkeypatch.setenv(
        "AIRFLOW__DATABASE__SQL_ALCHEMY_CONN",
        f"sqlite:///{tmp_path}/airflow.db",
    )
    from airflow.models import DagBag

    bag = DagBag(
        dag_folder=str(Path("dags").resolve()),
        include_examples=False,
    )
    assert bag.import_errors == {}
    dag = bag.get_dag("tlp_orders_daily")
    assert dag is not None
    assert dag.catchup is False
    fact = dag.get_task("load_fact_order_line")
    assert "cleanse_orders" in fact.upstream_task_ids
    assert "load_dim_learner" in fact.upstream_task_ids
    assert "load_dim_course" in fact.upstream_task_ids
    reconcile = dag.get_task("reconcile_revenue")
    assert reconcile.retries == 0
    assert "reconcile_revenue" in dag.get_task("publish_freshness").upstream_task_ids

monkeypatch must run before the Airflow import. If another test imported Airflow first, configuration may already be cached. Give DAG tests a separate pytest process or import Airflow only inside the test after the env is set. That is a common source of "works alone, fails in the suite."

Continuous integration workflow

A GitHub Actions workflow illustrates the gate. The same steps translate to any CI system.

yamlCopy
name: pipelines
on:
  pull_request:
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: tlp_test
        ports: ["5432:5432"]
        options: >-
          --health-cmd "pg_isready -U postgres"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install
        run: |
          pip install -r requirements.txt -r requirements-dev.txt
      - name: Unit and SQL tests
        env:
          TEST_DATABASE_URL: postgresql://postgres:postgres@localhost:5432/tlp_test
        run: pytest tests/unit tests/sql tests/expectations -q
      - name: DAG import
        run: pytest tests/dags -q

Pin Airflow using the official constraints file for that Airflow version and Python version when you install requirements.txt. An unconstrained Airflow install resolves provider versions that do not match production and makes the DagBag test a fiction.

Delivery

Tag the image or the DAG bundle with the git SHA. The staging deploy checks out that SHA. The production deploy checks out that SHA. A shell script that scps a working tree is not a promotion, because the working tree can differ from the commit CI tested. Store the SHA on ops.dataset_freshness or on the DagRun note so an incident can answer "which code wrote Tuesday."

Warehouse migrations ship as numbered SQL files applied by a migration tool you already trust, before the DAG rolls forward. The DAG assumes the schema. It does not migrate itself on a green task, because a partial ALTER plus a retry is how you get two meanings of a column.

Practical Implementation or Design Walkthrough

Problem. Add a gate that would have failed the float cast, the pack multiply, a non-idempotent merge, and a DAG edit that lets facts run without dimensions.

Repository layout.

textCopy
tlp/cleanse.py
tlp/warehouse.py
dags/tlp_orders_daily.py
dags/sql/reconcile_order_lines.sql
tests/fixtures/simple_line.json
tests/fixtures/ambiguous_pack.json
tests/fixtures/conflict_pair.json
tests/fixtures/learner_two_countries.json
tests/unit/test_cleanse.py
tests/sql/test_merge.py
tests/expectations/test_batch.py
tests/dags/test_orders_dag.py
requirements.txt
requirements-dev.txt

Steps.

1. Move parse, dedupe, and learner-change functions into tlp/cleanse.py with no Airflow import. The DAG wraps them.
2. Add the four fixtures from real lesson 18 rows, small enough to read in a review.
3. Write the unit tests. Run pytest tests/unit -q. They pass.
4. Introduce float(row["unit_price"]) on purpose. Show the unit test fail. Revert. That is the demonstration that CI is attached to the right function.
5. Apply DDL to test Postgres. Write test_merge_twice_is_stable and the frozen-period test.
6. Add expect_unique and expect_cents_identity. Point them at the winner rows the cleanse returns.
7. Add the optional Great Expectations test pinned to 1.x, asserting result.success.
8. Add the DagBag test. Delete the edge from load_dim_learner to the fact task on a branch and show the test fail.
9. Wire the GitHub Actions workflow, or the equivalent in your CI, including TEST_DATABASE_URL.
10. Document promotion: SHA to staging, one fixture DagRun, same SHA to production, migrations first.

Expected output. Pull requests cannot merge with a failing predicate or a broken edge. Staging proves the scheduler can import the artifact. Production receives only that artifact.

Enterprise Real-World Scenarios

The float cast never reaches the reseller file again

After the backfill, the platform added test_price_text_becomes_cents and a test that a Python float becomes MONEY_WAS_FLOAT. The next attempt to "simplify" parsing failed CI in under a minute. The author saw the fixture, not a finance ticket. The cost of the gate is a pytest job. The cost of missing it was eight days of warehouse writes and a flash report correction. The DAG structure tests would not have caught this, which is why they sit beside the unit tests and not instead of them. The lineage facet from lesson 22 still helps when a bug ships. The test is what stops the ship.

A well-meaning trigger rule change

Someone set the fact task's trigger rule so a late catalog would not block revenue. The DagBag test asserted load_dim_course in upstream_task_ids and also, in a second assertion, that the fact task's trigger_rule remained all_success. The pull request failed. The author had a real operational complaint: catalog landings were slow. The repair was a pool and a faster catalog task, not a weaker trigger. Without the assertion, staging would have looked fine on days the catalog succeeded, and the first missed catalog would have published facts against an old dimension. Structural tests encode the lesson 19 edges so incident pressure cannot delete them quietly.

Promoting Spark without forking the rule

Module 3 replaces the body of cleanse with a Spark job. CI continues to run the pytest fixtures in CPython on the shared functions, and adds a job that runs those fixtures through Spark when the cluster image changes. The expected JSON is the same file. The first Spark port used a float cast inside a UDF "because the column was numeric." The shared fixture failed in the Spark job's test. The DAG test still required cleanse_orders upstream of the fact load and retries unchanged on reconcile. Delivery promoted one SHA that contained both the DAG and the Spark job definition. A second path that copied a notebook to production was rejected because it was not the SHA. The freshness row records that SHA. An incident responder can see whether Tuesday ran before or after the port.

Best Practices

• Test pure functions without importing Airflow. Pass dates in. Do not call date.today() in the code under test.
• Store fixtures in git and reuse them for a future Spark port.
• Include the ugly rows: conflicts, ambiguous packs, naive timestamps, frozen dates.
• Assert idempotence with a second execution and a stable loaded_at.
• Keep revenue identity and uniqueness as pytest predicates even if you also run Great Expectations.
• Pin Great Expectations to 1.x or 0.18, not both, and assert success in process.
• Load DagBag with examples off and a local sqlite metadata URI set before import.
• Assert edges, catchup, retries on reconcile, and dataset outlets.
• Run unit, SQL, expectation, and DAG tests on the pull request.
• Promote an immutable SHA. Migrate schema before the DAG that needs it.
• Record the SHA on the freshness row.
• Keep secrets out of CI logs. Use a throwaway Postgres password for the ephemeral service.
• Do not backfill production from CI.
• When a threshold exists in the task, import the same constant in the test.

Common Mistakes and Anti-Patterns

Only testing the happy path line. The float and the pack and the conflict are the bugs you already had. Fixtures for them are the test.

Hitting production in a unit test. It will be flaky, slow, and one day destructive. Ephemeral Postgres and fixtures.

Asserting a row count of "whatever the query returns" by snapshotting prod. The test freezes today's accidents. Assert identities and invariants.

A DAG test that connects to the warehouse. It fails closed on laptops and tempts people to skip it. Structure only. Staging covers execution.

Importing Airflow before setting AIRFLOW_HOME. Configuration sticks to the first import. Set the env, then import, in one test module.

Trusting a shell exit code from a data-quality tool you have not checked. Assert the success flag. Confirm the version's exit behavior before you depend on it.

Two copies of the cents formula, in the test and in the job, edited separately. Import the function and the constant. The test calls the function. It does not reimplement it, except for the expected literal 4990, which is the oracle.

CI deploys a dirty working tree. Promote the SHA that passed.

A task that alters tables on startup. Migrations belong in a reviewed step. A retried ALTER is not a merge.

Disabling test_orders_dag_structure to get a hotfix through. The hotfix is how facts lose a parent. Fix the edge or explicitly change the assertion in the same pull request so the new graph is reviewed.

Running the full backfill in CI "to be sure." It is slow, it needs production-sized data, and it writes. Run fixtures. Backfill is an operation with a human, from lesson 21.

Troubleshooting and Design Review

Design review for the gate:

1. Which fixture would fail a float cast, a pack multiply, and a conflicting duplicate?
2. Does the merge test run twice and include a frozen date?
3. Which assertion fails if facts no longer wait for dimensions?
4. Is Great Expectations pinned, and is the revenue identity still in pytest?
5. Does CI use ephemeral Postgres and a sqlite Airflow metadata DB?
6. What artifact id does production run, and where is it recorded?
7. Which tests will the Spark port be required to pass without editing expected cents?

Interview Questions and Answers

Q1. Why is a green Airflow task not a test?

A task succeeds when the process exits zero for that run's data. It does not remember the fixture you meant, and it will not run the ambiguous pack row if today's file happens to be clean. Tests pin inputs and outputs and run on the change that introduces a bad cast. The grid is an operational record. The suite is the gate. You want both, and they fail for different reasons.

Q2. What belongs in a pure unit test versus a database test?

Pure tests cover parsing, quarantine reasons, dedup winners, and SCD change lists, with no network and no Airflow. Database tests cover the merge: one row after two runs, stable loaded_at when the hash matches, and no update when the period is frozen. If you only do the first, a bad ON CONFLICT clause ships. If you only do the second, you pay a database startup to learn that 49.90 parsed wrong, and you will run that case less often.

Q3. How do you test idempotence without a flaky timestamp?

Run the merge twice in one test transaction. Compare the primary key count, the cents, the hash, and loaded_at from the database. The second run must not change loaded_at when staging is unchanged. A third run with new cents must change them. A frozen business date must reject the correction. Roll back the transaction so order does not matter across tests.

Q4. What is an expectation, and why is it not the reconciliation?

An expectation is a predicate on a batch with a structured pass or fail, such as unique order_line_id or the cents identity. It catches shape and formula violations inside the table. Reconciliation compares the table to an external total from the payment provider. A unique, internally consistent table can still be short versus the provider if a line never landed. Keep the expectation and the reconcile SQL. They fail different incidents.

Q5. How do you use Great Expectations without betting the pipeline on one API?

Pin a major version. On 1.x, build an ephemeral context, a pandas data source, a suite of expectation objects, and assert result.success on a validation run. Read the docs for the patch you pin, because 0.18's context API is different. Keep the same rules as plain pytest predicates so a library bump cannot silently drop the cents identity from CI. Do not trust an exit code you have not checked. A failed expectation must fail the job in code you control.

Q6. What should a DagBag test assert for tlp_orders_daily?

No import errors, the DAG is present, catchup is false, the fact task is downstream of cleanse and both dimension loads, reconcile has zero retries, and publish is downstream of reconcile. If lesson 21 added a dataset outlet, assert it on the publish task. Set Airflow's metadata URI to sqlite and disable examples before importing Airflow. Do not open the production connection.

Q7. Why do DAG tests import Airflow after setting environment variables?

Airflow reads configuration early and caches it. Setting AIRFLOW_HOME and sql_alchemy_conn after the import does nothing useful, and the test may try to use a developer's real Airflow home or hang on a database. monkeypatch the variables, then import DagBag inside the test. Isolate this from tests that imported Airflow at module top.

Q8. What runs on a pull request, and what waits for staging?

Pull requests run unit tests, merge tests on ephemeral Postgres, expectation tests, and DagBag. Staging receives the immutable SHA and runs one fixture DagRun to prove the scheduler, the connection, and the SQL dialect. Production receives the same SHA after that. A full historical backfill is an operational action with a date range and a human, not a CI step.

Q9. How do you stop CI from deploying code that was not tested?

Build an artifact named by git SHA from the commit the tests ran on. Staging and production deploy that artifact, not a branch tip and not a copied directory. Record the SHA on the freshness row. A dirty tree, a notebook copied to the cluster, or a hot patch on the server breaks the chain. If you must hot patch, commit it and let the gate run, or accept that the environment and the repo have diverged and say so in the incident.

Q10. Where do schema migrations fit?

In reviewed SQL applied before the DAG version that depends on the new shape. The DAG should fail fast if the column is missing, not issue ALTER TABLE on its first task. Retries and partial alters do not have the idempotent merge's safety. CI can apply the same migration files to ephemeral Postgres before the merge test, which is how you notice a migration that does not match the code.

Q11. A test asserts len(rows) == 18420 by copying production. What is wrong, and what do you assert instead?

The number is an accident of one day, the test needs production data, and a legitimate volume change fails the build. Assert invariants: the grain key is unique, the cents identity holds, quarantined packs are not in the winners, and a second merge does not duplicate keys. Use a row-count floor only when a manifest declares that the batch is non-empty, and pair it with a ceiling that would indicate a join fan-out.

Q12. What must a Spark port pass before it replaces cleanse_orders?

The same fixtures and expected cents, quarantine codes, dedup survivors, and SCD change counts. The DagBag test still requires the task's place in the graph, zero retries on reconcile, and the publish outlet. Spark-only tests can add partition behavior later. They do not get a private definition of extended_amount_cents. The expected JSON in git is the contract Module 3 has to meet.

Hands-On Exercises

Exercise 1. Fixtures and the merge

Problem. Implement the unit and SQL tests for the lesson 18 rules.

Tasks.

1. Add fixtures for a simple line, an ambiguous pack, a cents conflict, and two learner countries.
2. Make the unit tests pass on the pure functions.
3. Break the parser with float and show the failure. Restore it.
4. Run the merge twice and show a stable loaded_at.
5. Add a frozen-date case that does not change the fact.

Acceptance. All five behaviors are asserted without Airflow. The float regression fails the unit test.

Exercise 2. DagBag and a workflow

Problem. Guard the graph and describe promotion.

Tasks.

1. Write the DagBag test for edges, catchup, and reconcile retries.
2. Remove a dimension edge locally and show the test fail.
3. Add a CI job definition that runs unit tests and the DagBag test.
4. State how the SHA reaches staging and production, and where it is recorded.
5. List one Spark-port command you will not allow: a notebook path around the fixture file.

Rubric.

Lesson Summary

The gate in front of the daily pipeline is a set of fixtures and assertions: cents and quarantine in pure pytest, an idempotent merge against PostgreSQL, expectation predicates on the batch, and a DagBag test that the fact load still waits for dimensions and that reconcile does not retry. Great Expectations 1.x can run the same class of check if you pin it and assert success in Python. The revenue identity stays in the repo either way.

CI runs those tests on the pull request. Delivery promotes the SHA that passed, migrates schema first, and proves the artifact once on staging. Production does not get a second definition of the transform and does not get a backfill from the build agent. When Module 3 moves cleanse to Spark, these fixtures remain the definition of a correct row. The cluster is a new runtime. It is not a new cent, a new grain, or a new excuse to drop the edge between dimensions and facts.

Further Reading

• pytest, fixtures and monkeypatch — the mechanism the DagBag test uses to set configuration before import.
• PostgreSQL, INSERT ON CONFLICT — the statement the idempotence test runs twice.
• Apache Airflow, DagBag and the testing guidance for DAG import errors and structure.
• Great Expectations 1.x documentation for get_context, pandas data sources, and ExpectColumnValuesToNotBeNull. Pin the version you install. The 0.18 API is a different surface.
• Apache Airflow constraints files for the 2.9 line — install providers from the constraint that matches your Python version so CI matches production.
• TechLearningPro, Data Cleansing and Transformation and Apache Airflow Architecture and DAGs — the code and the graph under test.
• Next module: Big Data Fundamentals. Take the fixtures with you. The engine changes. The expected rows should not.