Database Testing

Testing DynamoDB Locally: DynamoDB Local, Localstack, and AWS SDK Mocking

HelpMeTest

16 May 2026 — 6 min read

DynamoDB is fast and serverless in production — but testing against live AWS DynamoDB is expensive, slow, and requires real AWS credentials. DynamoDB Local, LocalStack, and AWS SDK mocking solve this, letting you run fast, reliable tests without touching AWS. This guide covers all three approaches across Python and Node.js.

Three Approaches to DynamoDB Testing

Approach	Fidelity	Speed	Cost	Best For
AWS SDK mocks	Low	Fastest	Free	Unit tests, pure logic
DynamoDB Local (JAR)	High	Fast	Free	Most tests
LocalStack	Very High	Moderate	Free/Pro	Multi-service AWS tests
Real DynamoDB	Perfect	Slow	$$	Final integration tests

DynamoDB Local: The Official Emulator

DynamoDB Local is an official AWS JAR file that runs a complete DynamoDB-compatible server locally. It supports virtually all DynamoDB features.

Option 1: Docker Image

docker run -p 8000:8000 amazon/dynamodb-local:latest \
  -jar DynamoDBLocal.jar -sharedDb -inMemory

For tests, add -inMemory to avoid disk writes and -sharedDb to use a single shared database file (or omit for per-connection isolation).

Option 2: Testcontainers

# Python
from testcontainers.core.container import DockerContainer
import boto3

class DynamoDBLocalContainer(DockerContainer):
    def __init__(self):
        super().__init__("amazon/dynamodb-local:latest")
        self.with_command("-jar DynamoDBLocal.jar -inMemory -sharedDb")
        self.with_exposed_ports(8000)
    
    def get_endpoint_url(self):
        return f"http://localhost:{self.get_exposed_port(8000)}"

// Node.js
const { GenericContainer } = require('testcontainers');

const container = await new GenericContainer('amazon/dynamodb-local:latest')
  .withCommand(['-jar', 'DynamoDBLocal.jar', '-inMemory', '-sharedDb'])
  .withExposedPorts(8000)
  .start();

const endpoint = `http://localhost:${container.getMappedPort(8000)}`;

Python Testing with boto3

Setup and Fixtures

import pytest
import boto3
from testcontainers.core.container import DockerContainer

@pytest.fixture(scope="session")
def dynamodb_container():
    container = DockerContainer("amazon/dynamodb-local:latest")
    container.with_command("-jar DynamoDBLocal.jar -inMemory -sharedDb")
    container.with_exposed_ports(8000)
    container.start()
    yield container
    container.stop()

@pytest.fixture(scope="session")
def dynamodb_resource(dynamodb_container):
    port = dynamodb_container.get_exposed_port(8000)
    return boto3.resource(
        'dynamodb',
        endpoint_url=f'http://localhost:{port}',
        region_name='us-east-1',
        aws_access_key_id='fake',
        aws_secret_access_key='fake'
    )

@pytest.fixture(scope="session")
def dynamodb_client(dynamodb_container):
    port = dynamodb_container.get_exposed_port(8000)
    return boto3.client(
        'dynamodb',
        endpoint_url=f'http://localhost:{port}',
        region_name='us-east-1',
        aws_access_key_id='fake',
        aws_secret_access_key='fake'
    )

@pytest.fixture(scope="function")
def users_table(dynamodb_resource):
    table = dynamodb_resource.create_table(
        TableName='Users',
        KeySchema=[
            {'AttributeName': 'pk', 'KeyType': 'HASH'},
            {'AttributeName': 'sk', 'KeyType': 'RANGE'}
        ],
        AttributeDefinitions=[
            {'AttributeName': 'pk', 'AttributeType': 'S'},
            {'AttributeName': 'sk', 'AttributeType': 'S'},
        ],
        BillingMode='PAY_PER_REQUEST'
    )
    table.wait_until_exists()
    yield table
    table.delete()

