Python Testing with pytest for Data Engineers: Fixtures, Parametrize, Mocking, conftest.py, Testing ETL Pipelines, Database Code, API Extractors, Coverage, and Production Patterns

Why Data Engineers Need Tests
pytest Basics
Fixtures — Reusable Test Setup
Parametrize — Test Multiple Inputs
Markers — Categorize Tests
Mocking — Isolate External Dependencies
Testing Database Code
Testing File Operations with tmp_path
Testing a Complete Pipeline
Coverage Reporting
Project Structure
Common Mistakes
Interview Questions
Wrapping Up

Our ETL Patterns post covered building production pipelines with extract, transform, and load phases. But how do you know those pipelines actually work? How do you catch bugs before they corrupt your data lake at 3 AM? The answer is testing — and pytest is the Python testing framework that every data engineer should know.

Analogy — A quality control lab in a factory. The factory (your pipeline) produces widgets (transformed data). Without a QC lab (tests), defective widgets ship to customers (bad data reaches dashboards). The QC lab tests each component independently: does the cutting machine cut to spec? (unit test for a transform function). Does the assembly line produce complete widgets? (integration test for the full pipeline). Does the finished widget survive a drop test? (edge case testing with parametrize). pytest is your QC lab — it runs checks automatically every time you change the factory.

Why Data Engineers Need Tests

Data engineering has a testing problem. Many teams write zero tests for their pipelines because “the data changes every day anyway” or “we just check the output manually.” This works until it does not — and the failure mode is always the same: corrupted data silently flows downstream, dashboards show wrong numbers, and nobody notices until a business decision is made on bad data.

Testing data pipelines is different from testing web applications, but it is just as important. Here is what you should test and why:

What to Test	Why	Example
Transform functions	These contain your business logic — type casting, cleaning, deduplication	Does `clean_column_names()` handle spaces, special characters, and mixed case?
Schema expectations	Source schemas change without warning	Does the pipeline fail fast when a required column is missing?
Edge cases	Real data has nulls, empty strings, negative numbers, unicode	Does `cast_types()` handle “N/A” in a numeric column without crashing?
Database writes	Wrong data types or constraint violations fail silently	Does `to_sql()` correctly map Python types to SQL types?
API extractors	APIs change response formats, rate limit, or go down	Does your paginated extractor handle an empty last page?
Idempotency	Reruns should not create duplicates	Does running the load twice produce the same row count?

pytest Basics

pytest discovers and runs test functions automatically. Any function starting with test_ in a file starting with test_ is a test. No boilerplate, no class inheritance, no special imports — just plain assert statements.

Installation and First Test

# Install pytest
# pip install pytest

# tests/test_transforms.py
def test_addition():
    assert 1 + 1 == 2

def test_string_upper():
    assert "hello".upper() == "HELLO"

def test_list_length():
    data = [1, 2, 3]
    assert len(data) == 3

# Run all tests
pytest

# Run with verbose output (shows each test name)
pytest -v

# Run a specific file
pytest tests/test_transforms.py

# Run a specific test function
pytest tests/test_transforms.py::test_addition

# Stop on first failure
pytest -x

# Show print() output (normally captured)
pytest -s

Testing Real Transform Functions

Here is how you test the transform functions from the ETL Patterns post. Each test follows the Arrange-Act-Assert pattern: set up test data (arrange), call the function (act), check the result (assert).

# tests/test_transforms.py
import pandas as pd
from etl.transforms import clean_column_names, deduplicate, cast_types

def test_clean_column_names_removes_spaces():
    # Arrange
    df = pd.DataFrame({"First Name": [1], " Last  Name ": [2], "ORDER-ID": [3]})

    # Act
    result = clean_column_names(df)

    # Assert
    assert list(result.columns) == ["first_name", "last_name", "order_id"]


def test_deduplicate_keeps_last():
    df = pd.DataFrame({
        "order_id": [1, 2, 1, 3],
        "amount": [10, 20, 15, 30]
    })
    result = deduplicate(df, subset=["order_id"], keep="last")
    assert len(result) == 3
    # The duplicate order_id=1 should keep the last occurrence (amount=15)
    assert result[result["order_id"] == 1]["amount"].iloc[0] == 15


def test_cast_types_handles_invalid_values():
    df = pd.DataFrame({"amount": ["100", "N/A", "200", ""]})
    result = cast_types(df, {"amount": float})
    # Valid values should be cast, invalid should become NaN
    assert result["amount"].iloc[0] == 100.0
    assert pd.isna(result["amount"].iloc[1])  # "N/A" becomes NaN


