Tutorial: CI/CD Pipeline Testing with LocalEmu

import boto3
import json
import os


def get_client(service):
    """Create a boto3 client. Uses AWS_ENDPOINT_URL for LocalEmu when set."""
    kwargs = {}
    endpoint = os.environ.get("AWS_ENDPOINT_URL")
    if endpoint:
        kwargs["endpoint_url"] = endpoint
    return boto3.client(service, region_name="us-east-1", **kwargs)


def process_file(bucket, key):
    """Read a JSON file from S3, extract records, write each to DynamoDB."""
    s3 = get_client("s3")
    dynamodb = get_client("dynamodb")

    # Download the file from S3
    response = s3.get_object(Bucket=bucket, Key=key)
    data = json.loads(response["Body"].read())

    # Write each record to DynamoDB
    written = 0
    for record in data.get("records", []):
        dynamodb.put_item(
            TableName="processed-records",
            Item={
                "id": {"S": record["id"]},
                "name": {"S": record["name"]},
                "status": {"S": "processed"},
                "source_file": {"S": key},
            },
        )
        written += 1

    return {"processed": written, "source": f"s3://{bucket}/{key}"}

boto3
pytest

import boto3
import pytest
import os

ENDPOINT = os.environ.get("AWS_ENDPOINT_URL", "http://localhost:4566")
CLIENT_KWARGS = dict(
    endpoint_url=ENDPOINT,
    aws_access_key_id="AKIAIOSFODNN7EXAMPLE",
    aws_secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    region_name="us-east-1",
)


@pytest.fixture(scope="session")
def s3_client():
    """Create a reusable S3 client for the entire test session."""
    return boto3.client("s3", **CLIENT_KWARGS)


@pytest.fixture(scope="session")
def dynamodb_client():
    """Create a reusable DynamoDB client for the entire test session."""
    return boto3.client("dynamodb", **CLIENT_KWARGS)

import json
import sys
import os
import pytest

# Add the src directory to the path so we can import app
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))

from app import process_file


@pytest.fixture(autouse=True)
def setup_resources(s3_client, dynamodb_client):
    """Create S3 bucket and DynamoDB table before each test, clean up after."""
    # Create bucket (ignore if it already exists)
    try:
        s3_client.create_bucket(Bucket="uploads")
    except s3_client.exceptions.BucketAlreadyOwnedByYou:
        pass

    # Create table (ignore if it already exists)
    try:
        dynamodb_client.create_table(
            TableName="processed-records",
            KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
            AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
            BillingMode="PAY_PER_REQUEST",
        )
    except dynamodb_client.exceptions.ResourceInUseException:
        pass

    yield

    # Clean up: delete all items and resources
    try:
        scan = dynamodb_client.scan(TableName="processed-records")
        for item in scan.get("Items", []):
            dynamodb_client.delete_item(
                TableName="processed-records",
                Key={"id": item["id"]},
            )
    except Exception:
        pass


def test_process_file_writes_all_records(s3_client, dynamodb_client):
    """Upload a JSON file with 3 records, verify all 3 land in DynamoDB."""
    test_data = json.dumps({"records": [
        {"id": "r1", "name": "Alice"},
        {"id": "r2", "name": "Bob"},
        {"id": "r3", "name": "Charlie"},
    ]})
    s3_client.put_object(Bucket="uploads", Key="batch-001.json", Body=test_data)

    result = process_file("uploads", "batch-001.json")
    assert result["processed"] == 3

    # Verify each record
    for rid, name in [("r1", "Alice"), ("r2", "Bob"), ("r3", "Charlie")]:
        item = dynamodb_client.get_item(
            TableName="processed-records",
            Key={"id": {"S": rid}},
        )["Item"]
        assert item["name"]["S"] == name
        assert item["status"]["S"] == "processed"
        assert item["source_file"]["S"] == "batch-001.json"


def test_process_file_handles_empty_records(s3_client):
    """An empty records array should process zero items without errors."""
    s3_client.put_object(
        Bucket="uploads",
        Key="empty.json",
        Body=json.dumps({"records": []}),
    )
    result = process_file("uploads", "empty.json")
    assert result["processed"] == 0


def test_process_file_sets_source_file(s3_client, dynamodb_client):
    """The source_file field should match the S3 key of the input file."""
    test_data = json.dumps({"records": [
        {"id": "src-1", "name": "Diana"},
    ]})
    s3_client.put_object(Bucket="uploads", Key="report-42.json", Body=test_data)

    process_file("uploads", "report-42.json")

    item = dynamodb_client.get_item(
        TableName="processed-records",
        Key={"id": {"S": "src-1"}},
    )["Item"]
    assert item["source_file"]["S"] == "report-42.json"

name: Integration Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      localemu:
        image: localemu/localemu:latest
        ports:
          - 4566:4566
        options: >-
          --health-cmd "curl -f http://localhost:4566/_localemu/health"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run integration tests
        run: pytest tests/ -v --tb=short
        env:
          AWS_ENDPOINT_URL: http://localhost:4566
          AWS_ACCESS_KEY_ID: AKIAIOSFODNN7EXAMPLE
          AWS_SECRET_ACCESS_KEY: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
          AWS_DEFAULT_REGION: us-east-1