pytest-data-engineering

name	pytest-data-engineering
description	Write pytest tests for data engineering projects in Python. Trigger when the user asks to write, add, or refactor unit tests for data engineering code — PySpark transformations, pandas pipelines, DuckDB queries, Polars pipelines, ETL modules, or data quality checks. Also trigger on pytest, conftest, fixtures, _test.py, _data.py, assertDataFrameEqual, pytest.mark, coverage, or testing-strategy requests. Trigger phrases: "write a test", "add tests", "test this transformation", "how do I test this", or "set up pytest". Also trigger when the user asks to set up a virtualenv, configure ruff, set up pre-commit hooks, or configure a CI pipeline for a data project. Do NOT use for writing production PySpark code (use pyspark-expert), general Python testing outside data engineering, or tasks that do not end in a pytest test file.
allowed-tools	Read Write Edit Bash Grep
license	GPL-3.0-only
status	unstable
metadata	{"author":"Leandro Kellermann de Oliveira <[e-mail]>","version":"0.1.0","model":"sonnet"}

Pytest for Data Engineering

This skill produces production-quality pytest test suites for data engineering projects. It supports PySpark 3.5+, pandas, Polars, and DuckDB, enforcing a consistent file layout, fixture naming conventions, and configuration files regardless of the compute engine.

---

Core Principles

1. No test classes — flat def test_*() functions only. Each function tests one behavior.
2. _data.py / _test.py pairs — mock data lives in a separate file from test logic. They are connected via pytest_plugins, not direct imports.
3. Schemas are contracts — import schema constants from src/schemas.py. Never redefine them in test files.
4. Test with the real engine — never mock DataFrame or connection internals. Use a session-scoped fixture for stateful engines (Spark, DuckDB). For stateless libraries (pandas, Polars), build DataFrames directly in function-scoped fixtures.
5. Use the engine's native assertion utility — validates both schema and data in one call. See the engine quick-reference table in the Routing section below.

---

TDD Workflow (enforced)

When developing a new transformation, always follow this order:

1. Define the mock input and expected output fixtures in _data.py.
2. Write the test function in _test.py — it will fail because the transformation doesn't exist yet.
3. Implement the transformation in src/transformations.py until the test passes.
4. Add edge cases (nulls, empty DataFrames, constraint violations) as new fixture + test pairs.

Use @pytest.mark.skip(reason="TBD") to stub tests you plan to implement later, or if you failed to implement a test more than 3 times. In case of repeated failure, write a TODO.md at the project root:

# TODO

## Skipped Tests

| Test       | File                                       | Reason | Attempted approaches |
| ---------- | ------------------------------------------ | ------ | -------------------- |
| `test_foo` | `tests/unit_tests/transformations_test.py` | <why>  | <what was tried>     |

## Known Blockers

- <dependency version conflict, missing built-in, etc.>

---

Project Test Layout (enforced)

project_name/
├── src/
│   ├── schemas.py             # DDL schema constants — single source of truth
│   └── transformations.py     # Pure DataFrame → DataFrame functions
├── tests/
│   └── unit_tests/
│       ├── conftest.py                # Shared session fixture (engine-dependent)
│       ├── transformations_data.py    # mock_* and expected_* fixtures
│       └── transformations_test.py    # Flat def test_*() functions
├── pytest.toml
├── .coveragerc
└── ruff.toml

---

Routing

1. Detect the stack

grep -E "pyspark|duckdb|polars|pandas" requirements.txt pyproject.toml 2>/dev/null

2. Engine routing

Detected dependency	Engine	Session fixture?	Assertion utility	Reference file
pyspark	Spark	Yes — SparkSession (session-scoped)	assertDataFrameEqual (pyspark.testing)	references/engines/spark.md
duckdb	DuckDB	Yes — duckdb.connect() (session-scoped)	convert to pandas → tm.assert_frame_equal	references/engines/duckdb.md
polars	Polars	No	assert_frame_equal (polars.testing)	references/engines/polars.md
pandas only	pandas	No	tm.assert_frame_equal (pandas.testing)	references/engines/pandas.md

Multiple engines can coexist — apply the relevant pattern for each.

For scaffolding tasks (virtualenv / pytest config / CI / ruff / pre-commit), also read references/setup.md.

Read only the file(s) needed for the current task. Do not read files for engines not present in the project.

---

Cross-engine patterns

Constraint validation (_test.py)

The _test.py assertion body is identical for all engines. The _data.py fixture uses the engine's DataFrame constructor — see the engine reference file for that part.

import pytest


def test_check_constraints_ok(mock_valid_data):
    """Test that valid data passes constraint check."""
    assert check_constraints(mock_valid_data) is True


def test_check_constraints_raises_on_violations(mock_constraint_violations):
    """Test that constraint violations raise ValueError with row count."""
    with pytest.raises(ValueError) as exc_info:
        check_constraints(mock_constraint_violations)

    bad_rows = exc_info.value.args[1]
    assert bad_rows == 5