def test_clean_column_names_does_not_modify_original():
    # Transform functions should return a new DataFrame, not modify in place
    original = pd.DataFrame({"First Name": [1]})
    clean_column_names(original)
    assert "First Name" in original.columns  # Original unchanged

Fixtures — Reusable Test Setup

A fixture provides a piece of test data or a resource that multiple tests need. Instead of creating the same DataFrame in every test, you create it once as a fixture and inject it by name.

Analogy — A restaurant kitchen prep station. Before service, the prep cook (fixture) slices onions, dices tomatoes, and portions sauces. Every chef (test function) grabs what they need from the prep station instead of prepping from scratch. This saves time and ensures consistency.

import pytest
import pandas as pd

@pytest.fixture
def sample_orders():
    # A realistic test DataFrame used by multiple tests.
    return pd.DataFrame({
        "order_id": [1, 2, 3, 4, 5],
        "product": ["Widget", "Gadget", "Widget", "Doohickey", "Gadget"],
        "amount": [29.99, 49.99, 29.99, 9.99, 49.99],
        "region": ["Ontario", "Quebec", "Ontario", "Alberta", "Ontario"],
        "order_date": pd.to_datetime(["2026-01-01", "2026-01-02", "2026-01-03",
                                      "2026-01-04", "2026-01-05"])
    })


@pytest.fixture
def sample_orders_with_nulls():
    # Test data with edge cases: nulls, duplicates, bad types.
    return pd.DataFrame({
        "order_id": [1, 2, None, 1, 5],
        "product": ["Widget", None, "Gadget", "Widget", ""],
        "amount": [29.99, "N/A", 49.99, 29.99, -5.0],
        "region": ["Ontario", "Quebec", "Ontario", None, "Alberta"]
    })


def test_filter_by_region(sample_orders):
    # Fixture injected by parameter name
    result = sample_orders[sample_orders["region"] == "Ontario"]
    assert len(result) == 3


def test_total_revenue(sample_orders):
    total = sample_orders["amount"].sum()
    assert total == pytest.approx(169.95)  # Floating-point comparison

Fixture Scope

By default, a fixture runs once per test function (scope="function"). For expensive resources like database connections, you can run the fixture once per module, class, or entire test session.

import pytest
from sqlalchemy import create_engine

@pytest.fixture(scope="session")
def db_engine():
    # Created once for the entire test session, shared by all tests.
    engine = create_engine("sqlite:///:memory:")
    yield engine
    engine.dispose()


@pytest.fixture(scope="function")
def clean_orders_table(db_engine):
    # Runs before each test: creates a fresh table.
    with db_engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS orders"))
        conn.execute(text(
            "CREATE TABLE orders (order_id INTEGER, product TEXT, amount REAL)"
        ))
    yield db_engine
    # Cleanup after each test (optional)

Scope	Fixture Runs	Best For
`function` (default)	Once per test function	Test DataFrames, small setup
`class`	Once per test class	Shared class state
`module`	Once per test file	File-level resources
`session`	Once for all tests	Database connections, heavy setup

Yield Fixtures (Setup + Teardown)

A yield fixture provides a resource, then cleans up after the test. Code before yield is setup. Code after yield is teardown — it runs even if the test fails.

import pytest
import sqlite3

@pytest.fixture
def db_connection():
    # Setup: create connection and table
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE orders (id INTEGER, amount REAL)")
    conn.execute("INSERT INTO orders VALUES (1, 29.99)")
    conn.execute("INSERT INTO orders VALUES (2, 49.99)")
    conn.commit()

    yield conn  # Test runs here with the connection

    # Teardown: always runs, even if test fails
    conn.close()


def test_query_orders(db_connection):
    cursor = db_connection.cursor()
    cursor.execute("SELECT SUM(amount) FROM orders")
    total = cursor.fetchone()[0]
    assert total == pytest.approx(79.98)

conftest.py — Shared Fixtures

Fixtures defined in conftest.py are automatically available to all test files in the same directory and subdirectories — no import needed. This is where you put fixtures that multiple test files share.

# tests/conftest.py
import pytest
import pandas as pd
from sqlalchemy import create_engine

@pytest.fixture(scope="session")
def db_engine():
    engine = create_engine("sqlite:///:memory:")
    yield engine
    engine.dispose()

