Testing DynamoDB: LocalStack, Mocking, and Integration Test Strategies

Testing DynamoDB: LocalStack, Mocking, and Integration Test Strategies

Testing DynamoDB effectively requires a combination of local emulation with LocalStack, fast unit tests with mocked clients, and targeted integration tests that validate single-table designs, GSIs/LSIs, and PartiQL queries — without touching production AWS.

Key Takeaways

Use LocalStack for integration tests, not the real AWS endpoint. LocalStack emulates DynamoDB locally at full fidelity — no IAM costs, no latency, no accidental writes to production tables.

Test your access patterns, not just CRUD. DynamoDB's query model is access-pattern-first. Your tests should validate that every GSI and LSI returns the correct items under the correct conditions.

Mock the client for pure unit tests. Repository classes that wrap DynamoDB calls should be unit-tested with a mocked DynamoDBDocumentClient so tests stay fast and deterministic.

Why Testing DynamoDB Is Different

DynamoDB's NoSQL model punishes bad access patterns at scale, but testing those patterns before production is where teams consistently fall short. Unlike relational databases, you can't just run SELECT * and validate results — your data model is inseparable from your query model. A GSI that looked correct during development can silently return empty results in production because you wrote items in the wrong partition key format.

The good news: DynamoDB has excellent local emulation options, and the AWS SDK makes mocking straightforward. The bad news: most tutorials only cover the happy path.

LocalStack vs DynamoDB Local

AWS ships an official DynamoDB Local JAR. LocalStack ships a Docker image that emulates the entire AWS API surface, including DynamoDB. Which should you use?

DynamoDB Local (AWS official):

  • Runs as a JAR or Docker image
  • Supports most DynamoDB features
  • Free, no API key needed
  • Does not emulate IAM, VPC, or other AWS services

LocalStack (community/pro):

  • Docker-first, works with docker-compose
  • Emulates 50+ AWS services in one container
  • Free tier covers DynamoDB fully
  • Better for integration tests that touch S3, SQS, Lambda alongside DynamoDB

For pure DynamoDB testing, both work. If your application uses DynamoDB Streams, Lambda triggers, or SQS queues together, LocalStack is the right choice.

Starting LocalStack

# docker-compose.yml
version: "3.8"
services:
  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=dynamodb
      - DEFAULT_REGION=us-east-1
      - DEBUG=1
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
docker-compose up -d localstack

Integration Tests with Jest and LocalStack

Here is a real-world example: a user session store backed by DynamoDB with a single-table design. The table uses a composite key (PK, SK) and a GSI for querying sessions by user ID.

Table Setup

// test/helpers/dynamo-setup.ts
import {
  DynamoDBClient,
  CreateTableCommand,
  DeleteTableCommand,
} from "@aws-sdk/client-dynamodb";

const client = new DynamoDBClient({
  endpoint: "http://localhost:4566",
  region: "us-east-1",
  credentials: { accessKeyId: "test", secretAccessKey: "test" },
});

export async function createTestTable(tableName: string) {
  await client.send(
    new CreateTableCommand({
      TableName: tableName,
      KeySchema: [
        { AttributeName: "PK", KeyType: "HASH" },
        { AttributeName: "SK", KeyType: "RANGE" },
      ],
      AttributeDefinitions: [
        { AttributeName: "PK", AttributeType: "S" },
        { AttributeName: "SK", AttributeType: "S" },
        { AttributeName: "GSI1PK", AttributeType: "S" },
        { AttributeName: "GSI1SK", AttributeType: "S" },
      ],
      GlobalSecondaryIndexes: [
        {
          IndexName: "GSI1",
          KeySchema: [
            { AttributeName: "GSI1PK", KeyType: "HASH" },
            { AttributeName: "GSI1SK", KeyType: "RANGE" },
          ],
          Projection: { ProjectionType: "ALL" },
          ProvisionedThroughput: { ReadCapacityUnits: 5, WriteCapacityUnits: 5 },
        },
      ],
      BillingMode: "PAY_PER_REQUEST",
    })
  );
}

