LocalStackのDynamoDBを使ってみた

https://zenn.dev/marumarumeruru/articles/2f66265422d370
の続きです

LocalStackはAWSのサービスのモックです。これを使うことで単体テストを行うことができます。
今回はDynamoDBをモックして単体テストをしてみました
https://localstack.cloud/

Cloud9で環境構築

デフォルトの10GBだと容量が足りなくなったので、20GBに増やしています
Cloud9が稼働しているEC2の容量を増やすことで対応できます
https://dev.classmethod.jp/articles/expand-the-disk-size-of-cloud9/

docker-compose install

sudo curl -L https://github.com/docker/compose/releases/download/1.16.1/docker-compose-`uname -s`-`uname -m` -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose

LocalStack

docker-compose.yaml

version: '3'

services:
  localstack:
    image: localstack/localstack:latest
    environment:
      - SERVICES=dynamodb
      - DEFAULT_REGION=ap-northeast-1
      - DATA_DIR=/tmp/localstack/data
    volumes:
      - ./localstack:/tmp/localstack
    ports:
      - 4566:4566

カンマ区切りでサービスを追加することで、いろいろなサービスを使うことができます

- SERVICES=dynamodb,secretsmanager

LocalStack起動

docker-compose up -d

この状態でテスト実施します

LocalStackのDynamoDBの操作方法

aws cliでのdynamo操作例

aws --endpoint-url=http://localhost:4566 dynamodb list-tables

aws --endpoint-url=http://localhost:4566 dynamodb create-table \
    --table-name todo \
    --attribute-definitions \
        AttributeName=user,AttributeType=S \
        AttributeName=time,AttributeType=S \
    --key-schema AttributeName=user,KeyType=HASH AttributeName=time,KeyType=RANGE \
    --billing-mode PAY_PER_REQUEST

aws --endpoint-url=http://localhost:4566 dynamodb describe-table \
    --table-name todo

aws --endpoint-url=http://localhost:4566 dynamodb put-item \
    --table-name todo  \
    --item \
        '{"user": {"S": "user1"}, "time": {"S": "202204271018000"}, "title": {"S": "title1"}, "contents": {"S": "contents1"}, "done": {"N": "0"}}'

aws --endpoint-url=http://localhost:4566 dynamodb put-item \
    --table-name todo  \
    --item \
        '{"user": {"S": "user2"}, "time": {"S": "202204271028000"}, "title": {"S": "title2"}, "contents": {"S": "contents2"}, "done": {"N": "0"}}'

aws --endpoint-url=http://localhost:4566 dynamodb scan \
    --table-name todo

aws --endpoint-url=http://localhost:4566 dynamodb get-item \
    --table-name todo \
    --key '{ "user": { "S": "user1" },  "time": { "S": "202204271018000" }  }' 

aws --endpoint-url=http://localhost:4566 dynamodb update-item \
    --table-name todo \
    --key '{ "user": { "S": "user1" },  "time": { "S": "202204271018000" }  }' \
    --update-expression 'set contents = :contents, done=:done' \
    --expression-attribute-values '{ ":contents": { "S": "updated" },  ":done": { "N": "1" }  }' \
    --return-values 'ALL_NEW'

aws --endpoint-url=http://localhost:4566 dynamodb delete-item \
    --table-name todo \
    --key '{ "user": { "S": "user1" },  "time": { "S": "202204271018000" }  }'

aws --endpoint-url=http://localhost:4566 dynamodb delete-table \
    --table-name todo

pytestのサンプル

ファイル構成　作成したファイルのみ抜粋

├── docker-compose.yaml
├── insert_todo
│   ├── app.py
│   ├── __init__.py
│   └── requirements.txt
├── localstack
├── samconfig.toml
├── template.yaml
└── tests
    └── unit
        └── test_insert_todo.py

samconfig.tomlは前回のまま変更なし

insertするLambda

ENDPOINT_URLの環境変数が設定された場合、localstackを利用するようになります
os.getenvを利用して環境変数を取得した場合は、環境変数が設定されていない場合はNoneになります。
その場合、クラウド上のDynamodbを参照します

insert_todo/app.py

import boto3
import os

def lambda_handler(event, context):
  endpoint_url = os.getenv('ENDPOINT_URL')
  try:
    dynamoDB = boto3.resource("dynamodb",endpoint_url=endpoint_url)
    table = dynamoDB.Table("todo")
  
    response = table.put_item(
      Item = {
        "user": event['user'],
        "time": event['time'],
        "title": event['title'],
        "contents": event['contents'] 
      }
    )
    return {
      "statusCode": 200
    }
  except Exception as e:
    print (e)
    return {
        "statusCode": 500
    }    

insert_todo/requirements.txt

boto3

LambdaとAPIGatewayをデプロイするCloudFormation

