Integration testing EventBridge events
Tags: aws, javascript, testing, eventbridge
Published: 2023-11-06 19:04:32

## Feature to be tested

I have a Lambda that will consume DynamoDB Stream events, transform them, then send them to EventBridge.

## Testing approach

Many articles mentioned using an SQS queue to consume the events and validate the messages as a solution. Other solutions mentioned using a Pub/Sub, or the [aws-testing-library](https://github.com/erezrokah/aws-testing-library/blob/main/src/jest/README.md#tohavelog) to check for the presence of CloudWatch log statements.

I chose the SQS queue option as it seemed like it would allow me to conduct thorough and reliable tests. Also, since I might swap out the Lambda for [EventBridge Pipes](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html) in the future, there should be no need to change any of these tests; unlike if I watched for CloudWatch logs.

### Testing EventBridge Events with SQS

The code for the test is below, and it is commented to describe each piece of code in better detail.

#### The test setup

The test requires three resources to be created:
1. An SQS Queue to consume the EventBridge Events
2. An EventBridge Rule for targeting specific EventBridge events. I will match against a record's ID so that it won't acknowledge any other non-test EB events.
3. An EventBridge Target to set the SQS queue as a target for EB events, and to only consume events matching the Eventbridge Rule.

Once these are created, each test will insert, modify, or delete a record in a DynamoDB table. This will trigger the Lambda which handles DynamoDB Stream Events and sends them to EventBridge.

#### Verifying SQS messages

Once a DynamoDB record has been updated, I use long polling to check the SQS queue for any messages. When messages are received, I verify the message body.

Since the EventBridge rule is specific to the test case, I can trust that the message received is only meant for the test SQS queue.

#### Cleanup after each test

The test scenario will contain several tests. 
As part of the setup for each test a record will be inserted, so that it can be modified. Then, the test will delete the same record as part of it's cleanup. This will create a lot of additional events that I'm not interested in testing. So, I make sure that these messages are cleared before the intended SQS message can be asserted.

Once all tests have been executed, I delete the test resources and destroy the clients.

#### Final notes before viewing the code

In the code below I create the test resources before each test run. If you are usign infrastructure as code, you might prefer to create and destroy the resources via Terraform, for example.

#### The code

```
const dynamoDBTableName = 'dynamo-db-table'
const mockRecordId = 'integration-test-emit-ddb-stream-change-event'
const sqsQueueName = 'integration-test-queue'
const eventBridgeRuleName = `${mockRecordId}-eb-rule-name`
const eventBridgeTargetID = `${mockRecordId}-eb-target-id`
const eventBridgeName = 'event-bridge-events'
const eventSource = 'event-bridge-source-service'

let eventBridgeClient: EventBridgeClient
let sqsClient: SQSClient
let dynamoDbClient: DynamoDBClient
let queueURL: string | undefined

const testRecord: DDBRecord = {
  recordType: 'someType',
  user: 'developer',
  created: '2024-01-01T00:00:00.000Z',
  recordId: mockRecordId,
  recordName: `${mockRecordId}-record-name`,
  version: '0',
  status: 'initialStatus'
}

const setupSQSQueue = async () => {
  // Create the test queue for consuming EventBridge events
  const createQueueResponse = await sqsClient.send(
    new CreateQueueCommand({
      QueueName: sqsQueueName,
    }),
  )

  // Set the queue URL so that it can be used throughout the tests
  // Alternatively, if the queue is created via IaC, you can use the `GetQueueUrlCommand` command instead.
  queueURL = createQueueResponse.QueueUrl

  // Create an EB rule for the `testRecord` defined above
  await eventBridgeClient.send(
    new PutRuleCommand({
      Name: eventBridgeRuleName,
      EventPattern: JSON.stringify({
        source: [eventSource],
        detail: {
          data: {
            recordId: [mockRecordId],
          },
        },
      }),
      State: 'ENABLED',
      EventBusName: eventBridgeName,
    }),
  )

  // Get the newly created queue's ARN
  const queueAttributes = await sqsClient.send(
    new GetQueueAttributesCommand({
      QueueUrl: queueURL,
      AttributeNames: ['QueueArn'],
    }),
  )

  // Set the queue as a target for EventBridge events
  await eventBridgeClient.send(
    new PutTargetsCommand({
      Rule: eventBridgeRuleName,
      EventBusName: eventBridgeName,
      Targets: [
        {
          Arn: queueAttributes.Attributes?.QueueArn,
          Id: eventBridgeTargetID,
        },
      ],
    }),
  )
}

// Utility function to delete a single message on the queue
const deleteMessageOnQueue = async (receiptHandle: string) => {
  await sqsClient.send(
    new DeleteMessageCommand({
      QueueUrl: queueURL,
      ReceiptHandle: receiptHandle,
    }),
  )
}

// Utility function to clear a queue of all messages
const clearMessagesOnQueue = async () => {
  const sqsMessages = await getMessageFromQueue()
  const messagesToDeleteOnCleanup = getSQSMessageReceiptHandles(sqsMessages)
  for (const receiptHandle of messagesToDeleteOnCleanup) {
    await deleteMessageOnQueue(receiptHandle)
  }

  // Rather than using the above method, the `PurgeQueueCommand` command might work for you. 
  // I didn't use this because it could take up to 60 seconds to delete the messages on the queue.
}

const deleteTestRecordFromTable = async () => {
  await dynamoDbClient.send(
    new DeleteCommand({
      TableName: dynamoDBTableName,
      Key: {
        recordId: mockRecordId,
      },
    }),
  )
}

// Delete the SQS resource
const deleteQueue = async () => {
  await sqsClient.send(new DeleteQueueCommand({ QueueUrl: queueURL }))
}

// Delete EB Target resource
const deleteEventBridgeTarget = async () => {
  await eventBridgeClient.send(
    new RemoveTargetsCommand({
      Ids: [eventBridgeTargetID],
      Rule: eventBridgeRuleName,
      Force: true,
      EventBusName: eventBridgeName,
    }),
  )
}

// Delete EB Rule resource
const deleteEventBridgeRule = async () => {
  await eventBridgeClient.send(
    new DeleteRuleCommand({
      Name: eventBridgeRuleName,
      Force: true,
      EventBusName: eventBridgeName,
    }),
  )
}

const getMessageFromQueue = async () =>
  await sqsClient.send(
    new ReceiveMessageCommand({
      MaxNumberOfMessages: 10,
      QueueUrl: queueURL,
      AttributeNames: ['All'],
      WaitTimeSeconds: 30,
      VisibilityTimeout: 0,
    }),
  )

const getSQSMessageReceiptHandles = (
  sqsMessageCommandOutput: ReceiveMessageCommandOutput,
): string[] => {
  if (!sqsMessageCommandOutput.Messages) return []

  return sqsMessageCommandOutput.Messages.map((message) => message.ReceiptHandle).filter(
    (message): message is string => message !== undefined,
  )
}

// The test block
describe('Events are sent to EventBridge WHEN records are modified in a DynamoDB table', () => {

  beforeAll(async () => {
    // Create AWS SDK Clients
    const awsConfig = { // Any AWS Config }
    eventBridgeClient = new EventBridgeClient(awsConfig)
    sqsClient = new SQSClient(awsConfig)
    dynamoDbClient = new DynamoDBClient(awsConfig)

    // Setup test resources
    await setupSQSQueue()
  })

  afterEach(async () => {
    await deleteTestRecordFromTable()
    await clearMessagesOnQueue()
  })

  afterAll(async () => {
    await deleteQueue()
    await deleteEventBridgeTarget()
    await deleteEventBridgeRule()
    dynamoDbClient.destroy()
    eventBridgeClient.destroy()
    sqsClient.destroy()
  })

  it(`SHOULD create an event with the recordType 'X' WHEN a record is inserted into the DynamoDB table`, async () => {
    // Keeping track of messages on the queue so they can be deleted later
    const messagesToDeleteOnCleanup: string[] = []

    // Insert a new record into the table
    await dynamoDbClient.send(
      new PutCommand({
        TableName: dynamoDBTableName,
        Item: testRecord,
      }),
    )

    const sqsMessages = await getMessageFromQueue()

    expect(sqsMessages.Messages).toBeDefined()
    expect(sqsMessages.Messages).toHaveLength(1)

    // Add the messages to the list for deletion on cleanup
   messagesToDeleteOnCleanup.push(...getSQSMessageReceiptHandles(sqsMessages))

    const parsedMessageBody: EventBridgeEventType<'X', DDBRecord> = JSON.parse(
      sqsMessages.Messages?.[0].Body as string,
    )

    expect(parsedMessageBody['record-type']).toEqual('X')
    expect(parsedMessageBody.detail.data).toEqual({
      ...testRecord,
      status: 'ANewStatus'
    })

    // Clenup messages created as part of this test
    for (const receiptHandle of messagesToDeleteOnCleanup) {
      await deleteMessageOnQueue(receiptHandle)
    }
  })

  describe('AND a record already exists in the database', () => {
    beforeEach(async () => {
      // Create the record in DynamoDB
      await dynamoDbClient.send(
        new PutCommand({
          TableName: dynamoDBTableName,
          Item: testRecord,
        }),
      )
      await clearMessagesOnQueue()
    })

    it(`SHOULD create an event with the recordType 'Y' WHEN a record is modified and the status is set to 'Z'`, async () => {
      const messagesToDeleteOnCleanup: string[] = []

      // Update the record in the table to have a failed status
      await dynamoDbClient.send(
        new PutCommand({
          TableName: dynamoDBTableName,
          Item: {
            ...testRecord,
            status: 'Z',
          } as DDBRecord,
        }),
      )

      const sqsMessages = await getMessageFromQueue()

      expect(sqsMessages.Messages).toBeDefined()
      expect(sqsMessages.Messages).toHaveLength(1)

      // Add the messages to the list for deletion on cleanup
      messagesToDeleteOnCleanup.push(...getSQSMessageReceiptHandles(sqsMessages))

      const parsedMessageBody: EventBridgeEventType<'Y', DDBRecord> = JSON.parse(
        sqsMessages.Messages?.[0].Body as string,
      )

      expect(parsedMessageBody['record-type']).toEqual('Y')
      expect(parsedMessageBody.detail.data).toEqual({
        ...testRecord,
        status: 'Z'
      })

      // Clenup messages created as part of this test
      for (const receiptHandle of messagesToDeleteOnCleanup) {
        await deleteMessageOnQueue(receiptHandle)
      }
    })

    ... more tests using the same approach as above
  })
})

```