# Data Pipeline & Quality Testing Cheatsheet

A comprehensive guide to testing data pipelines across unit, integration, schema validation, data warehouse, and end-to-end observability layers.

---

## 1. Unit Testing (PyTest + Mock DataFrames)
Test logical data cleaning, mapping, and transformations in isolation without connecting to databases.

```python
# test_transformations.py
import pytest
import pandas as pd

def clean_emails(df):
    df['email'] = df['email'].str.strip().str.lower()
    return df

def test_clean_emails():
    # Setup mock input
    mock_input = pd.DataFrame({'email': [' JOHN@example.com ', 'JANE@gmail.com']})
    
    # Process
    result = clean_emails(mock_input)
    
    # Assertions
    expected = pd.DataFrame({'email': ['john@example.com', 'jane@gmail.com']})
    pd.testing.assert_frame_equal(result, expected)
```

---

## 2. Schema Validation (Pandera SchemaModel)
Enforce row-level constraints, column datatypes, null bounds, and range conditions on pipeline ingestion boundaries.

```python
# schema_validation.py
import pandas as pd
import pandera as pa
from pandera.typing import Series

class CustomerSchema(pa.SchemaModel):
    id: Series[int] = pa.Field(unique=True)
    age: Series[int] = pa.Field(ge=1, le=120)
    email: Series[str] = pa.Field(str_matches=r"^.+@.+\..+$")
    status: Series[str] = pa.Field(isin=["ACTIVE", "INACTIVE"])

# Validate incoming DataFrame at runtime
try:
    CustomerSchema.validate(df, lazy=True)
except pa.errors.SchemaErrors as err:
    print("Schema contract broken:", err.failure_cases)
```

---

## 3. Integration Testing (Docker Testcontainers)
Test network connections and read/write capabilities by spawning a real database container on the fly during local testing.

```python
# test_integration.py
from testcontainers.postgres import PostgresContainer
import psycopg2

def test_db_write():
    # Spin up postgres container
    with PostgresContainer("postgres:15-alpine") as postgres:
        conn = psycopg2.connect(
            host=postgres.get_container_host_ip(),
            port=postgres.get_exposed_port(5432),
            user=postgres.username,
            password=postgres.password,
            database=postgres.dbname
        )
        
        # Test table creation and query execution
        with conn.cursor() as cursor:
            cursor.execute("CREATE TABLE users (id SERIAL PRIMARY KEY, name VARCHAR);")
            cursor.execute("INSERT INTO users (name) VALUES ('Somsak');")
            conn.commit()
            
            cursor.execute("SELECT name FROM users;")
            result = cursor.fetchone()
            assert result[0] == 'Somsak'
```

---

## 4. Analytical Warehouse Validation (dbt Core Tests)
Assert unique keys, referential relationships, null presence, and value sets directly inside your data warehouse tables.

```yaml
# models/schema.yml
version: 2

models:
  - name: fct_sales
    columns:
      - name: sales_key
        tests:
          - unique
          - not_null
      - name: customer_key
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_key
      - name: status
        tests:
          - accepted_values:
              values: ['completed', 'returned', 'pending']
```

---

## 5. End-to-End Quality Observability (Great Expectations)
Track data quality drifts, column statistics variations, and output profiling reports before delivering datasets to BI layers.

```python
# great_expectations_run.py
import great_expectations as ge

# Wrap data into expectation suite context
ge_df = ge.from_pandas(df)

# Assert data distribution characteristics
ge_df.expect_column_values_to_not_be_null("user_id")
ge_df.expect_column_min_to_be_between("age", min_value=1)
ge_df.expect_column_mean_to_be_between("amount", min_value=10.0, max_value=500.0)

# Retrieve validation results
results = ge_df.validate()
if not results["success"]:
    raise ValueError("E2E Quality check failed! Blocking release.")
```