@pytest.fixture
def sample_orders():
    return pd.DataFrame({
        "order_id": [1, 2, 3],
        "product": ["Widget", "Gadget", "Doohickey"],
        "amount": [29.99, 49.99, 9.99]
    })

# Now test_transforms.py, test_loaders.py, and test_pipeline.py
# can all use sample_orders and db_engine without importing anything

Parametrize — Test Multiple Inputs

@pytest.mark.parametrize runs the same test with different inputs. Instead of writing five nearly identical tests for five edge cases, you write one test and parametrize it.

Analogy — A crash test lab. They do not build a new lab for every car model. They put different cars (parameters) into the same crash test rig (test function) and record the results.

import pytest
import pandas as pd
from etl.transforms import cast_types

@pytest.mark.parametrize("input_value, expected", [
    ("100", 100.0),          # Valid string number
    ("29.99", 29.99),        # Valid decimal
    ("0", 0.0),              # Zero
    ("-15.5", -15.5),        # Negative
])
def test_cast_types_valid_values(input_value, expected):
    df = pd.DataFrame({"amount": [input_value]})
    result = cast_types(df, {"amount": float})
    assert result["amount"].iloc[0] == pytest.approx(expected)


@pytest.mark.parametrize("input_value", [
    "N/A", "null", "", "abc", "--", "None"
])
def test_cast_types_invalid_values_become_nan(input_value):
    df = pd.DataFrame({"amount": [input_value]})
    result = cast_types(df, {"amount": float})
    assert pd.isna(result["amount"].iloc[0])


@pytest.mark.parametrize("input_cols, expected_cols", [
    (["First Name"], ["first_name"]),
    (["ORDER ID"], ["order_id"]),
    (["  spaces  "], ["spaces"]),
    (["MiXeD-CaSe!"], ["mixed_case"]),
    (["already_clean"], ["already_clean"]),
])
def test_clean_column_names_various_inputs(input_cols, expected_cols):
    df = pd.DataFrame({col: [1] for col in input_cols})
    result = clean_column_names(df)
    assert list(result.columns) == expected_cols

Markers — Categorize Tests

Markers let you tag tests so you can run subsets. Common markers for data engineering: unit (fast, no dependencies), integration (needs a database), slow (takes more than a few seconds).

import pytest

@pytest.mark.unit
def test_clean_column_names():
    # Fast, no external dependencies
    ...

@pytest.mark.integration
def test_load_to_database(db_engine):
    # Needs a database connection
    ...

@pytest.mark.slow
def test_full_pipeline_end_to_end():
    # Takes 30+ seconds
    ...

# Run only unit tests
pytest -m unit

# Run everything except slow tests
pytest -m "not slow"

# Run integration and unit tests
pytest -m "unit or integration"

[tool.pytest.ini_options]
markers = [
    "unit: fast tests with no external dependencies",
    "integration: tests that require database or API access",
    "slow: tests that take more than 10 seconds",
]

Mocking — Isolate External Dependencies

When testing a function that calls an API or a database, you do not want the test to actually hit the API. Mocking replaces the real dependency with a fake that returns controlled data.

Analogy — A flight simulator. Pilots train in simulators (mocks) that look and feel like real cockpits but do not actually fly anywhere. Your tests use mocked APIs and databases that behave like real ones but return predictable, controlled responses.

unittest.mock.patch

from unittest.mock import patch, MagicMock
import pandas as pd
from etl.extractors import extract_from_api

@patch("etl.extractors.requests.get")
def test_extract_from_api(mock_get):
    # Arrange: configure the mock to return fake API data
    mock_response = MagicMock()
    mock_response.json.return_value = {
        "results": [
            {"id": 1, "name": "Widget", "price": 29.99},
            {"id": 2, "name": "Gadget", "price": 49.99}
        ]
    }
    mock_response.raise_for_status = MagicMock()  # No exception
    mock_get.return_value = mock_response

    # Act
    result = extract_from_api("https://api.example.com/products")

    # Assert
    assert len(result) == 2
    assert "name" in result.columns
    mock_get.assert_called_once()  # Verify the API was called


@patch("etl.extractors.requests.get")
def test_extract_from_api_handles_error(mock_get):
    # Simulate an API error
    mock_get.side_effect = requests.exceptions.ConnectionError("API down")

    with pytest.raises(requests.exceptions.ConnectionError):
        extract_from_api("https://api.example.com/products")

monkeypatch — pytest’s Built-in Mocking

monkeypatch is a pytest fixture that temporarily replaces attributes, environment variables, or dictionary items during a test. It automatically undoes changes after the test.

