Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use cognito_idp_client method in localstack

Best Python code snippet using localstack_python

CognitoUserPoolResourceHandler.py

Source:CognitoUserPoolResourceHandler.py

1#2# All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or3# its licensors.4#5# For complete copyright and license terms please see the LICENSE at the root of this6# distribution (the "License"). All use of this software is governed by the License,7# or, if provided, by the license below or the license accompanying this file. Do not8# remove or modify any license notices. This file is distributed on an "AS IS" BASIS,9# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.10#11# $Revision: #5 $12import boto313import botocore14import custom_resource_response15from resource_manager_common import aws_utils16import properties17from resource_manager_common import stack_info18import user_pool19import identity_pool20import json21def handler(event, context):22    props = properties.load(event, {23            'ClientApps': properties.StringOrListOfString(),24            'ExplicitAuthFlows': properties.StringOrListOfString(default=[]),25            'RefreshTokenValidity': properties.String('30'),26            'ConfigurationKey': properties.String(), ##this is only here to force the resource handler to execute on each update to the deployment27            'LambdaConfig': properties.Dictionary({}),28            'PoolName': properties.String(),29            'Groups': properties.ObjectOrListOfObject(default=[],30                                                      schema={31                                                          'Name': properties.String(),32                                                          'Description': properties.String(''),33                                                          'Role': properties.String(),34                                                          'Precedence': properties.String('99')35                                                      }),36            'AllowAdminCreateUserOnly': properties.String('')37        })38    #give the identity pool a unique name per stack39    stack_name = aws_utils.get_stack_name_from_stack_arn(event['StackId'])40    stack_name = stack_name.replace('-', ' ') # Prepare stack_name to be used by _associate_user_pool_with_player_access41    pool_name = props.PoolName.replace('-', ' ')42    pool_name = stack_name+pool_name43    cognito_idp_client = user_pool.get_idp_client()44    pool_id = event.get('PhysicalResourceId')45    found_pool = user_pool.get_user_pool(pool_id)46    request_type = event['RequestType']47    if request_type == 'Delete':48        if found_pool != None:49            cognito_idp_client.delete_user_pool(UserPoolId = pool_id) 50        data = {}51    else:52        #if the pool exists just update it, otherwise create a new one53        54        mfaConfig = 'OFF'   # MFA is currently unsupported by Lumberyard55        # Users are automatically prompted to verify these things. 56        # At least one auto-verified thing (email or phone) is required to allow password recovery.57        auto_verified_attributes = [ 'email' ]58        59        client_app_data = {};60        lambda_config = props.LambdaConfig61        user_pool.validate_identity_metadata(event['StackId'], event['LogicalResourceId'], props.ClientApps)62        admin_create_user_config = __get_admin_create_user_config(props.AllowAdminCreateUserOnly)63        print json.dumps(admin_create_user_config)64        if found_pool != None:  # Update65            response = cognito_idp_client.update_user_pool(66                    UserPoolId=pool_id,67                    MfaConfiguration=mfaConfig,68                    AutoVerifiedAttributes=auto_verified_attributes,69                    LambdaConfig=lambda_config,70                    AdminCreateUserConfig=admin_create_user_config71                )72            existing_client_apps = user_pool.get_client_apps(pool_id)73            client_app_data = update_client_apps(pool_id, props.ClientApps, existing_client_apps, False, props.ExplicitAuthFlows, props.RefreshTokenValidity)74            response = cognito_idp_client.list_groups(75                UserPoolId=pool_id76            )77            found_groups = {}78            for actual_group in response['Groups']:79                group_name = actual_group['GroupName']80                for requested_group in props.Groups:81                    # does the group exist in the resource template82                    if group_name == requested_group.Name:83                        found_groups.update({group_name: True})84                        break85                #delete the group as it is no longer in the resource template86                if group_name not in found_groups:87                    cognito_idp_client.delete_group(88                        GroupName=actual_group['GroupName'],89                        UserPoolId=pool_id90                    )91            print "Found groups=>", json.dumps(found_groups)92            #iterate the groups defined in the user pool resource template93            for group in props.Groups:94                #update the group as it is currently a group in the user pool95                group_definition = __generate_group_defintion(pool_id, group)96                print "Group '{}' is defined by {}".format(group.Name, json.dumps(group_definition))97                if group.Name in found_groups:98                    cognito_idp_client.update_group(**group_definition)99                else:100                    #group is a new group on the user pool101                    cognito_idp_client.create_group(**group_definition)102        else: # Create103            response = cognito_idp_client.create_user_pool(104                    PoolName=pool_name,105                    MfaConfiguration=mfaConfig,106                    AutoVerifiedAttributes=auto_verified_attributes,107                    LambdaConfig=lambda_config,108                    AdminCreateUserConfig=admin_create_user_config109                )110            pool_id = response['UserPool']['Id']111            print 'User pool creation response: ', response112            for group in props.Groups:113                group_definition = __generate_group_defintion(pool_id, group)114                print "Group '{}' is defined by {}".format(group.Name, json.dumps(group_definition))115                cognito_idp_client.create_group(**group_definition)116            client_app_data = update_client_apps(pool_id, props.ClientApps, [], False, props.ExplicitAuthFlows, props.RefreshTokenValidity)117        updated_resources = {118            event['StackId']: {119                event['LogicalResourceId']: {120                    'physical_id': pool_id,121                    'client_apps': {client_app['ClientName']: {'client_id': client_app['ClientId']} for client_app in client_app_data['Created'] + client_app_data['Updated']}122                }123            }124        }125        identity_pool.update_cognito_identity_providers(event['StackId'], pool_id, updated_resources)126        data = {127            'UserPoolName': pool_name,128            'UserPoolId': pool_id,129            'ClientApps': client_app_data,130        }131    132    physical_resource_id = pool_id133    custom_resource_response.succeed(event, context, data, physical_resource_id)134def update_client_apps(user_pool_id, new_client_name_list, existing_clients, should_generate_secret, explicitauthflows, refreshtokenvalidity):135    client_changes = {'Created':[], 'Deleted':[], 'Updated':[]}136    new_client_names = set(new_client_name_list)137    # Update or delete existing clients.138    for existing_client in existing_clients:139        client_id = existing_client['ClientId']140        client_name = existing_client['ClientName']141        if client_name in new_client_names:142            updated = user_pool.get_idp_client().update_user_pool_client(143                UserPoolId=user_pool_id,144                ClientId=client_id,145                ExplicitAuthFlows=explicitauthflows,146                RefreshTokenValidity=int(refreshtokenvalidity)147            )148            client_changes['Updated'].append({149                'UserPoolId': user_pool_id,150                'ClientName': updated['UserPoolClient']['ClientName'],151                'ClientId': client_id,152                'ExplicitAuthFlows': explicitauthflows,153                'RefreshTokenValidity': int(refreshtokenvalidity)154            })155        else:156            deleted = user_pool.get_idp_client().delete_user_pool_client(UserPoolId=user_pool_id, ClientId=client_id)157            client_changes['Deleted'].append({158                'UserPoolId': user_pool_id,159                'ClientName': client_name,160                'ClientId': client_id161            })162    # Create new clients163    existing_client_names = {client['ClientName'] for client in existing_clients}164    for client_name in new_client_names.difference(existing_client_names):165        created = user_pool.get_idp_client().create_user_pool_client(166            UserPoolId=user_pool_id,167            ClientName=client_name,168            GenerateSecret=should_generate_secret,169            ExplicitAuthFlows=explicitauthflows,170            RefreshTokenValidity=int(refreshtokenvalidity)171        )172        client_data = {173            'UserPoolId': user_pool_id,174            'ClientName': created['UserPoolClient']['ClientName'],175            'ClientId': created['UserPoolClient']['ClientId'],176            'ExplicitAuthFlows': explicitauthflows,177            'RefreshTokenValidity': int(refreshtokenvalidity)178        }179        if 'ClientSecret' in created['UserPoolClient']:180            client_data['ClientSecret'] = created['UserPoolClient']['ClientSecret']181        client_changes['Created'].append(client_data)182    return client_changes183def __generate_group_defintion(user_pool_id, props):184    return {185        'GroupName': props.Name,186        'UserPoolId': user_pool_id,187        'RoleArn': props.Role,188        'Description': props.Description,189        'Precedence': int(props.Precedence)190    }191def __get_admin_create_user_config(allow_admin_create_user_only):192    admin_create_user_only = allow_admin_create_user_only.lower() == 'true'193    return {194        'AllowAdminCreateUserOnly': admin_create_user_only...

