模拟AWS服务和Lambda最佳实践

我正在开发一个简单的AWS lambda函数，该函数由DynamoDB Streams事件触发，应该将除REMOVE事件外的所有记录转发到SQS队列。该功能按预期工作，这并不奇怪。

我想写一个单元测试来测试当它是DELETE事件时不向SQS提交任何内容的行为。我第一次尝试使用aws-sdk-mock。正如您在函数代码中看到的，我试图通过在处理程序代码之外初始化SQS客户端来遵守lambda最佳实践。显然，这阻止了aws-sdk-mock模拟SQS服务(GitHub上有一个问题：https://github.com/dwyl/aws-sdk-mock/issues/206)。

然后，我尝试使用jest来模拟SQS，这需要更多的代码才能使其正确，但我最终遇到了同样的问题，被要求将SQS的初始化放在处理程序函数中，这违反了lambda最佳实践。

如何为该函数编写单元测试，同时让SQS客户端(const sqs: SQS = new SQS()(的初始化在处理程序之外？我是以错误的方式嘲笑服务，还是为了更容易测试而更改了处理程序的结构？

我知道这个lambda函数非常直接，可能没有必要进行单元测试，但我必须用更复杂的逻辑编写更多的lambda，我认为这个函数非常适合演示这个问题。

指数.ts

import {DynamoDBStreamEvent, DynamoDBStreamHandler} from "aws-lambda";
import SQS = require("aws-sdk/clients/sqs");
import DynamoDB = require("aws-sdk/clients/dynamodb");
const sqs: SQS = new SQS()
export const handleDynamoDbEvent: DynamoDBStreamHandler = async (event: DynamoDBStreamEvent, context, callback) => {
const QUEUE_URL = process.env.TARGET_QUEUE_URL
if (QUEUE_URL.length == 0) {
throw new Error('TARGET_QUEUE_URL not set or empty')
}
await Promise.all(
event.Records
.filter(_ => _.eventName !== "REMOVE")
.map((record) => {
const unmarshalled = DynamoDB.Converter.unmarshall(record.dynamodb.NewImage);
let request: SQS.SendMessageRequest = {
MessageAttributes: {
"EVENT_NAME": {
DataType: "String",
StringValue: record.eventName
}
},
MessageBody: JSON.stringify(unmarshalled),
QueueUrl: QUEUE_URL,
}
return sqs.sendMessage(request).promise()
})
);
}

索引.spec.ts

import {DynamoDBRecord, DynamoDBStreamEvent, StreamRecord} from "aws-lambda";
import {AttributeValue} from "aws-lambda/trigger/dynamodb-stream";
import {handleDynamoDbEvent} from "./index";
import {AWSError} from "aws-sdk/lib/error";
import {PromiseResult, Request} from "aws-sdk/lib/request";
import * as SQS from "aws-sdk/clients/sqs";
import {mocked} from "ts-jest/utils";
import DynamoDB = require("aws-sdk/clients/dynamodb");

jest.mock('aws-sdk/clients/sqs', () => {
return jest.fn().mockImplementation(() => {
return {
sendMessage: (params: SQS.Types.SendMessageRequest, callback?: (err: AWSError, data: SQS.Types.SendMessageResult) => void): Request<SQS.Types.SendMessageResult, AWSError> => {
// @ts-ignore
const Mock = jest.fn<Request<SQS.Types.SendMessageResult, AWSError>>(()=>{
return {
promise: (): Promise<PromiseResult<SQS.Types.SendMessageResult, AWSError>> => {
return new Promise<PromiseResult<SQS.SendMessageResult, AWSError>>(resolve => {
resolve(null)
})
}
}
})
return new Mock()
}
}
})
});

describe.only('Handler test', () => {
const mockedSqs = mocked(SQS, true)
process.env.TARGET_QUEUE_URL = 'test'
const OLD_ENV = process.env;
beforeEach(() => {
mockedSqs.mockClear()
jest.resetModules();
process.env = {...OLD_ENV};
});
it('should write INSERT events to SQS', async () => {
console.log('Starting test')
await handleDynamoDbEvent(createEvent(), null, null)
expect(mockedSqs).toHaveBeenCalledTimes(1)
});
})

我应该如何处理这个问题的大致想法：

我将为消息客户端创建一个接口，而不是在主函数中进行实际的SQS发送/操作。类似于以下内容：

interface QueueClient {
send(eventName: string, body: string): Promise<any>;
}

并创建一个实现该接口的实际类来与SQS进行交互：

class SQSQueueClient implements QueueClient {
queueUrl: string
sqs: SQS
constructor() {
this.queueUrl = process.env.TARGET_QUEUE_URL;
if (this.queueUrl.length == 0) {
throw new Error('TARGET_QUEUE_URL not set or empty')
}
this.sqs = new SQS();
}
send(eventName: string, body: string): Promise<any> {
let request: SQS.SendMessageRequest = {
MessageAttributes: {
"EVENT_NAME": {
DataType: "String",
StringValue: eventName
}
},
MessageBody: body,
QueueUrl: this.queueUrl,
}
return this.sqs.sendMessage()
}
}

此类了解如何将数据转换为SQS格式的详细信息

然后我将把主函数分成2。入口点只是解析队列url，创建sqs队列客户端的实际实例并调用process()。主逻辑在process()中

const queueClient = new SQSQueueClient();
export const handleDynamoDbEvent: DynamoDBStreamHandler = async (event: DynamoDBStreamEvent, context, callback) => {
return process(queueClient, event);
}
export const process = async (queueClient: QueueClient, event: DynamoDBStreamEvent) => {
return await Promise.all(
event.Records
.filter(_ => _.eventName !== "REMOVE")
.map((record) => {
const unmarshalled = DynamoDB.Converter.unmarshall(record.dynamodb.NewImage);
return queueClient.send(record.eventName, JSON.stringify(unmarshalled));
})
);
}

现在在process()中测试主逻辑要容易得多。您可以提供一个通过手写实现接口QueueClient的mock实例，也可以使用您喜欢的任何mocking框架
对于SQSQueueClient类，单元测试没有太多好处，所以我将更多地依赖集成测试(例如，使用类似localstack的东西(

我现在没有实际的IDE，所以如果到处都有语法错误，请原谅我

我添加了一个从处理程序函数内部调用的初始化方法。如果之前已经调用过它，它会立即返回，否则将初始化SQS客户端。它可以很容易地扩展到初始化其他客户端。

这符合lambda最佳实践，并使测试代码正常工作。

let sqs: SQS = null
let initialized = false
export const handleDynamoDbEvent: DynamoDBStreamHandler = async (event: DynamoDBStreamEvent, context, callback) => {
init()
const QUEUE_URL = process.env.TARGET_QUEUE_URL
if (QUEUE_URL.length == 0) {
throw new Error('TARGET_QUEUE_URL not set or empty')
}
await Promise.all(
event.Records
.filter(_ => _.eventName !== "REMOVE")
.map((record) => {
const unmarshalled = DynamoDB.Converter.unmarshall(record.dynamodb.NewImage);
let request: SQS.SendMessageRequest = {
MessageAttributes: {
"EVENT_NAME": {
DataType: "String",
StringValue: record.eventName
}
},
MessageBody: JSON.stringify(unmarshalled),
QueueUrl: QUEUE_URL,
}
return sqs.sendMessage(request).promise()
})
);
}
function init() {
if (initialized) {
return
}
console.log('Initializing...')
initialized = true
sqs = new SQS()
}

相关内容

最新更新

热门标签：