export async function deleteTestTable(tableName: string) {
  await client.send(new DeleteTableCommand({ TableName: tableName }));
}

Repository Under Test

// src/session-repository.ts
import { DynamoDBDocumentClient, PutCommand, QueryCommand, DeleteCommand } from "@aws-sdk/lib-dynamodb";

export interface Session {
  sessionId: string;
  userId: string;
  createdAt: string;
  expiresAt: string;
}

export class SessionRepository {
  constructor(
    private readonly client: DynamoDBDocumentClient,
    private readonly tableName: string
  ) {}

  async put(session: Session): Promise<void> {
    await this.client.send(
      new PutCommand({
        TableName: this.tableName,
        Item: {
          PK: `SESSION#${session.sessionId}`,
          SK: `SESSION#${session.sessionId}`,
          GSI1PK: `USER#${session.userId}`,
          GSI1SK: `SESSION#${session.createdAt}`,
          ...session,
        },
      })
    );
  }

  async getByUser(userId: string): Promise<Session[]> {
    const result = await this.client.send(
      new QueryCommand({
        TableName: this.tableName,
        IndexName: "GSI1",
        KeyConditionExpression: "GSI1PK = :pk",
        ExpressionAttributeValues: { ":pk": `USER#${userId}` },
      })
    );
    return (result.Items ?? []) as Session[];
  }
}

Jest Integration Tests

// test/session-repository.test.ts
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient } from "@aws-sdk/lib-dynamodb";
import { SessionRepository } from "../src/session-repository";
import { createTestTable, deleteTestTable } from "./helpers/dynamo-setup";

const TABLE_NAME = `sessions-test-${Date.now()}`;

let repo: SessionRepository;

beforeAll(async () => {
  await createTestTable(TABLE_NAME);
  const baseClient = new DynamoDBClient({
    endpoint: "http://localhost:4566",
    region: "us-east-1",
    credentials: { accessKeyId: "test", secretAccessKey: "test" },
  });
  repo = new SessionRepository(DynamoDBDocumentClient.from(baseClient), TABLE_NAME);
});

afterAll(async () => {
  await deleteTestTable(TABLE_NAME);
});

test("puts a session and retrieves it via GSI", async () => {
  const session = {
    sessionId: "sess-001",
    userId: "user-42",
    createdAt: "2026-05-17T10:00:00Z",
    expiresAt: "2026-05-18T10:00:00Z",
  };

  await repo.put(session);
  const results = await repo.getByUser("user-42");

  expect(results).toHaveLength(1);
  expect(results[0].sessionId).toBe("sess-001");
  expect(results[0].expiresAt).toBe("2026-05-18T10:00:00Z");
});

test("GSI returns sessions ordered by createdAt", async () => {
  const userId = "user-99";
  await repo.put({ sessionId: "s1", userId, createdAt: "2026-05-17T08:00:00Z", expiresAt: "2026-05-18T08:00:00Z" });
  await repo.put({ sessionId: "s2", userId, createdAt: "2026-05-17T09:00:00Z", expiresAt: "2026-05-18T09:00:00Z" });
  await repo.put({ sessionId: "s3", userId, createdAt: "2026-05-17T10:00:00Z", expiresAt: "2026-05-18T10:00:00Z" });

  const results = await repo.getByUser(userId);
  const ids = results.map((s) => s.sessionId);

  expect(ids).toEqual(["s1", "s2", "s3"]);
});

Mocking DynamoDB for Unit Tests

Integration tests are slow by nature. For pure business logic that happens to call DynamoDB, use jest.mock or aws-sdk-client-mock:

// test/session-repository.unit.test.ts
import { mockClient } from "aws-sdk-client-mock";
import { DynamoDBDocumentClient, PutCommand, QueryCommand } from "@aws-sdk/lib-dynamodb";
import { SessionRepository } from "../src/session-repository";

const ddbMock = mockClient(DynamoDBDocumentClient);

beforeEach(() => ddbMock.reset());

