如何为使用 boto3 导入的 Cognito 客户端编写单元测试?

how to write unittests for cognito client which is imported using boto3?

这是我要测试的代码块。从逻辑上讲,我认为我想做的是模拟 cognito_client 和 cognito_client.admin_add_user_to_group 所以他们不会 return 任何错误但不明白如何模拟导入 boto3。我也想测试异常,但由于我在尝试中遇到另一个异常,我的测试失败了。

def run_cognito_client(user_pool_id, username, group_name):
    try:
        cognito_client = boto3.client('cognito-idp')
        cognito_client.admin_add_user_to_group(UserPoolId=user_pool_id, Username=username, GroupName=group_name)
    except ClientError as e:
        raise (e.response['Error']['Message'])

2 个解决方案

  1. 手动模拟 botocore.client.BaseClient._make_api_call 这是为 boto3 客户端调用调用的功能。如果您有其他需要模拟的 boto3 调用,您也可以使用它,因为它不仅限于 Cognito.AdminAddUserToGroup,还适用于 S3.ListObjectsV2SecretsManager.GetSecretValue 等其他调用.
    • 见下文test_manual_patch
  2. 使用moto.mock_cognitoidp which already supports所需的功能。
    • 见下文test_lib_patch
import boto3
import botocore
from botocore.exceptions import ClientError
from moto import mock_cognitoidp
import pytest


def run_cognito_client(user_pool_id, username, group_name):
    try:
        cognito_client = boto3.client('cognito-idp')
        response = cognito_client.admin_add_user_to_group(UserPoolId=user_pool_id, Username=username, GroupName=group_name)
    except ClientError as e:
        print(type(e), e)
        # raise (e.response['Error']['Message'])
    else:
        return response


@pytest.fixture
def amend_get_secret_value(mocker):
    orig = botocore.client.BaseClient._make_api_call

    def amend_make_api_call(self, operation_name, kwargs):
        # Intercept boto3 operations for <cognito-idp.admin_add_user_to_group> and return a dummy
        # response. Here, we would only return the dummy response if the kwargs received are the
        # ones we expected. If this doesn't fit your usecase and just want to return the dummy
        # response all the time, just remove that conditional. Actually if you wish to mock all
        # boto3 calls for all AWS services, then just return the dummy response automatically
        # without this checks.
        if operation_name == 'AdminAddUserToGroup' and kwargs == {'UserPoolId': 'a', 'Username': 'b', 'GroupName': 'c'}:
            return {
                'dummy': 'response from my mock'
            }

        return orig(self, operation_name, kwargs)

    mocker.patch('botocore.client.BaseClient._make_api_call', new=amend_make_api_call)


def test_manual_patch(amend_get_secret_value):
    response = run_cognito_client("a", "b", "c")
    print(f"{response=}")


@mock_cognitoidp
def test_lib_patch():
    cognito_client = boto3.client('cognito-idp')

    pool = cognito_client.create_user_pool(PoolName='SolarSystem')['UserPool']
    print(f"{pool=}")

    group = cognito_client.create_group(
        GroupName='Earth',
        UserPoolId=pool['Id'],
    )
    print(f"{group=}")

    user = cognito_client.admin_create_user(
        UserPoolId=pool['Id'],
        Username='chopin',
    )
    print(f"{user=}")

    response = run_cognito_client(pool['Id'], user['User']['Username'], group['Group']['GroupName'])
    print(f"{response=}")
$ pytest -q -rP
================================================================================================= PASSES ==================================================================================================
____________________________________________________________________________________________ test_manual_patch ____________________________________________________________________________________________
------------------------------------------------------------------------------------------ Captured stdout call -------------------------------------------------------------------------------------------
response={'dummy': 'response from my mock'}
_____________________________________________________________________________________________ test_lib_patch ______________________________________________________________________________________________
------------------------------------------------------------------------------------------ Captured stdout call -------------------------------------------------------------------------------------------
pool={'Id': 'ap-southeast-1_e843cd99604648e9bda9b2950eb15752', 'Name': 'SolarSystem', 'LastModifiedDate': datetime.datetime(2021, 8, 27, 8, 31, 10, tzinfo=tzlocal()), 'CreationDate': datetime.datetime(2021, 8, 27, 8, 31, 10, tzinfo=tzlocal()), 'MfaConfiguration': 'OFF', 'Arn': 'arn:aws:cognito-idp:ap-southeast-1:123456789012:userpool/ap-southeast-1_e843cd99604648e9bda9b2950eb15752'}
group={'Group': {'GroupName': 'Earth', 'UserPoolId': 'ap-southeast-1_e843cd99604648e9bda9b2950eb15752', 'Description': '', 'LastModifiedDate': datetime.datetime(2021, 8, 27, 16, 31, 10, tzinfo=tzlocal()), 'CreationDate': datetime.datetime(2021, 8, 27, 16, 31, 10, tzinfo=tzlocal())}, 'ResponseMetadata': {'HTTPStatusCode': 200, 'HTTPHeaders': {'server': 'amazon.com'}, 'RetryAttempts': 0}}
user={'User': {'Username': 'chopin', 'Attributes': [], 'UserCreateDate': datetime.datetime(2021, 8, 27, 8, 31, 10, tzinfo=tzlocal()), 'UserLastModifiedDate': datetime.datetime(2021, 8, 27, 8, 31, 10, tzinfo=tzlocal()), 'Enabled': True, 'UserStatus': 'FORCE_CHANGE_PASSWORD', 'MFAOptions': []}, 'ResponseMetadata': {'HTTPStatusCode': 200, 'HTTPHeaders': {'server': 'amazon.com'}, 'RetryAttempts': 0}}
response={'ResponseMetadata': {'HTTPStatusCode': 200, 'HTTPHeaders': {'server': 'amazon.com'}, 'RetryAttempts': 0}}
2 passed in 0.96s