import os

def test_db_url_from_env(monkeypatch):
    # Temporarily set an environment variable
    monkeypatch.setenv("DATABASE_URL", "postgresql://test:test@localhost/testdb")

    from etl.config import get_db_url
    assert get_db_url() == "postgresql://test:test@localhost/testdb"
    # After the test, DATABASE_URL is restored to its original value


def test_missing_db_url_raises(monkeypatch):
    monkeypatch.delenv("DATABASE_URL", raising=False)

    from etl.config import get_db_url
    with pytest.raises(ValueError, match="No connection URL"):
        get_db_url()

Testing Database Code

Use an in-memory SQLite database for testing database operations. It is fast (no disk I/O), isolated (each test gets a fresh database), and requires no external services.

import pytest
import pandas as pd
import sqlite3
from sqlalchemy import create_engine, text
from etl.loaders import load_to_database

@pytest.fixture
def test_engine():
    # In-memory SQLite for testing -- fast and isolated
    engine = create_engine("sqlite:///:memory:")
    yield engine
    engine.dispose()


def test_load_creates_table(test_engine):
    df = pd.DataFrame({"order_id": [1, 2], "amount": [29.99, 49.99]})
    load_to_database(df, test_engine, "orders", if_exists="replace")

    # Verify the data was loaded
    result = pd.read_sql("SELECT * FROM orders", test_engine)
    assert len(result) == 2
    assert result["amount"].sum() == pytest.approx(79.98)


def test_load_append_adds_rows(test_engine):
    df1 = pd.DataFrame({"order_id": [1], "amount": [29.99]})
    df2 = pd.DataFrame({"order_id": [2], "amount": [49.99]})

    load_to_database(df1, test_engine, "orders", if_exists="replace")
    load_to_database(df2, test_engine, "orders", if_exists="append")

    result = pd.read_sql("SELECT * FROM orders", test_engine)
    assert len(result) == 2


def test_idempotent_load(test_engine):
    # Running the same load twice should not create duplicates
    df = pd.DataFrame({"order_id": [1, 2], "amount": [29.99, 49.99]})

    load_to_database(df, test_engine, "orders", if_exists="replace")
    load_to_database(df, test_engine, "orders", if_exists="replace")  # Second run

    result = pd.read_sql("SELECT * FROM orders", test_engine)
    assert len(result) == 2  # Same count, not 4

Testing File Operations with tmp_path

pytest provides a tmp_path fixture that gives you a fresh temporary directory for each test. Files created there are automatically cleaned up.

import pandas as pd

def test_csv_round_trip(tmp_path):
    # Write a CSV, read it back, verify contents
    df = pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]})
    csv_path = tmp_path / "test.csv"

    df.to_csv(csv_path, index=False)
    result = pd.read_csv(csv_path)

    assert len(result) == 2
    assert list(result.columns) == ["id", "name"]


def test_parquet_round_trip(tmp_path):
    df = pd.DataFrame({"id": [1, 2], "amount": [29.99, 49.99]})
    parquet_path = tmp_path / "test.parquet"

    df.to_parquet(parquet_path, index=False)
    result = pd.read_parquet(parquet_path)

    assert len(result) == 2
    assert result["amount"].sum() == pytest.approx(79.98)


def test_dead_letter_file_created(tmp_path):
    # Test that bad rows are written to a dead letter file
    from etl.transforms import transform_with_dead_letter

    df = pd.DataFrame({
        "order_id": [1, None, 3],
        "amount": [29.99, 49.99, -5.0]
    })

    valid = transform_with_dead_letter(df, output_dir=str(tmp_path))

    # Check valid rows
    assert len(valid) == 1  # Only order_id=1, amount=29.99 is valid

    # Check dead letter file was created
    dead_files = list(tmp_path.glob("rejected_*.csv"))
    assert len(dead_files) == 1
    rejected = pd.read_csv(dead_files[0])
    assert len(rejected) == 2  # Two bad rows quarantined

Testing a Complete Pipeline

Integration tests verify that all phases work together. Use fixtures for setup and tmp_path for file I/O.

import pytest
import pandas as pd
from sqlalchemy import create_engine
from etl.pipeline import run_orders_pipeline

