AWS

Testing AWS Lambda Functions: Unit, Integration, and End-to-End Patterns

HelpMeTest

16 May 2026 — 5 min read

Testing Lambda functions is different from testing traditional services. Your function has no persistent process, it runs in an ephemeral container, and it's tightly coupled to AWS events and services. A function that works locally can fail in production due to IAM permissions, environment variable differences, timeout constraints, or cold start behavior.

This guide covers a layered testing strategy that catches bugs at each level before they reach production.

The Testing Pyramid for Lambda

          ┌─────────────────┐
          │  E2E (deployed) │  ← Deployed function, real AWS events
          ├─────────────────┤
          │   Integration   │  ← SAM local, real AWS services (S3, DynamoDB)
          ├─────────────────┤
          │   Unit Tests    │  ← Pure business logic, mocked AWS SDK
          └─────────────────┘

Unit tests are fast and cheap. Integration tests verify AWS service interactions. E2E tests validate the real deployment. All three are necessary.

Unit Testing Lambda Functions

Structure Your Handler for Testability

The key is separating handler logic from business logic:

# handler.py
import json
from services import UserService

# Handler — thin wrapper, hard to unit test directly
def lambda_handler(event, context):
    user_id = event['pathParameters']['userId']
    service = UserService()
    user = service.get_user(user_id)
    
    if not user:
        return {'statusCode': 404, 'body': json.dumps({'error': 'User not found'})}
    
    return {'statusCode': 200, 'body': json.dumps(user.to_dict())}

# services.py — pure business logic, testable without AWS
class UserService:
    def __init__(self, repository=None):
        self.repository = repository or DynamoDBUserRepository()
    
    def get_user(self, user_id: str) -> Optional[User]:
        if not user_id or not user_id.strip():
            raise ValueError("user_id cannot be empty")
        return self.repository.find(user_id)

Now test UserService independently:

# test_user_service.py
import pytest
from unittest.mock import Mock
from services import UserService, User

def test_get_user_returns_user_when_found():
    mock_repo = Mock()
    mock_repo.find.return_value = User(id="123", name="Alice")
    service = UserService(repository=mock_repo)
    
    result = service.get_user("123")
    
    assert result.name == "Alice"
    mock_repo.find.assert_called_once_with("123")

def test_get_user_raises_on_empty_id():
    service = UserService(repository=Mock())
    
    with pytest.raises(ValueError, match="cannot be empty"):
        service.get_user("")

def test_get_user_returns_none_when_not_found():
    mock_repo = Mock()
    mock_repo.find.return_value = None
    service = UserService(repository=mock_repo)
    
    result = service.get_user("nonexistent")
    
    assert result is None

Testing the Handler with Mock Events

# test_handler.py
import json
from unittest.mock import Mock, patch
from handler import lambda_handler

def make_event(user_id):
    return {
        'pathParameters': {'userId': user_id},
        'httpMethod': 'GET',
        'headers': {}
    }

def test_handler_returns_200_for_existing_user():
    with patch('handler.UserService') as mock_service_class:
        mock_service = Mock()
        mock_service.get_user.return_value = Mock(
            to_dict=lambda: {'id': '123', 'name': 'Alice'}
        )
        mock_service_class.return_value = mock_service
        
        response = lambda_handler(make_event('123'), {})
        
        assert response['statusCode'] == 200
        body = json.loads(response['body'])
        assert body['name'] == 'Alice'

def test_handler_returns_404_for_missing_user():
    with patch('handler.UserService') as mock_service_class:
        mock_service = Mock()
        mock_service.get_user.return_value = None
        mock_service_class.return_value = mock_service
        
        response = lambda_handler(make_event('unknown'), {})
        
        assert response['statusCode'] == 404

Mocking AWS Services with moto

moto intercepts AWS SDK calls and simulates AWS services in-process:

pip install moto[dynamodb]

import boto3
import pytest
from moto import mock_dynamodb
from repositories import DynamoDBUserRepository

@mock_dynamodb
def test_repository_saves_and_retrieves_user():
    # Create the table (moto simulates DynamoDB)
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    dynamodb.create_table(
        TableName='users',
        KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )
    
    repo = DynamoDBUserRepository(table_name='users', region='us-east-1')
    user = User(id='123', name='Alice', email='alice@example.com')
    
    repo.save(user)
    retrieved = repo.find('123')
    
    assert retrieved.name == 'Alice'
    assert retrieved.email == 'alice@example.com'

@mock_dynamodb
def test_repository_returns_none_for_missing_user():
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    dynamodb.create_table(
        TableName='users',
        KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )
    
    repo = DynamoDBUserRepository(table_name='users', region='us-east-1')
    result = repo.find('nonexistent')
    
    assert result is None

moto supports S3, SQS, SNS, Secrets Manager, and most other AWS services.

Integration Testing with AWS SAM Local

AWS SAM Local runs your Lambda functions locally in a Docker container that mimics the Lambda runtime environment. It catches issues that unit tests can't: handler path errors, missing dependencies, runtime version mismatches.

pip install aws-sam-cli

# Invoke a function locally with a test event
sam local invoke GetUser --event events/get-user-event.json

# Start a local API Gateway
sam local start-api

# Run a specific function with environment variables
sam local invoke GetUser \
  --event events/get-user-event.json \
  --env-vars env.json

Create test events that match real AWS event shapes:

// events/get-user-event.json
{
  "httpMethod": "GET",
  "pathParameters": {
    "userId": "123"
  },
  "headers": {
    "Authorization": "Bearer test-token"
  },
  "requestContext": {
    "authorizer": {
      "claims": {
        "sub": "user-123"
      }
    }
  }
}

Automate SAM local invocations in tests:

import subprocess
import json
import pytest

def invoke_sam_local(function_name, event):
    """Invoke a Lambda locally via SAM CLI"""
    import tempfile
    import os
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
        json.dump(event, f)
        event_file = f.name
    
    try:
        result = subprocess.run(
            ['sam', 'local', 'invoke', function_name, '--event', event_file],
            capture_output=True,
            text=True,
            timeout=30
        )
        return json.loads(result.stdout)
    finally:
        os.unlink(event_file)

def test_get_user_sam_local():
    event = {
        'httpMethod': 'GET',
        'pathParameters': {'userId': '123'}
    }
    response = invoke_sam_local('GetUser', event)
    
    assert response['statusCode'] == 200
    body = json.loads(response['body'])
    assert 'id' in body

Testing Event Sources

Each event source produces a different event shape. Test against realistic event structures.

SQS Trigger

def test_process_sqs_messages():
    event = {
        'Records': [
            {
                'messageId': 'msg-1',
                'body': json.dumps({'orderId': 'order-123', 'total': 99.99}),
                'attributes': {
                    'ApproximateReceiveCount': '1'
                },
                'receiptHandle': 'handle-1',
                'eventSource': 'aws:sqs'
            },
            {
                'messageId': 'msg-2',
                'body': json.dumps({'orderId': 'order-456', 'total': 49.99}),
                'attributes': {
                    'ApproximateReceiveCount': '1'
                },
                'receiptHandle': 'handle-2',
                'eventSource': 'aws:sqs'
            }
        ]
    }
    
    with patch('handler.OrderService') as mock_service_class:
        mock_service = Mock()
        mock_service_class.return_value = mock_service
        
        result = process_orders_handler(event, {})
        
        assert mock_service.process.call_count == 2

def test_partial_batch_failure():
    """Test that failed messages are returned for retry"""
    event = {'Records': [
        {'messageId': 'msg-1', 'body': json.dumps({'orderId': 'valid'})},
        {'messageId': 'msg-2', 'body': 'invalid-json-that-will-fail'}
    ]}
    
    result = process_orders_handler(event, {})
    
    # Handler should report msg-2 as failed (partial batch response)
    assert 'batchItemFailures' in result
    failed_ids = [f['itemIdentifier'] for f in result['batchItemFailures']]
    assert 'msg-2' in failed_ids
    assert 'msg-1' not in failed_ids  # successful message should not be retried

S3 Event

def test_process_s3_upload():
    event = {
        'Records': [{
            'eventSource': 'aws:s3',
            'eventName': 'ObjectCreated:Put',
            's3': {
                'bucket': {'name': 'my-upload-bucket'},
                'object': {
                    'key': 'uploads/image.jpg',
                    'size': 1024
                }
            }
        }]
    }
    
    with patch('handler.ImageProcessor') as mock_processor:
        mock_instance = Mock()
        mock_processor.return_value = mock_instance
        
        image_processor_handler(event, {})
        
        mock_instance.process.assert_called_once_with(
            bucket='my-upload-bucket',
            key='uploads/image.jpg'
        )

End-to-End Testing Against Deployed Functions

E2E tests invoke the real deployed function to catch IAM permission issues, environment variable problems, and integration failures:

import boto3
import json
import pytest

# Only run E2E tests when explicitly enabled
@pytest.mark.skipif(
    os.environ.get('RUN_E2E_TESTS') != 'true',
    reason="E2E tests require deployed Lambda and AWS credentials"
)
def test_deployed_get_user():
    lambda_client = boto3.client('lambda', region_name='us-east-1')
    
    payload = {
        'httpMethod': 'GET',
        'pathParameters': {'userId': 'test-user-e2e'},
        'headers': {}
    }
    
    response = lambda_client.invoke(
        FunctionName='GetUser-production',
        InvocationType='RequestResponse',
        Payload=json.dumps(payload)
    )
    
    result = json.loads(response['Payload'].read())
    
    assert response['StatusCode'] == 200
    assert result['statusCode'] in [200, 404]  # valid response codes
    
    # Verify no function error
    assert 'FunctionError' not in response

Run E2E tests after deployment in CI:

- name: Deploy
  run: sam deploy --no-confirm-changeset

- name: Run E2E tests
  env:
    RUN_E2E_TESTS: "true"
  run: pytest tests/e2e/

Testing Timeouts and Cold Starts

Lambda has a maximum execution timeout. Test that your function handles long operations within limits:

import time

def test_function_completes_within_timeout():
    """Verify handler completes well within Lambda timeout"""
    event = make_large_event()  # worst-case payload
    
    start = time.time()
    result = lambda_handler(event, Mock(remaining_time_in_millis=lambda: 30000))
    elapsed = time.time() - start
    
    # Should complete within 10 seconds (Lambda timeout is 30s)
    assert elapsed < 10, f"Handler took {elapsed:.2f}s, too slow"
    assert result['statusCode'] == 200

Lambda function testing requires discipline at all three levels: fast unit tests for logic, SAM local for runtime environment validation, and E2E for deployment verification. Skip any layer and you'll ship bugs that only appear in production.

Lambda functions are one part of your stack. For monitoring deployed serverless functions, detecting endpoint regressions, and 24/7 uptime checks, HelpMeTest covers the full application layer with usage-based pricing starting at $0.003 per test run.