AWS Glue ETL Testing: Local Development and CI/CD Integration

# Pull Glue 4.0 image (Python 3.10, Spark 3.3)
docker pull amazon/aws-glue-libs:glue_libs_4.0.0_image_01

# Start interactive development environment
docker run -it \
  --name glue-dev \
  -v $(pwd):/home/glue_user/workspace \
  -e AWS_DEFAULT_REGION=us-east-1 \
  -e AWS_ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID \
  -e AWS_SECRET_ACCESS_KEY=$AWS_SECRET_ACCESS_KEY \
  -p 4040:4040 \  # Spark UI
  amazon/aws-glue-libs:glue_libs_4.0.0_image_01 \
  /home/glue_user/jupyter/jupyter_start.sh  # For Jupyter

# Or run pytest directly
docker run --rm \
  -v $(pwd):/home/glue_user/workspace \
  -e AWS_DEFAULT_REGION=us-east-1 \
  -e AWS_ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID \
  -e AWS_SECRET_ACCESS_KEY=$AWS_SECRET_ACCESS_KEY \
  amazon/aws-glue-libs:glue_libs_4.0.0_image_01 \
  python -m pytest /home/glue_user/workspace/tests/

# transformations.py — pure transformation logic, testable without Glue
import pandas as pd
from typing import Optional

def clean_orders(df: pd.DataFrame) -> pd.DataFrame:
    """Clean and validate order data."""
    df = df.copy()
    
    # Remove nulls in required fields
    df = df.dropna(subset=['order_id', 'customer_id', 'amount'])
    
    # Filter invalid amounts
    df = df[df['amount'] > 0]
    
    # Normalize status
    df['status'] = df['status'].str.lower().str.strip()
    
    # Parse dates
    df['order_date'] = pd.to_datetime(df['order_date'])
    df['year'] = df['order_date'].dt.year
    df['month'] = df['order_date'].dt.month
    
    return df

def aggregate_by_customer(df: pd.DataFrame) -> pd.DataFrame:
    """Aggregate orders by customer."""
    return df.groupby('customer_id').agg(
        total_orders=('order_id', 'count'),
        total_revenue=('amount', 'sum'),
        first_order_date=('order_date', 'min'),
        last_order_date=('order_date', 'max'),
    ).reset_index()

# glue_job.py — the actual Glue script
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
import pyspark.pandas as ps

# Import pure transformation logic
from transformations import clean_orders, aggregate_by_customer

