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

user_16699373 0
Viewed 615

This is the chunk of code I am trying to test. Logically I think what I want to do is mock cognito_client and cognito_client.admin_add_user_to_group so they do not return any errors but do not understand how to mock the import boto3. I would also like to test the exception but as I am getting another exception in the try, my test fails.

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'])
python-unittest python
1 Answers

2 solutions

  1. Manually mock botocore.client.BaseClient._make_api_call which is the functionality called for boto3 client calls. Should you have other boto3 calls that needs to be mocked, you can use it too as it isn't limited to Cognito.AdminAddUserToGroup but to others as well such as S3.ListObjectsV2, SecretsManager.GetSecretValue, and others.
    • See test_manual_patch below
  2. Use moto.mock_cognitoidp which already supports the needed functionalities.
    • See test_lib_patch below
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
user_110438250
Related