DynamoにアクセスできるようにAmazonDynamoDBFullAccessを設定しています

template.yaml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
  sam-app

  Sample SAM Template for sam-app

# More info about Globals: https://github.com/awslabs/serverless-application-model/blob/master/docs/globals.rst
Globals:
  Function:
    Timeout: 3

Resources:
  HelloWorldFunction:
    Type: AWS::Serverless::Function # More info about Function Resource: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#awsserverlessfunction
    Properties:
      CodeUri: hello_world/
      Handler: app.lambda_handler
      Runtime: python3.7
      Events:
        HelloWorld:
          Type: Api # More info about API Event Source: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#api
          Properties:
            Path: /hello
            Method: get

  InsertFunction:
    Type: AWS::Serverless::Function # More info about Function Resource: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#awsserverlessfunction
    Properties:
      CodeUri: insert_todo/
      Handler: app.lambda_handler
      Runtime: python3.7
      Policies:
        - AmazonDynamoDBFullAccess
      Events:
        HelloWorld:
          Type: Api # More info about API Event Source: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#api
          Properties:
            Path: /insert
            Method: post

Outputs:
  # ServerlessRestApi is an implicit API created out of Events key under Serverless::Function
  # Find out more about other implicit resources you can reference within SAM
  # https://github.com/awslabs/serverless-application-model/blob/master/docs/internals/generated_resources.rst#api
  HelloWorldApi:
    Description: "API Gateway endpoint URL for Prod stage for Hello World function"
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/hello/"
  HelloWorldFunction:
    Description: "Hello World Lambda Function ARN"
    Value: !GetAtt HelloWorldFunction.Arn
  HelloWorldFunctionIamRole:
    Description: "Implicit IAM Role created for Hello World function"
    Value: !GetAtt HelloWorldFunctionRole.Arn
  InsertApi:
    Description: "API Gateway endpoint URL for Prod stage for Insert function"
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/insert/"
  InsertFunction:
    Description: "Insert Lambda Function ARN"
    Value: !GetAtt InsertFunction.Arn
  InsertFunctionIamRole:
    Description: "Implicit IAM Role created for Insert function"
    Value: !GetAtt InsertFunctionRole.Arn

Lambdaが２つある場合にどうなるのかわかりやすいように、前回のHelloWorldFunctionの設定も残した例です

pytest

テーブルを作成したあとにテスト実行をし、最後にテーブル削除します
テストでは、insertされていることを確認しています

tests/unit/test_insert_todo.py

import json
import boto3
import pytest
import os

from insert_todo import app

@pytest.fixture()
def table():
    print("todo table setup...")
    dynamoDB = boto3.resource("dynamodb",endpoint_url='http://localhost:4566/')
    table = dynamoDB.create_table(
        TableName="todo",
        KeySchema=[
            {
                'AttributeName': 'user',
                'KeyType': 'HASH'  # Partition key
            },
            {
                'AttributeName': 'time',
                'KeyType': 'RANGE'  # Sort key
            }
        ],
        AttributeDefinitions=[
            {
                'AttributeName': 'user',
                'AttributeType': 'S'
            },
            {
                'AttributeName': 'time',
                'AttributeType': 'S'
            },
        ],
        ProvisionedThroughput={
            'ReadCapacityUnits': 1,
            'WriteCapacityUnits': 1
        }
    )

    print("test start...")
    yield table
    
    print("todo table drop...")
    table.delete()


def test_lambda_handler(table):

    event =  {
      "user": "user1",
      "time": "time1",
      "title": "title1",
      "contents": "contents1"
    }

    os.environ['ENDPOINT_URL'] = 'http://localhost:4566/'

    ret = app.lambda_handler(event, "")
    
    assert ret["statusCode"] == 200

    response = table.get_item(
        Key={
            'user': "user1",
            'time': "time1"
        },
    )
    
    assert response["Item"]["user"] == "user1"
    assert response["Item"]["time"] == "time1"
    assert response["Item"]["title"] == "title1"
    assert response["Item"]["contents"] == "contents1"

テスト実行

-sオプションを設定すると標準出力も表示されます

~/environment/sam-app $ python -m pytest -s tests/unit/test_insert_todo.py -v
========================================== test session starts ===========================================
platform linux -- Python 3.7.10, pytest-7.1.2, pluggy-1.0.0 -- /usr/bin/python3
cachedir: .pytest_cache
rootdir: /home/ec2-user/environment/sam-app
plugins: mock-3.7.0
collected 1 item                                                                                         

tests/unit/test_insert_todo.py::test_lambda_handler todo table setup...
test start...
PASSEDtodo table drop...


=========================================== 1 passed in 0.55s ============================================

build

template.yamlで記載したInsertFunctionの名称で個別にbuildできます

~/environment/sam-app $ sam build InsertFunction

deploy

~/environment/sam-app $ sam deploy