[ DynamoDB ] mock 사용해보기

반응형

DynamoDB 관련 테스트를 할 때는 moto에서 제공해주는 mock을 사용할 수 있습니다

 

오늘은 local환경에서 mock을 이용해서 테스트를 해보겠습니다.

 

제 환경은 이렇게 됩니다.

python 3.7
botocore == 1.31.1
moto == 3.1.6

import unittest
import boto3
from moto import mock_dynamodb2

class TestDynamo(unittest.TestCase):

    def setUp(self):
        pass

    @mock_dynamodb2
    def test_recoverBsaleAssociation(self):

        table_name = 'test'
        dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1')
    
        table = dynamodb.create_table(
            TableName=table_name,
            KeySchema=[
                {
                    'AttributeName': 'key',
                    'KeyType': 'HASH'
                },
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'key',
                    'AttributeType': 'S'
                },

            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 5,
                'WriteCapacityUnits': 5
            }
        )

        item = {}
        item['key'] = 'value'

        table.put_item(Item=item)

        table = dynamodb.Table(table_name)
        response = table.get_item(
            Key={
                'key': 'value'
            }
        )
        if 'Item' in response:
            item = response['Item']

        self.assertTrue("key" in item)
        self.assertEquals(item["key"], "value")

        print("success")

테스트 코드입니다.

 

실행시켜보면

 

성공한 걸 알 수 있습니다.

 

그럼 이제 실제 코드를 mock처럼 사용해보겠습니다.

 

폴더 구조는 이렇습니다

# dynamodb.py

import boto3

dynamodb = None

def init_dynamodb():
  global dynamodb
  if dynamodb is None:
    dynamodb = boto3.resource('dynamodb', aws_access_key_id="test", aws_secret_access_key="test", region_name='test')

init_dynamodb()

def res_data():

    table = dynamodb.Table('test')

    response = table.get_item(
        Key={
            'key': 'value'
        }
    )

    return response

아마도 dynamodb 세션을 글로벌로 사용하는 로직이 많을 것 같습니다.

 

# test_mock_dynamodb.py

import unittest
import boto3
from moto import mock_dynamodb2

from dynamodb import dynamodb as db

class TestDynamo(unittest.TestCase):

    def setUp(self):
        pass


    @mock_dynamodb2
    def test_recoverBsaleAssociation(self):

        table_name = 'test'
        dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1')
    
        table = dynamodb.create_table(
            TableName=table_name,
            KeySchema=[
                {
                    'AttributeName': 'key',
                    'KeyType': 'HASH'
                },
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'key',
                    'AttributeType': 'S'
                },

            ],
            ProvisionedThroughput={
                'ReadCapacityUnits': 5,
                'WriteCapacityUnits': 5
            }
        )

        item = {}
        item['key'] = 'value'

        table.put_item(Item=item)

        db.dynamodb = dynamodb

        response = db.res_data()

        # table = dynamodb.Table(table_name)
        # response = table.get_item(
        #     Key={
        #         'key': 'value'
        #     }
        # )

        if 'Item' in response:
            item = response['Item']

        self.assertTrue("key" in item)
        self.assertEquals(item["key"], "value")

기존에 있던 취득 부분은 주석 처리하고 실제 코드에서 취득하도록 바꿔줬습니다.

또 실제 코드에 있는 글로벌 변수에 미리 dynamodb의 세션을 지정해서 취득이 안되도록 설정해줬습니다.

 

이제 테스트를 실행해보면 문제없이 실행되는 걸 확인할 수 있습니다.