@pytest.fixture
def pipeline_setup(tmp_path):
    # Create a test CSV and an in-memory database
    csv_path = tmp_path / "orders.csv"
    test_data = pd.DataFrame({
        "order_id": [1, 2, 3, 2],  # Note: order_id 2 is duplicated
        "product": ["Widget", "Gadget", "Doohickey", "Gadget"],
        "amount": ["29.99", "49.99", "N/A", "49.99"],
        "zip_code": ["07102", "M5V", "K1A", "M5V"]
    })
    test_data.to_csv(csv_path, index=False)

    engine = create_engine("sqlite:///:memory:")
    return str(csv_path), engine


@pytest.mark.integration
def test_pipeline_end_to_end(pipeline_setup):
    csv_path, engine = pipeline_setup
    result = run_orders_pipeline(csv_path, engine)

    # Verify metrics
    assert result["rows"] == 3  # 4 raw rows minus 1 duplicate
    assert result["duration_seconds"] > 0

    # Verify loaded data
    loaded = pd.read_sql("SELECT * FROM orders_clean", engine)
    assert len(loaded) == 3
    assert loaded["order_id"].is_unique

Coverage Reporting

Coverage shows which lines of your code are executed during tests. It helps you find untested paths — the branches and edge cases that have no tests.

# Install
pip install pytest-cov

# Run tests with coverage
pytest --cov=etl --cov-report=term-missing

# Output:
# Name                    Stmts   Miss  Cover   Missing
# etl/transforms.py          45      3    93%   67-69
# etl/loaders.py             32      8    75%   41-48
# etl/pipeline.py            28      0   100%
# TOTAL                     105     11    90%

# Generate HTML report (open htmlcov/index.html in browser)
pytest --cov=etl --cov-report=html

# Fail if coverage drops below 80%
pytest --cov=etl --cov-fail-under=80

Project Structure

A well-organized project separates source code from tests and uses conftest.py for shared fixtures.

my_project/
  src/
    etl/
      __init__.py
      extractors.py
      transforms.py
      loaders.py
      pipeline.py
      config.py
  tests/
    conftest.py              # Shared fixtures (db_engine, sample_orders)
    test_extractors.py       # Tests for extraction functions
    test_transforms.py       # Tests for transform functions
    test_loaders.py          # Tests for load functions
    test_pipeline.py         # Integration tests for full pipeline
  pyproject.toml             # pytest config, markers, coverage settings

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-v --strict-markers"
markers = [
    "unit: fast tests with no dependencies",
    "integration: tests requiring database or API",
    "slow: tests taking more than 10 seconds",
]

[tool.coverage.run]
source = ["src/etl"]
omit = ["tests/*"]

Common Mistakes

1. Not testing transform functions at all. Transform functions contain your business logic — type casting, cleaning, filtering. If cast_types() silently drops valid rows or deduplicate() keeps the wrong record, bad data flows downstream. Test every transform with at least one happy path and one edge case.

2. Testing against a real database in unit tests. Unit tests should be fast and isolated. Use sqlite:///:memory: for database tests. Reserve real database connections for integration tests that run separately (marked with @pytest.mark.integration).

3. Hardcoding test data inside test functions instead of using fixtures. When five test functions create the same DataFrame, a schema change requires updating all five. Use a fixture in conftest.py and update it in one place.

4. Not testing edge cases. Real data contains nulls, empty strings, negative numbers, unicode characters, and values like “N/A” in numeric columns. Use @pytest.mark.parametrize to test every edge case with one test function.

5. Mocking too much or too little. Mock external dependencies (APIs, databases, file systems) but do not mock the function you are testing. If you mock pandas.DataFrame, you are not testing anything real. Mock the boundary (the API call), not the logic.

6. Ignoring test isolation. Tests that share state (a database table, a global variable) can pass individually but fail when run together. Each test should set up and tear down its own state. Use function-scoped fixtures and tmp_path for isolation.

7. Not running tests in CI/CD. Tests that run only on a developer’s laptop are tests that eventually stop running. Add pytest --cov --cov-fail-under=80 to your GitHub Actions or Azure DevOps pipeline so every commit is validated.

8. Writing tests after the pipeline is already in production. Retrofitting tests is hard because the code was not designed for testability (hardcoded paths, mixed concerns, global state). Write tests alongside your code. If you cannot test a function easily, the function needs refactoring.

Interview Questions

Q: Why should data engineers write tests for their pipelines? A: Data pipelines handle business-critical information. A bug in a transform function can silently corrupt millions of rows, leading to wrong reports and bad business decisions. Tests catch these bugs before they reach production. They also enable confident refactoring — when you need to optimize a pipeline, tests verify that the output remains identical. Without tests, every change is a gamble.