Testing CRUD Operations

def test_put_and_get_item(users_table):
    users_table.put_item(Item={
        'pk': 'USER#alice',
        'sk': 'PROFILE',
        'name': 'Alice',
        'email': 'alice@example.com',
        'age': 30
    })
    
    response = users_table.get_item(Key={
        'pk': 'USER#alice',
        'sk': 'PROFILE'
    })
    
    item = response.get('Item')
    assert item is not None
    assert item['name'] == 'Alice'
    assert item['email'] == 'alice@example.com'

def test_conditional_write(users_table):
    """Test optimistic locking with condition expressions"""
    users_table.put_item(Item={
        'pk': 'USER#bob',
        'sk': 'PROFILE',
        'name': 'Bob',
        'version': 1
    })
    
    # Update succeeds with correct version
    users_table.update_item(
        Key={'pk': 'USER#bob', 'sk': 'PROFILE'},
        UpdateExpression='SET #name = :name, version = :new_version',
        ConditionExpression='version = :expected_version',
        ExpressionAttributeNames={'#name': 'name'},
        ExpressionAttributeValues={
            ':name': 'Robert',
            ':new_version': 2,
            ':expected_version': 1
        }
    )
    
    # Verify update
    item = users_table.get_item(
        Key={'pk': 'USER#bob', 'sk': 'PROFILE'}
    )['Item']
    assert item['name'] == 'Robert'
    assert item['version'] == 2

def test_conditional_write_conflict(users_table):
    """Conflicting update is rejected"""
    from botocore.exceptions import ClientError
    
    users_table.put_item(Item={
        'pk': 'USER#carol', 'sk': 'PROFILE', 'version': 1
    })
    
    with pytest.raises(ClientError) as exc:
        users_table.update_item(
            Key={'pk': 'USER#carol', 'sk': 'PROFILE'},
            UpdateExpression='SET version = :v',
            ConditionExpression='version = :expected',
            ExpressionAttributeValues={':v': 2, ':expected': 99}  # Wrong version
        )
    
    assert exc.value.response['Error']['Code'] == 'ConditionalCheckFailedException'

def test_query_with_sort_key(users_table):
    """Test Query operation with sort key conditions"""
    user_id = 'USER#dana'
    
    users_table.put_item(Item={'pk': user_id, 'sk': 'ORDER#2024-01-01', 'total': 99})
    users_table.put_item(Item={'pk': user_id, 'sk': 'ORDER#2024-02-01', 'total': 149})
    users_table.put_item(Item={'pk': user_id, 'sk': 'ORDER#2024-03-01', 'total': 49})
    users_table.put_item(Item={'pk': user_id, 'sk': 'PROFILE', 'name': 'Dana'})
    
    from boto3.dynamodb.conditions import Key
    
    response = users_table.query(
        KeyConditionExpression=Key('pk').eq(user_id) & Key('sk').begins_with('ORDER#')
    )
    
    items = response['Items']
    assert len(items) == 3
    assert all(item['sk'].startswith('ORDER#') for item in items)

Testing with moto (AWS Mock Library)

For pure unit tests without Docker, moto mocks AWS services in-process:

import boto3
import pytest
from moto import mock_aws

@pytest.fixture
def aws_credentials(monkeypatch):
    monkeypatch.setenv('AWS_ACCESS_KEY_ID', 'fake')
    monkeypatch.setenv('AWS_SECRET_ACCESS_KEY', 'fake')
    monkeypatch.setenv('AWS_DEFAULT_REGION', 'us-east-1')