args = getResolvedOptions(sys.argv, ['JOB_NAME', 'source_path', 'target_path'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read
source_df = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": [args['source_path']], "recurse": True},
    format="parquet"
)

# Convert to pandas-on-Spark for transformations
pandas_df = source_df.toDF().toPandas()

# Apply transformations (pure functions, tested separately)
cleaned = clean_orders(pandas_df)
aggregated = aggregate_by_customer(cleaned)

# Write
output_df = spark.createDataFrame(aggregated)
output_dynamic_frame = DynamicFrame.fromDF(output_df, glueContext, "output")

glueContext.write_dynamic_frame.from_options(
    frame=output_dynamic_frame,
    connection_type="s3",
    connection_options={"path": args['target_path']},
    format="parquet"
)

job.commit()

# tests/test_transformations.py
import pytest
import pandas as pd
from datetime import datetime
from transformations import clean_orders, aggregate_by_customer

class TestCleanOrders:
    @pytest.fixture
    def sample_orders(self):
        return pd.DataFrame({
            'order_id': ['o1', 'o2', 'o3', 'o4'],
            'customer_id': ['c1', 'c1', 'c2', None],
            'amount': [100.0, -50.0, 200.0, 150.0],
            'status': ['PENDING', ' completed ', 'Cancelled', 'pending'],
            'order_date': ['2024-01-15', '2024-01-16', '2024-01-17', '2024-01-18'],
        })
    
    def test_removes_null_customer_ids(self, sample_orders):
        result = clean_orders(sample_orders)
        assert result['customer_id'].isna().sum() == 0
    
    def test_removes_negative_amounts(self, sample_orders):
        result = clean_orders(sample_orders)
        assert (result['amount'] <= 0).sum() == 0
    
    def test_normalizes_status(self, sample_orders):
        result = clean_orders(sample_orders)
        assert result['status'].isin(['pending', 'completed', 'cancelled']).all()
    
    def test_parses_order_date(self, sample_orders):
        result = clean_orders(sample_orders)
        assert result['order_date'].dtype == 'datetime64[ns]'
        assert 'year' in result.columns
        assert 'month' in result.columns
    
    def test_preserves_valid_records(self, sample_orders):
        result = clean_orders(sample_orders)
        # o1 (pending), o3 (cancelled) should be kept; o2 (negative), o4 (null) dropped
        assert len(result) == 2
        assert set(result['order_id']) == {'o1', 'o3'}

class TestAggregateByCustomer:
    @pytest.fixture
    def cleaned_orders(self):
        return pd.DataFrame({
            'order_id': ['o1', 'o2', 'o3', 'o4'],
            'customer_id': ['c1', 'c1', 'c2', 'c1'],
            'amount': [100.0, 200.0, 150.0, 50.0],
            'order_date': pd.to_datetime(['2024-01-01', '2024-01-15', '2024-01-10', '2024-01-20']),
        })
    
    def test_aggregates_per_customer(self, cleaned_orders):
        result = aggregate_by_customer(cleaned_orders)
        
        c1 = result[result['customer_id'] == 'c1'].iloc[0]
        assert c1['total_orders'] == 3
        assert c1['total_revenue'] == pytest.approx(350.0)
        
        c2 = result[result['customer_id'] == 'c2'].iloc[0]
        assert c2['total_orders'] == 1
    
    def test_first_and_last_order_dates(self, cleaned_orders):
        result = aggregate_by_customer(cleaned_orders)
        
        c1 = result[result['customer_id'] == 'c1'].iloc[0]
        assert c1['first_order_date'] == pd.Timestamp('2024-01-01')
        assert c1['last_order_date'] == pd.Timestamp('2024-01-20')

# tests/test_glue_transforms.py
import pytest

# Only run in Glue environment
try:
    from pyspark.context import SparkContext
    from awsglue.context import GlueContext
    HAS_GLUE = True
except ImportError:
    HAS_GLUE = False

pytestmark = pytest.mark.skipif(not HAS_GLUE, reason="Requires Glue runtime")

@pytest.fixture(scope="session")
def glue_context():
    sc = SparkContext.getOrCreate()
    glueContext = GlueContext(sc)
    yield glueContext
    sc.stop()

def test_dynamic_frame_schema(glue_context, tmp_path):
    """Test that Glue reads data with correct schema."""
    spark = glue_context.spark_session
    
    # Create test parquet file
    test_data = spark.createDataFrame([
        ('o1', 'c1', 100.0, 'pending', '2024-01-15'),
        ('o2', 'c2', 200.0, 'completed', '2024-01-16'),
    ], ['order_id', 'customer_id', 'amount', 'status', 'order_date'])
    
    test_path = str(tmp_path / 'test-data')
    test_data.write.parquet(test_path)
    
    # Read with Glue
    dynamic_frame = glue_context.create_dynamic_frame.from_options(
        connection_type="s3",
        connection_options={"paths": [test_path]},
        format="parquet",
        transformation_ctx="test_input",
    )
    
    schema = dynamic_frame.schema()
    field_names = [f.name for f in schema]
    
    assert 'order_id' in field_names
    assert 'amount' in field_names
    
    count = dynamic_frame.count()
    assert count == 2

# docker-compose.yml for tests
version: '3'
services:
  localstack:
    image: localstack/localstack
    ports:
      - "4566:4566"
    environment:
      - SERVICES=s3,glue
      - DEFAULT_REGION=us-east-1
      - AWS_DEFAULT_REGION=us-east-1

# tests/test_s3_integration.py
import boto3
import pytest
import pandas as pd
import io

LOCALSTACK_ENDPOINT = "http://localhost:4566"

@pytest.fixture(scope="session")
def s3_client():
    return boto3.client(
        's3',
        endpoint_url=LOCALSTACK_ENDPOINT,
        aws_access_key_id='test',
        aws_secret_access_key='test',
        region_name='us-east-1',
    )

@pytest.fixture
def test_bucket(s3_client):
    bucket_name = f'test-bucket-{id(s3_client)}'
    s3_client.create_bucket(Bucket=bucket_name)
    yield bucket_name
    # Cleanup
    objects = s3_client.list_objects_v2(Bucket=bucket_name).get('Contents', [])
    for obj in objects:
        s3_client.delete_object(Bucket=bucket_name, Key=obj['Key'])
    s3_client.delete_bucket(Bucket=bucket_name)

def test_etl_reads_from_s3(s3_client, test_bucket):
    # Upload test data
    test_df = pd.DataFrame([
        {'order_id': 'o1', 'amount': 100.0, 'status': 'pending'},
    ])
    
    buffer = io.BytesIO()
    test_df.to_parquet(buffer, index=False)
    
    s3_client.put_object(
        Bucket=test_bucket,
        Key='input/orders.parquet',
        Body=buffer.getvalue(),
    )
    
    # Run ETL logic
    from transformations import clean_orders
    
    # Read from S3
    obj = s3_client.get_object(Bucket=test_bucket, Key='input/orders.parquet')
    df = pd.read_parquet(io.BytesIO(obj['Body'].read()))
    
    result = clean_orders(df)
    
    # Write output
    output_buffer = io.BytesIO()
    result.to_parquet(output_buffer, index=False)
    s3_client.put_object(
        Bucket=test_bucket,
        Key='output/cleaned.parquet',
        Body=output_buffer.getvalue(),
    )
    
    # Verify output in S3
    output_obj = s3_client.get_object(Bucket=test_bucket, Key='output/cleaned.parquet')
    output_df = pd.read_parquet(io.BytesIO(output_obj['Body'].read()))
    
    assert len(output_df) == 1
    assert output_df.iloc[0]['status'] == 'pending'

# .github/workflows/glue-tests.yml
name: Glue ETL Tests

on: [push, pull_request]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.10'
      - run: pip install pandas pytest pytest-cov
      - name: Run unit tests (no Glue runtime needed)
        run: pytest tests/test_transformations.py -v --cov=transformations
  
  glue-integration-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Start LocalStack
        run: |
          pip install localstack awscli-local
          localstack start -d
          sleep 10
      
      - name: Run integration tests
        run: |
          docker run --rm \
            --network host \
            -v $(pwd):/home/glue_user/workspace \
            -e AWS_DEFAULT_REGION=us-east-1 \
            -e AWS_ACCESS_KEY_ID=test \
            -e AWS_SECRET_ACCESS_KEY=test \
            -e AWS_ENDPOINT_URL=http://localhost:4566 \
            amazon/aws-glue-libs:glue_libs_4.0.0_image_01 \
            python -m pytest /home/glue_user/workspace/tests/test_s3_integration.py -v