Q: What is a pytest fixture and how does it help with testing data pipelines? A: A fixture is a function decorated with @pytest.fixture that provides reusable test data or resources. For data pipelines, fixtures typically provide sample DataFrames, database connections, or temporary directories. They eliminate duplicate setup code across tests and ensure consistency. Fixtures can be shared via conftest.py and scoped to control their lifecycle — scope="session" for expensive database connections, scope="function" for test DataFrames that need fresh state.

Q: How do you test a function that calls an external API? A: Use mocking to replace the HTTP call with a controlled fake response. In pytest, use @patch("module.requests.get") from unittest.mock and configure the mock to return a MagicMock with a .json() method that returns your test data. This makes the test fast (no network call), deterministic (same response every time), and isolated (not affected by API downtime). Test both the happy path (valid response) and error paths (connection error, rate limit, empty response).

Q: What is the difference between unit tests and integration tests for a data pipeline? A: Unit tests verify individual functions in isolation — a single transform function with a small test DataFrame. They are fast (milliseconds), have no external dependencies, and catch logic bugs. Integration tests verify that multiple components work together — extracting from a file, transforming, and loading into a database. They are slower, may need a test database, and catch interface bugs (wrong column names, type mismatches). Run unit tests on every commit and integration tests before deployment.

Q: What is @pytest.mark.parametrize and when would you use it? A: Parametrize runs the same test function with different inputs, eliminating duplicated test code. For data pipelines, use it to test edge cases: valid values (“100”), invalid values (“N/A”, “”, “null”), boundary values (0, negative numbers), and special characters (unicode, quotes). One parametrized test with 10 inputs replaces 10 nearly identical test functions.

Q: How do you test database write operations without a production database? A: Use an in-memory SQLite database (sqlite:///:memory:) for testing. It is instant to create, requires no server, and each test gets a fresh instance. Create the database in a fixture, load test data, run the function, query the result, and verify. For SQL Server or PostgreSQL-specific syntax, use a Docker container with the real database in integration tests.

Q: What does test coverage measure and what is a good target? A: Test coverage measures the percentage of source code lines executed during tests. It shows which parts of your code have no tests at all. A common target is 80% for data pipelines. 100% is not practical or necessary — some error handling paths are hard to trigger in tests. Coverage is most valuable for finding completely untested functions, not for chasing a number. Use pytest-cov with --cov-fail-under=80 to enforce a minimum in CI/CD.

Wrapping Up

Testing is what separates hobby scripts from production pipelines. Start with the transforms — they contain your business logic and are the easiest to test. Add fixtures for reusable test data, parametrize for edge cases, and mocks for external dependencies. Use tmp_path for file operations and in-memory SQLite for database tests. Run everything in CI/CD with coverage reporting. The first test is the hardest — after that, it becomes second nature.

← Previous: ETL Patterns Python (25/31) Next: Concurrency →

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.

Python Testing with pytest for Data Engineers: Fixtures, Parametrize, Mocking, conftest.py, Testing ETL Pipelines, Database Code, API Extractors, Coverage, and Production Patterns

Table of Contents

Why Data Engineers Need Tests

pytest Basics

Installation and First Test

Testing Real Transform Functions

Fixtures — Reusable Test Setup

Fixture Scope

Yield Fixtures (Setup + Teardown)

conftest.py — Shared Fixtures

Parametrize — Test Multiple Inputs

Markers — Categorize Tests

Mocking — Isolate External Dependencies

unittest.mock.patch

monkeypatch — pytest’s Built-in Mocking

Testing Database Code

Testing File Operations with tmp_path

Testing a Complete Pipeline

Coverage Reporting

Project Structure

Common Mistakes

Interview Questions

Wrapping Up

Leave a Comment Cancel Reply

Table of Contents

Why Data Engineers Need Tests

pytest Basics

Installation and First Test

Testing Real Transform Functions

Fixtures — Reusable Test Setup

Fixture Scope

Yield Fixtures (Setup + Teardown)

conftest.py — Shared Fixtures

Parametrize — Test Multiple Inputs

Markers — Categorize Tests

Mocking — Isolate External Dependencies

unittest.mock.patch

monkeypatch — pytest’s Built-in Mocking

Testing Database Code

Testing File Operations with tmp_path

Testing a Complete Pipeline

Coverage Reporting

Project Structure

Common Mistakes

Interview Questions

Wrapping Up

Related Posts

Leave a Comment Cancel Reply