@mock_aws
def test_user_service_creates_item(aws_credentials):
    # Create table in mock
    client = boto3.client('dynamodb', region_name='us-east-1')
    client.create_table(
        TableName='Users',
        KeySchema=[{'AttributeName': 'pk', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'pk', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )
    
    # Import your service (it uses boto3 internally)
    from myapp.user_service import UserService
    service = UserService(table_name='Users', region='us-east-1')
    
    user_id = service.create_user(email='test@example.com', name='Test')
    user = service.get_user(user_id)
    
    assert user['email'] == 'test@example.com'

Node.js Testing with aws-sdk

DynamoDB Client Configuration for Tests

// config/dynamodb.js
const { DynamoDBClient } = require('@aws-sdk/client-dynamodb');
const { DynamoDBDocumentClient } = require('@aws-sdk/lib-dynamodb');

function createClient(options = {}) {
  const clientConfig = {
    region: process.env.AWS_REGION || 'us-east-1',
    ...options
  };
  
  if (process.env.DYNAMODB_ENDPOINT) {
    clientConfig.endpoint = process.env.DYNAMODB_ENDPOINT;
    clientConfig.credentials = {
      accessKeyId: 'fake',
      secretAccessKey: 'fake'
    };
  }
  
  const client = new DynamoDBClient(clientConfig);
  return DynamoDBDocumentClient.from(client);
}

module.exports = { createClient };

Jest Test Setup

// jest.setup.js
const { GenericContainer } = require('testcontainers');

let container;

beforeAll(async () => {
  container = await new GenericContainer('amazon/dynamodb-local:latest')
    .withCommand(['-jar', 'DynamoDBLocal.jar', '-inMemory', '-sharedDb'])
    .withExposedPorts(8000)
    .start();
  
  process.env.DYNAMODB_ENDPOINT = `http://localhost:${container.getMappedPort(8000)}`;
  process.env.AWS_REGION = 'us-east-1';
}, 30000);

afterAll(async () => {
  if (container) await container.stop();
});

Table Setup and Tests

const { DynamoDBClient, CreateTableCommand, DeleteTableCommand } = require('@aws-sdk/client-dynamodb');
const { DynamoDBDocumentClient, PutCommand, GetCommand, QueryCommand, UpdateCommand } = require('@aws-sdk/lib-dynamodb');

describe('ProductRepository', () => {
  let docClient;
  const TABLE_NAME = 'Products';
  
  beforeAll(async () => {
    const client = new DynamoDBClient({
      endpoint: process.env.DYNAMODB_ENDPOINT,
      region: 'us-east-1',
      credentials: { accessKeyId: 'fake', secretAccessKey: 'fake' }
    });
    
    docClient = DynamoDBDocumentClient.from(client);
    
    await client.send(new CreateTableCommand({
      TableName: TABLE_NAME,
      KeySchema: [
        { AttributeName: 'pk', KeyType: 'HASH' },
        { AttributeName: 'sk', KeyType: 'RANGE' }
      ],
      AttributeDefinitions: [
        { AttributeName: 'pk', AttributeType: 'S' },
        { AttributeName: 'sk', AttributeType: 'S' },
      ],
      BillingMode: 'PAY_PER_REQUEST'
    }));
  });
  
  afterAll(async () => {
    const client = new DynamoDBClient({ endpoint: process.env.DYNAMODB_ENDPOINT });
    await client.send(new DeleteTableCommand({ TableName: TABLE_NAME }));
  });
  
  test('stores and retrieves a product', async () => {
    await docClient.send(new PutCommand({
      TableName: TABLE_NAME,
      Item: {
        pk: 'PRODUCT#widget-001',
        sk: 'METADATA',
        name: 'Widget',
        price: 9.99,
        category: 'hardware'
      }
    }));
    
    const result = await docClient.send(new GetCommand({
      TableName: TABLE_NAME,
      Key: { pk: 'PRODUCT#widget-001', sk: 'METADATA' }
    }));
    
    expect(result.Item.name).toBe('Widget');
    expect(result.Item.price).toBe(9.99);
  });
  
  test('handles missing items gracefully', async () => {
    const result = await docClient.send(new GetCommand({
      TableName: TABLE_NAME,
      Key: { pk: 'PRODUCT#nonexistent', sk: 'METADATA' }
    }));
    
    expect(result.Item).toBeUndefined();
  });
  
  test('atomic counter increment', async () => {
    await docClient.send(new PutCommand({
      TableName: TABLE_NAME,
      Item: { pk: 'COUNTER#pageviews', sk: 'GLOBAL', count: 0 }
    }));
    
    // Increment atomically
    const result = await docClient.send(new UpdateCommand({
      TableName: TABLE_NAME,
      Key: { pk: 'COUNTER#pageviews', sk: 'GLOBAL' },
      UpdateExpression: 'SET #count = #count + :incr',
      ExpressionAttributeNames: { '#count': 'count' },
      ExpressionAttributeValues: { ':incr': 1 },
      ReturnValues: 'UPDATED_NEW'
    }));
    
    expect(result.Attributes.count).toBe(1);
  });
});

Mocking aws-sdk with Jest

For pure unit tests with no container:

// __mocks__/@aws-sdk/client-dynamodb.js
const mockSend = jest.fn();

const DynamoDBClient = jest.fn().mockImplementation(() => ({
  send: mockSend
}));

module.exports = { DynamoDBClient, mockSend };

// In test files
describe('UserService unit tests', () => {
  const { mockSend } = require('@aws-sdk/client-dynamodb');
  
  beforeEach(() => {
    mockSend.mockReset();
  });
  
  test('get user returns null when not found', async () => {
    mockSend.mockResolvedValueOnce({ Item: undefined });
    
    const service = new UserService('Users');
    const result = await service.getUser('nonexistent');
    
    expect(result).toBeNull();
    expect(mockSend).toHaveBeenCalledTimes(1);
  });
  
  test('handles DynamoDB throttling with retry', async () => {
    const { ProvisionedThroughputExceededException } = require('@aws-sdk/client-dynamodb');
    
    mockSend
      .mockRejectedValueOnce(new ProvisionedThroughputExceededException({ $metadata: {} }))
      .mockResolvedValueOnce({ Item: { pk: 'USER#123', name: 'Alice' } });
    
    const user = await service.getUserWithRetry('123');
    expect(user.name).toBe('Alice');
    expect(mockSend).toHaveBeenCalledTimes(2);
  });
});

CI/CD Integration

GitHub Actions

name: DynamoDB Tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Start DynamoDB Local
        run: |
          docker run -d -p 8000:8000 amazon/dynamodb-local:latest \
            -jar DynamoDBLocal.jar -inMemory -sharedDb
      
      - name: Wait for DynamoDB Local
        run: |
          for i in {1..30}; do
            if curl -s http://localhost:8000 > /dev/null; then break; fi
            sleep 1
          done
      
      - name: Run tests
        run: npm test
        env:
          DYNAMODB_ENDPOINT: http://localhost:8000
          AWS_REGION: us-east-1
          AWS_ACCESS_KEY_ID: fake
          AWS_SECRET_ACCESS_KEY: fake

Monitoring DynamoDB in Production

Local testing covers logic, but DynamoDB production failures are often capacity-related:

ProvisionedThroughputExceededException under traffic spikes
Hot partition keys causing uneven load
GSI throttling separate from main table
Backup/restore operations consuming capacity

HelpMeTest monitors your live API endpoints that depend on DynamoDB, running continuous end-to-end tests every 5 minutes. Get alerted immediately when DynamoDB throttling causes your application to degrade — before users notice. Write tests in plain English, no infrastructure required.

Summary

Start with moto (Python) or Jest SDK mocks for pure unit tests — zero infrastructure, maximum speed. Use DynamoDB Local via Testcontainers for integration tests that need full DynamoDB fidelity. Reserve real DynamoDB for final pre-production smoke tests. Monitor production continuously to catch the capacity and operational issues that local tests can't simulate.