test("getByUser maps GSI results to Session objects", async () => {
  ddbMock.on(QueryCommand).resolves({
    Items: [
      {
        sessionId: "sess-abc",
        userId: "user-1",
        createdAt: "2026-05-17T10:00:00Z",
        expiresAt: "2026-05-18T10:00:00Z",
      },
    ],
  });

  const repo = new SessionRepository(DynamoDBDocumentClient.from({} as any), "table");
  const sessions = await repo.getByUser("user-1");

  expect(sessions[0].sessionId).toBe("sess-abc");
});

Testing PartiQL Queries

DynamoDB supports PartiQL — a SQL-compatible query language. Test PartiQL queries against LocalStack the same way, using ExecuteStatementCommand:

import { ExecuteStatementCommand } from "@aws-sdk/client-dynamodb";
import { marshall, unmarshall } from "@aws-sdk/util-dynamodb";

test("PartiQL filters expired sessions", async () => {
  const now = new Date().toISOString();

  const result = await baseClient.send(
    new ExecuteStatementCommand({
      Statement: `SELECT * FROM "${TABLE_NAME}" WHERE PK BEGINS_WITH 'SESSION#' AND expiresAt > ?`,
      Parameters: [{ S: now }],
    })
  );

  const items = (result.Items ?? []).map(unmarshall);
  expect(items.every((i) => i.expiresAt > now)).toBe(true);
});

Testing LSIs

Local Secondary Indexes share the same partition key as the base table but allow a different sort key. Test them the same as GSIs — create the LSI at table creation time and query it directly:

// In CreateTableCommand:
LocalSecondaryIndexes: [
  {
    IndexName: "LSI-ExpiresAt",
    KeySchema: [
      { AttributeName: "PK", KeyType: "HASH" },
      { AttributeName: "expiresAt", KeyType: "RANGE" },
    ],
    Projection: { ProjectionType: "KEYS_ONLY" },
  },
],

LSI queries in tests catch two common bugs: items that were written without the LSI sort key attribute (they simply don't appear in LSI results — a silent failure), and projection mismatches where you try to read an attribute that wasn't projected.

pytest Integration Tests

For Python services, moto is the standard DynamoDB mock:

# test_session_repo.py
import boto3
import pytest
from moto import mock_dynamodb
from myapp.session_repo import SessionRepository

@pytest.fixture
def dynamodb_table():
    with mock_dynamodb():
        ddb = boto3.resource("dynamodb", region_name="us-east-1")
        table = ddb.create_table(
            TableName="sessions",
            KeySchema=[
                {"AttributeName": "PK", "KeyType": "HASH"},
                {"AttributeName": "SK", "KeyType": "RANGE"},
            ],
            AttributeDefinitions=[
                {"AttributeName": "PK", "AttributeType": "S"},
                {"AttributeName": "SK", "AttributeType": "S"},
            ],
            BillingMode="PAY_PER_REQUEST",
        )
        yield table

def test_put_and_get(dynamodb_table):
    repo = SessionRepository(dynamodb_table)
    repo.put("sess-1", "user-1", "2026-05-18T10:00:00Z")
    sessions = repo.get_by_user("user-1")
    assert len(sessions) == 1
    assert sessions[0]["sessionId"] == "sess-1"

Tips for Reliable DynamoDB Tests

Use unique table names per test run. Suffix with Date.now() or a UUID. This prevents state leakage between parallel test suites and makes cleanup optional (tables are ephemeral in LocalStack).

Seed data in beforeEach, not beforeAll. DynamoDB tests that share table state across tests are brittle. Reset per test.

Assert on absence, not just presence. A common bug is writing items to the wrong partition. Test that getByUser("user-A") returns zero results for items written under user-B.

Test conditional writes. DynamoDB's ConditionExpression is a primary concurrency primitive. Test the happy path (condition passes) and the conflict path (ConditionalCheckFailedException).


HelpMeTest can run your DynamoDB integration tests automatically on every pull request — sign up free.

Read more