---
description: Data Pipeline Testing
alwaysApply: false
---

# Data Pipeline Testing

Strategies for testing data pipelines effectively.

## Testing Pyramid

- **Unit** — single transformation, fast, test logic in isolation
- **Integration** — end-to-end pipeline flow, medium speed
- **Contract** — schema/interface compatibility, prevent breaking changes
- **Data Quality** — validate production data, slowest

## Unit Tests

Test individual transformations with small, controlled DataFrames.

```python
def test_calculate_order_total(spark):
    df = spark.createDataFrame([{"order_id": "1", "quantity": 2, "unit_price": Decimal("10.00")}])
    assert calculate_order_total(df).collect()[0]["total"] == Decimal("20.00")
```

## Integration Tests

Test full pipeline flows in isolated test databases. Use fixtures to create/drop DBs.

```python
def test_idempotency(spark, test_db):
    for _ in range(2):
        run_pipeline(source=f"{test_db}.raw", target=f"{test_db}.curated", date=date(2024,1,15))
    assert spark.table(f"{test_db}.curated").count() == 3  # Not doubled
```

## Contract Tests

Ensure schema backward compatibility for downstream consumers.

```python
def test_schema_backward_compatible():
    schema = spark.table("curated.orders").schema
    required = {"order_id": StringType(), "total_amount": DecimalType(12,2)}
    for col, expected_type in required.items():
        assert col in [f.name for f in schema.fields]
        assert schema[col].dataType == expected_type
```

## DBT Tests

```yaml
models:
  - name: orders
    columns:
      - name: order_id
        tests: [not_null, unique]
      - name: customer_id
        tests: [not_null, {relationships: {to: ref('customers'), field: customer_id}}]
```

## Best Practices

- Test behavior, not implementation (test WHAT, not HOW)
- Each test sets up its own data — no shared mutable state
- Use descriptive names: `test_calculate_total_handles_zero_quantity`
- Test edge cases: nulls, empty inputs, duplicates, large values
- Use `assert_df_equality` (chispa) for DataFrame comparisons

## Anti-Patterns

- Only testing the happy path; skipping null/empty/duplicate cases
- Tests depending on shared state or execution order
- Testing implementation details (e.g., checking query plans)