conftest.py

Source:conftest.py

...17    environ["AWS_SECRET_ACCESS_KEY"] = "testing"18    environ["AWS_SECURITY_TOKEN"] = "testing"19    environ["AWS_DEFAULT_REGION"] = "us-east-1"20@pytest.fixture(scope="session")21def cognito_idp_client(aws_credentials, region_name: str = "us-east-1") -> BaseClient:22    with mock_cognitoidp():23        yield boto3.client("cognito-idp", region_name=region_name)24@pytest.fixture(scope="session")25def pool_name() -> str:26    return "fake-cognito-user-pool"27@pytest.fixture(scope="session")28def user_pool_id(cognito_idp_client: BaseClient, pool_name: str) -> str:29    response = cognito_idp_client.create_user_pool(PoolName=pool_name, UsernameAttributes=["email"])30    return response["UserPool"]["Id"]31@pytest.fixture(scope="session")32def app_client_id(cognito_idp_client: BaseClient, user_pool_id: str, pool_name: str) -> str:33    return cognito_idp_client.create_user_pool_client(34        UserPoolId=user_pool_id,35        ClientName=pool_name,...

cognito_user_list.py

Source:cognito_user_list.py

1# Function to get a given set of attributes for all cognito users2import boto33import datetime4import json5import pandas as pd6REGION = 'us-west-2'7USER_POOL_ID = 'specify_user_pool_id_here'8client = boto3.client('cognito-idp', REGION)9pagination_token = None10page_count = 111users_all = []12def get_cognito_users_list(cognito_idp_client, next_pagination_token = None):13    if next_pagination_token:14        return cognito_idp_client.list_users(15            UserPoolId = USER_POOL_ID,16            PaginationToken = next_pagination_token17        )18    else:19        return cognito_idp_client.list_users(20            UserPoolId = USER_POOL_ID)21while page_count ==1 or pagination_token is not None:22    user_records = get_cognito_users_list(23        cognito_idp_client = client,24        next_pagination_token = pagination_token)25    if "PaginationToken" in user_records.keys():26        pagination_token = user_records["PaginationToken"]27    else:28        pagination_token = None29    page_count += 130    users = user_records['Users']31    users_all += users32df = pd.json_normalize(users_all, 'Attributes', ['Username','UserCreateDate','UserLastModifiedDate','Enabled','UserStatus'])33df1 = df[df.Name.isin(["custom:UserID","given_name","family_name","email"])]34df2 = df1.pivot(index='Username', columns='Name', values='Value')35df2.reset_index(inplace=True)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.