0

我们正在开发一个使用 dynamoDB 作为数据库的 Flask REST API。我想为同一个应用程序编写单元测试代码。我正在使用 moto 来模拟 dynamoDB。这是 conftest.py 的代码

import os
import boto3
import pytest
from moto import mock_dynamodb2


@pytest.fixture(scope='module')
def aws_credentials():
    # Mocked AWS credentials for moto
    os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
    os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
    os.environ['AWS_SECURITY_TOKEN'] = 'testing'
    os.environ['AWS_SESSION_TOKEN'] = 'testing'


@pytest.yield_fixture(scope='module')
def dynamodb_resource(aws_credentials):
    # DDB mock resource
    with mock_dynamodb2():
        conn = boto3.resource('dynamodb', region_name='ap-south-1')
        yield conn

我还可以在这个模拟数据库中插入、删除和创建表。

创建表：

@contextmanager
def ddb_setup(client_resource):
    # create mock ddb bucket and object to be available to all methods in the test class
    client_resource.create_table(
        TableName=TableName,
        KeySchema=[
            {
                'AttributeName': 'user_name',
                'KeyType': 'HASH'
            }, {
                'AttributeName': 'last_name',
                'KeyType': 'RANGE'
            }
        ],
        AttributeDefinitions=[
            {
                'AttributeName': 'user_name',
                'AttributeType': 'S'
            }, {
                'AttributeName': 'last_name',
                'AttributeType': 'S'
            }
        ],
        ProvisionedThroughput={
            'ReadCapacityUnits': 5,
            'WriteCapacityUnits': 5
        }
    )
    yield

测试创建表并向其插入数据（test_user.py）

class TestClassDDB:

    def test_create_table(self, dynamodb_resource, dynamodb_client):
        # test the successfulcreation of a table
        with ddb_setup(dynamodb_resource):
            try:
                response = dynamodb_client.describe_table(TableName=TableName)
                assert response['Table']['TableName'] == TableName
            except ClientError as err:
                assert err.response['Error']['Code'] == 'ResourceNotFoundException'

    def test_put_object(self, dynamodb_resource):
        # test the successful adding of an object to a table

        table = dynamodb_resource.Table(TableName)
        table.put_item(
            Item={
                'user_name': 'WillSmith',
                'last_name': 'Smith',
                'account_type': 'standard_user'
            }
        )
        results = table.get_item(
            Key={
                'user_name': 'WillSmith',
                'last_name': 'Smith'
            }
        )
        assert results['Item']['user_name'] == 'WillSmith'

上面的代码运行没有任何错误。

我的问题是，目前烧瓶应用程序正在使用实时 dynamodb。为了测试应用程序，我应该使用模拟 dynamodb，那么如何将这个模拟 dynamodb 指向烧瓶应用程序。这是我编写的烧瓶应用程序的示例代码（app.py）。

import os
import boto3    
from flask import Flask, jsonify, request

app = Flask(__name__)    
USERS_TABLE = os.environ['USERS_TABLE']
client = boto3.client('dynamodb')

@app.route("/getUsers/<string:user_id>")
def get_user(user_id):
    try:
        resp = client.get_item(
            TableName=USERS_TABLE,
            Key={
                'userId': {'S': user_id}
            }
        )
        item = resp.get('Item')
        if not item:
            return jsonify({'error': 'User does not exist'}), 404
        if item.get('status').get('N') == '-1':
            return jsonify({'error': 'user is not active'}), 404

        return jsonify({
            'userId': item.get('userId').get('S'),
            'name': item.get('name').get('S'),
            'message': True
        })
    except Exception as e:
        print(e)
        traceback.print_exc()
        return jsonify({
            "status": False,
            "message": "some internal error occurred!"
        })
4

0 回答 0