Most ETL testing tutorials show you one test. Maybe two. Then they stop.
But a real ETL pipeline has dozens of things that can go wrong — and you need a systematic workflow, not a handful of isolated examples. This guide gives you the complete picture: a structured folder layout, reusable fixtures, parametrized tests for every pipeline stage, and a CI/CD configuration that ties it all together.
By the end, you'll have a workflow for ETL testing automation using Python that you can drop into any ETL project and adapt in minutes. For the bigger picture — frameworks, tools, and CI/CD — see our ETL automation testing guide.
Stack used: Python 3.10+, pytest 7+, pandas 2+, SQLAlchemy 2+. Install with:
pip install pytest pandas sqlalchemy pytest-cov
The Folder Structure
Mirror your pipeline code with your test code. This makes it immediately obvious which test covers which pipeline component:
Project Layout
my_etl_project/
├── pipeline/
│ ├── __init__.py
│ ├── extract.py # Data extraction from source systems
│ ├── transform.py # Business logic and data cleaning
│ └── load.py # Loading to target database / warehouse
├── tests/
│ ├── __init__.py
│ ├── conftest.py # Shared fixtures — the heart of the workflow
│ ├── test_extract.py # Tests for the Extract stage
│ ├── test_transform.py # Tests for the Transform stage
│ └── test_load.py # Tests for the Load stage
├── requirements.txt
└── pytest.ini # pytest configurationConfigure pytest to find your tests consistently:
pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_functions = test_*
addopts = -v --tb=shortThe conftest.py — Your Fixture Hub
conftest.py is automatically loaded by pytest. Define all shared test data here so every test file can access it without imports:
tests/conftest.py
import pytest
import pandas as pd
from sqlalchemy import create_engine
# ── Schema definitions ────────────────────────────────────────────────────────
@pytest.fixture(scope="session")
def required_source_columns():
return ['order_id', 'customer_id', 'product_id', 'quantity',
'unit_price', 'order_date', 'status']
@pytest.fixture(scope="session")
def required_non_null_columns():
return ['order_id', 'product_id', 'quantity', 'unit_price', 'order_date']
# ── Sample DataFrames ─────────────────────────────────────────────────────────
@pytest.fixture
def clean_orders():
"""Well-formed source data — all tests should pass against this."""
return pd.DataFrame({
'order_id': [1001, 1002, 1003, 1004, 1005],
'customer_id': [201, 202, 203, 204, 205],
'product_id': [301, 302, 303, 304, 305],
'quantity': [2, 1, 5, 3, 1],
'unit_price': [49.99, 199.00, 12.50, 89.99, 299.00],
'order_date': pd.to_datetime([
'2024-01-10', '2024-01-11', '2024-01-12',
'2024-01-13', '2024-01-14'
]),
'status': ['completed', 'completed', 'pending', 'completed', 'cancelled']
})
@pytest.fixture
def orders_with_nulls():
"""Source data with NULL values in critical columns."""
return pd.DataFrame({
'order_id': [1001, 1002, None, 1004],
'customer_id': [201, None, 203, 204],
'product_id': [301, 302, 303, None],
'quantity': [2, 1, 5, 3],
'unit_price': [49.99, 199.00, None, 89.99],
'order_date': pd.to_datetime(['2024-01-10', '2024-01-11', None, '2024-01-13']),
'status': ['completed', 'completed', 'pending', 'completed']
})
@pytest.fixture
def orders_with_duplicates():
"""Source data with duplicate primary key values."""
return pd.DataFrame({
'order_id': [1001, 1001, 1002, 1003], # 1001 is a duplicate
'customer_id': [201, 201, 202, 203],
'product_id': [301, 301, 302, 303],
'quantity': [2, 2, 1, 5],
'unit_price': [49.99, 49.99, 199.00, 12.50],
'order_date': pd.to_datetime([
'2024-01-10', '2024-01-10', '2024-01-11', '2024-01-12'
]),
'status': ['completed', 'completed', 'completed', 'pending']
})
@pytest.fixture
def orders_raw_strings():
"""Raw source data before type conversion — as it arrives from some APIs."""
return pd.DataFrame({
'order_id': ['1001', '1002', '1003'],
'customer_id': ['201', '202', '203'],
'product_id': ['301', '302', '303'],
'quantity': ['2', '1', '5'],
'unit_price': ['$49.99', '$199.00', '$12.50'],
'order_date': ['01/10/2024', '01/11/2024', '01/12/2024'],
'status': ['completed', 'completed', 'pending']
})
# ── In-memory database ────────────────────────────────────────────────────────
@pytest.fixture(scope="session")
def db_engine():
"""SQLite in-memory engine for load tests — no external DB required."""
engine = create_engine("sqlite:///:memory:")
yield engine
engine.dispose()
@pytest.fixture
def customers_dim():
"""Simulated customer dimension table for FK validation."""
return pd.DataFrame({
'customer_id': [201, 202, 203, 204, 205],
'customer_name': ['Alice', 'Bob', 'Carol', 'David', 'Eve'],
'region': ['North', 'South', 'East', 'West', 'North']
})test_extract.py — The Extract Stage
Extract tests validate that data arriving from the source is structurally correct before any transformation happens:
tests/test_extract.py
import pytest
import pandas as pd
# ── Schema tests ──────────────────────────────────────────────────────────────
def test_all_required_columns_present(clean_orders, required_source_columns):
"""Catch schema drift — source renamed or removed a column."""
missing = [c for c in required_source_columns if c not in clean_orders.columns]
assert not missing, f"Missing required columns: {missing}"
@pytest.mark.parametrize("col", [
'order_id', 'customer_id', 'product_id',
'quantity', 'unit_price', 'order_date', 'status'
])
def test_each_column_present_individually(clean_orders, col):
"""Parametrized: each required column gets its own test result."""
assert col in clean_orders.columns, f"Column '{col}' not found in source data."
# ── Row count tests ───────────────────────────────────────────────────────────
def test_extract_returns_rows(clean_orders):
"""Guard against empty extracts caused by API failures or wrong date filters."""
assert len(clean_orders) > 0, "Extract returned 0 rows."
def test_extract_count_within_expected_range(clean_orders):
"""Alert if row count is suspiciously low or high."""
min_expected, max_expected = 1, 10_000_000
row_count = len(clean_orders)
assert min_expected <= row_count <= max_expected, (
f"Row count {row_count} is outside expected range [{min_expected}, {max_expected}]."
)
# ── Null tests ────────────────────────────────────────────────────────────────
@pytest.mark.parametrize("col", ['order_id', 'product_id', 'quantity', 'unit_price', 'order_date'])
def test_required_column_has_no_nulls(clean_orders, col):
"""Each required column gets its own null check."""
null_count = clean_orders[col].isna().sum()
assert null_count == 0, (
f"Column '{col}' has {null_count} NULL value(s). This column is required."
)
def test_null_count_in_required_columns_fails_on_bad_data(orders_with_nulls):
"""Verify that our null tests correctly detect problematic data."""
required_cols = ['order_id', 'unit_price', 'order_date']
has_nulls = any(orders_with_nulls[col].isna().any() for col in required_cols)
assert has_nulls, "Expected nulls in required columns but found none — check fixture."test_transform.py — The Transform Stage
Transform tests validate business logic. These are the most important tests — wrong transformation logic is silent and produces numbers that look right but aren't:
pipeline/transform.py
import pandas as pd
def calculate_line_total(df: pd.DataFrame) -> pd.DataFrame:
"""Add a line_total column: quantity * unit_price."""
df = df.copy()
df['line_total'] = df['quantity'] * df['unit_price']
return df
def clean_price(df: pd.DataFrame) -> pd.DataFrame:
"""Convert string prices like '$49.99' to float."""
df = df.copy()
df['unit_price'] = (
df['unit_price']
.astype(str)
.str.replace('$', '', regex=False)
.str.replace(',', '', regex=False)
.str.strip()
)
df['unit_price'] = pd.to_numeric(df['unit_price'], errors='coerce')
return df
def filter_active_orders(df: pd.DataFrame) -> pd.DataFrame:
"""Keep only completed and pending orders; drop cancelled."""
return df[df['status'].isin(['completed', 'pending'])].copy()
def add_order_year_month(df: pd.DataFrame) -> pd.DataFrame:
"""Extract year and month from order_date for partitioning."""
df = df.copy()
df['order_year'] = df['order_date'].dt.year
df['order_month'] = df['order_date'].dt.month
return dftests/test_transform.py
import pytest
import pandas as pd
from pipeline.transform import (
calculate_line_total, clean_price,
filter_active_orders, add_order_year_month
)
# ── Business logic tests ──────────────────────────────────────────────────────
def test_line_total_calculation(clean_orders):
"""line_total = quantity * unit_price."""
result = calculate_line_total(clean_orders)
expected = clean_orders['quantity'] * clean_orders['unit_price']
pd.testing.assert_series_equal(
result['line_total'].reset_index(drop=True),
expected.reset_index(drop=True),
check_names=False
)
def test_line_total_is_always_positive(clean_orders):
"""Business rule: no negative line totals."""
result = calculate_line_total(clean_orders)
assert (result['line_total'] >= 0).all(), "Negative line_total values found."
@pytest.mark.parametrize("raw_price,expected", [
('$49.99', 49.99),
('$1,200.00', 1200.00),
('$0.99', 0.99),
('199.00', 199.00),
])
def test_price_cleaning_parametrized(raw_price, expected):
"""Test clean_price against multiple string formats."""
df = pd.DataFrame({'unit_price': [raw_price]})
result = clean_price(df)
assert abs(result['unit_price'].iloc[0] - expected) < 0.001
# ── Filter tests ──────────────────────────────────────────────────────────────
def test_cancelled_orders_are_filtered_out(clean_orders):
"""Cancelled orders should not reach the warehouse."""
result = filter_active_orders(clean_orders)
assert 'cancelled' not in result['status'].values, (
"Cancelled orders found after filtering — check filter_active_orders()."
)
def test_completed_and_pending_orders_are_kept(clean_orders):
"""Ensure we don't accidentally filter out valid statuses."""
result = filter_active_orders(clean_orders)
valid_statuses = set(result['status'].unique())
assert valid_statuses.issubset({'completed', 'pending'}), (
f"Unexpected status values after filter: {valid_statuses}"
)
def test_row_count_decreases_after_filter(clean_orders):
"""Filtering should reduce (or equal) the original count."""
original_count = len(clean_orders)
filtered_count = len(filter_active_orders(clean_orders))
assert filtered_count <= original_count, (
f"Filter produced MORE rows ({filtered_count}) than original ({original_count})."
)
# ── Date extraction tests ─────────────────────────────────────────────────────
def test_year_month_columns_added(clean_orders):
"""Verify add_order_year_month creates expected columns."""
result = add_order_year_month(clean_orders)
assert 'order_year' in result.columns
assert 'order_month' in result.columns
@pytest.mark.parametrize("col,expected_dtype", [
('order_year', 'int64'),
('order_month', 'int64'),
])
def test_date_derived_columns_are_integers(clean_orders, col, expected_dtype):
"""Year and month should be integer types."""
result = add_order_year_month(clean_orders)
assert str(result[col].dtype) == expected_dtype, (
f"Column '{col}' should be {expected_dtype}, got {result[col].dtype}."
)
def test_transform_does_not_modify_original_dataframe(clean_orders):
"""All transforms should operate on copies, not mutate the input."""
original_cols = list(clean_orders.columns)
_ = calculate_line_total(clean_orders)
assert list(clean_orders.columns) == original_cols, (
"calculate_line_total() mutated the original DataFrame."
)test_load.py — The Load Stage
Load tests validate the data just before and after it enters the target database:
tests/test_load.py
import pytest
import pandas as pd
from sqlalchemy import text
# ── Pre-load validation ───────────────────────────────────────────────────────
def test_no_duplicate_primary_keys_before_load(clean_orders):
"""Primary key must be unique — duplicates cause upsert issues."""
dupes = clean_orders.duplicated(subset=['order_id']).sum()
assert dupes == 0, f"Found {dupes} duplicate order_id value(s) before load."
def test_duplicate_detection_works(orders_with_duplicates):
"""Verify our duplicate check actually catches the bad fixture."""
dupes = orders_with_duplicates.duplicated(subset=['order_id']).sum()
assert dupes > 0, "Expected duplicates in orders_with_duplicates fixture."
@pytest.mark.parametrize("col", ['order_id', 'product_id'])
def test_foreign_key_columns_have_no_nulls_before_load(clean_orders, col):
"""FK columns must not be NULL or they'll violate warehouse constraints."""
null_count = clean_orders[col].isna().sum()
assert null_count == 0, f"Column '{col}' has {null_count} NULL(s) — violates FK constraint."
def test_customer_ids_exist_in_dimension(clean_orders, customers_dim):
"""Referential integrity: every customer_id must exist in the dim table."""
source_ids = set(clean_orders['customer_id'].dropna().astype(int))
dim_ids = set(customers_dim['customer_id'])
orphaned = source_ids - dim_ids
assert not orphaned, (
f"Orphaned customer_id values: {orphaned}. "
"These orders reference non-existent customers."
)
# ── Post-load validation ──────────────────────────────────────────────────────
def test_row_count_matches_after_load(clean_orders, db_engine):
"""Rows loaded to DB should match the DataFrame row count."""
table_name = 'orders_test'
clean_orders.to_sql(table_name, db_engine, if_exists='replace', index=False)
with db_engine.connect() as conn:
result = conn.execute(text(f"SELECT COUNT(*) FROM {table_name}"))
db_count = result.scalar()
assert db_count == len(clean_orders), (
f"Expected {len(clean_orders)} rows in DB, found {db_count}."
)
def test_no_data_loss_on_load(clean_orders, db_engine):
"""Verify that all order_ids are present after loading."""
table_name = 'orders_integrity_test'
clean_orders.to_sql(table_name, db_engine, if_exists='replace', index=False)
with db_engine.connect() as conn:
result = conn.execute(text(f"SELECT order_id FROM {table_name}"))
loaded_ids = {row[0] for row in result}
source_ids = set(clean_orders['order_id'].tolist())
missing_ids = source_ids - loaded_ids
assert not missing_ids, (
f"These order_ids were lost during load: {missing_ids}"
)Running the Full Test Suite
With all test files in place, run the full suite:
Terminal
# Run all tests with verbose output
pytest tests/ -v
# Run only transform tests
pytest tests/test_transform.py -v
# Run with coverage report
pytest tests/ -v --cov=pipeline --cov-report=term-missing
# Run a specific test by name
pytest tests/ -k "test_line_total" -vExpected output when everything passes:
Terminal — Full output
tests/test_extract.py::test_all_required_columns_present PASSED
tests/test_extract.py::test_each_column_present_individually[order_id] PASSED
tests/test_extract.py::test_each_column_present_individually[customer_id] PASSED
tests/test_extract.py::test_each_column_present_individually[product_id] PASSED
tests/test_extract.py::test_each_column_present_individually[quantity] PASSED
tests/test_extract.py::test_each_column_present_individually[unit_price] PASSED
tests/test_extract.py::test_each_column_present_individually[order_date] PASSED
tests/test_extract.py::test_each_column_present_individually[status] PASSED
tests/test_extract.py::test_extract_returns_rows PASSED
tests/test_extract.py::test_extract_count_within_expected_range PASSED
tests/test_extract.py::test_required_column_has_no_nulls[order_id] PASSED
tests/test_extract.py::test_required_column_has_no_nulls[product_id] PASSED
tests/test_extract.py::test_required_column_has_no_nulls[quantity] PASSED
tests/test_extract.py::test_required_column_has_no_nulls[unit_price] PASSED
tests/test_extract.py::test_required_column_has_no_nulls[order_date] PASSED
tests/test_transform.py::test_line_total_calculation PASSED
tests/test_transform.py::test_line_total_is_always_positive PASSED
tests/test_transform.py::test_price_cleaning_parametrized[$49.99-49.99] PASSED
tests/test_transform.py::test_price_cleaning_parametrized[$1,200.00-1200.0] PASSED
tests/test_transform.py::test_price_cleaning_parametrized[$0.99-0.99] PASSED
tests/test_transform.py::test_price_cleaning_parametrized[199.00-199.0] PASSED
tests/test_transform.py::test_cancelled_orders_are_filtered_out PASSED
tests/test_transform.py::test_completed_and_pending_orders_are_kept PASSED
tests/test_transform.py::test_row_count_decreases_after_filter PASSED
tests/test_transform.py::test_year_month_columns_added PASSED
tests/test_transform.py::test_date_derived_columns_are_integers[order_year-int64] PASSED
tests/test_transform.py::test_date_derived_columns_are_integers[order_month-int64] PASSED
tests/test_transform.py::test_transform_does_not_modify_original_dataframe PASSED
tests/test_load.py::test_no_duplicate_primary_keys_before_load PASSED
tests/test_load.py::test_duplicate_detection_works PASSED
tests/test_load.py::test_foreign_key_columns_have_no_nulls_before_load[order_id] PASSED
tests/test_load.py::test_foreign_key_columns_have_no_nulls_before_load[product_id] PASSED
tests/test_load.py::test_customer_ids_exist_in_dimension PASSED
tests/test_load.py::test_row_count_matches_after_load PASSED
tests/test_load.py::test_no_data_loss_on_load PASSED
========== 35 passed in 1.03s ==========Automating with GitHub Actions
Set this up once and your tests run automatically on every push and pull request:
.github/workflows/etl-tests.yml
name: ETL Pipeline Tests
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python 3.11
uses: actions/setup-python@v4
with:
python-version: '3.11'
cache: pip
- name: Install dependencies
run: pip install -r requirements.txt
- name: Run ETL test suite
run: pytest tests/ -v --tb=short
- name: Coverage report
run: pytest tests/ --cov=pipeline --cov-report=term-missing --cov-fail-under=80
- name: Upload coverage
if: always()
uses: actions/upload-artifact@v3
with:
name: coverage-report
path: htmlcov/
What this gives you: Every pull request is blocked from merging unless 100% of tests pass AND pipeline code coverage stays above 80%. This is your safety net for team collaboration.
Where to Go From Here
| Topic | What to Learn | Tool |
|---|---|---|
| Data profiling | Automatically detect schema and statistical anomalies | Great Expectations, Soda Core |
| dbt tests | Run tests inside your transformation layer at the SQL level | dbt (not_null, unique, accepted_values) |
| Contract testing | Formalize source system schema contracts | Pandera, PyDantic |
| Performance tests | Validate pipeline completes within time SLA | pytest-benchmark |
| Mock external APIs | Test extract stage without hitting real source systems | unittest.mock, responses |
Frequently Asked Questions
Use pytest to write test functions organized by pipeline stage. Create shared fixtures in conftest.py for sample DataFrames and expected schemas, then write separate test files for extract (schema/row count checks), transform (business logic/type conversions), and load (duplicate checks/referential integrity). Run with pytest tests/ -v from your project root.
conftest.py is a special pytest file for shared fixtures and configuration. In ETL testing, you define your sample DataFrames, database connections, and expected schemas here so they can be reused across all your test files without any import statements. pytest discovers conftest.py automatically.
Yes. pytest.mark.parametrize is excellent for ETL testing when you want to run the same validation against multiple columns or multiple business rules. For example, checking that 7 required columns all have no nulls can be expressed as a single parametrized test instead of 7 separate functions. Each parametrized case appears as a separate test result in the output.
Mirror your pipeline structure: one test file per pipeline stage. Create test_extract.py, test_transform.py, and test_load.py. Put all shared fixtures in conftest.py. This makes it easy to locate the source of a failing test and understand which stage of the pipeline has an issue.
Asim Noaman Lodhi
Data Engineer & ETL Testing Specialist | Google Partner | 913+ students trained
Author of the Complete ETL Testing Course on Udemy — 79 lectures, 4.5 ★ rating. Covering SQL validation, Python testing, data warehouse QA, and real-world ETL projects.