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

How to use acm_client method in localstack

Best Python code snippet using localstack_python

lambda_function.py

Source:lambda_function.py

1import json2import logging3import boto34import cfnresponse5import time6import re7acm_client = boto3.client('acm')8r53_client = boto3.client('route53')9lambda_client = boto3.client('lambda')10logs_client = boto3.client('logs')11def handler(event, context):12    print('Received event: %s' % json.dumps(event))13    status = cfnresponse.SUCCESS14    physical_resource_id = None15    data = {}16    reason = None17    try:18        if event['RequestType'] == 'Create':19            token = ''.join(ch for ch in str(event['StackId'] + event['LogicalResourceId']) if ch.isalnum())20            token = token[len(token)-32:]21            if len(event['ResourceProperties']['HostNames']) > 1:22                arn = acm_client.request_certificate(23                    ValidationMethod='DNS',24                    DomainName=event['ResourceProperties']['HostNames'][0],25                    SubjectAlternativeNames=event['ResourceProperties']['HostNames'][1:],26                    IdempotencyToken=token27                )['CertificateArn']28            else:29                arn = acm_client.request_certificate(30                    ValidationMethod='DNS',31                    DomainName=event['ResourceProperties']['HostNames'][0],32                    IdempotencyToken=token33                )['CertificateArn']34            physical_resource_id = arn35            logging.info("certificate arn: %s" % arn)36            rs = {}37            while True:38                try:39                    for d in acm_client.describe_certificate(CertificateArn=arn)['Certificate']['DomainValidationOptions']:40                        rs[d['ResourceRecord']['Name']] = d['ResourceRecord']['Value']41                    break42                except KeyError:43                    if (context.get_remaining_time_in_millis() / 1000.00) > 20.0:44                        print('waiting for ResourceRecord to be available')45                        time.sleep(15)46                    else:47                        logging.error('timed out waiting for ResourceRecord')48                        status = cfnresponse.FAILED49                    time.sleep(15)50            rs = [{'Action': 'CREATE', 'ResourceRecordSet': {'Name': r, 'Type': 'CNAME', 'TTL': 600,'ResourceRecords': [{'Value': rs[r]}]}} for r in rs.keys()]51            try:52                r53_client.change_resource_record_sets(HostedZoneId=event['ResourceProperties']['HostedZoneId'], ChangeBatch={'Changes': rs})53            except Exception as e:54                if not str(e).endswith('but it already exists'):55                    raise56            while 'PENDING_VALIDATION' in [v['ValidationStatus'] for v in acm_client.describe_certificate(CertificateArn=arn)['Certificate']['DomainValidationOptions']]:57                print('waiting for validation to complete')58                if (context.get_remaining_time_in_millis() / 1000.00) > 20.0:59                    time.sleep(15)60                else:61                    logging.error('validation timed out')62                    status = cfnresponse.FAILED63            for r in [v for v in acm_client.describe_certificate(CertificateArn=arn)['Certificate']['DomainValidationOptions']]:64                if r['ValidationStatus'] != 'SUCCESS':65                    logging.debug(r)66                    status = cfnresponse.FAILED67                    reason = 'One or more domains failed to validate'68                    logging.error(reason)69            data['Arn'] = arn70            # delay as long as possible to give the cert a chance to propogate71            while context.get_remaining_time_in_millis() / 1000.00 > 10.0:72                time.sleep(5)73        elif event['RequestType'] == 'Update':74            reason = 'Exception: Stack updates are not supported'75            logging.error(reason)76            status = cfnresponse.FAILED77            physical_resource_id = event['PhysicalResourceId']78        elif event['RequestType'] == 'Delete':79            physical_resource_id=event['PhysicalResourceId']80            if not re.match(r'arn:[\w+=/,.@-]+:[\w+=/,.@-]+:[\w+=/,.@-]*:[0-9]+:[\w+=,.@-]+(/[\w+=,.@-]+)*', physical_resource_id):81                logging.info("PhysicalId is not an acm arn, assuming creation never happened and skipping delete")82            else:83                rs={}84                for d in acm_client.describe_certificate(CertificateArn=physical_resource_id)['Certificate']['DomainValidationOptions']:85                    rs[d['ResourceRecord']['Name']] = d['ResourceRecord']['Value']86                rs = [{'Action': 'DELETE', 'ResourceRecordSet': {'Name': r, 'Type': 'CNAME', 'TTL': 600,'ResourceRecords': [{'Value': rs[r]}]}} for r in rs.keys()]87                try:88                    r53_client.change_resource_record_sets(HostedZoneId=event['ResourceProperties']['HostedZoneId'], ChangeBatch={'Changes': rs})89                except r53_client.exceptions.InvalidChangeBatch as e:90                    pass91                time.sleep(30)92                try:93                    acm_client.delete_certificate(CertificateArn=physical_resource_id)94                except acm_client.exceptions.ResourceInUseException as e:95                    time.sleep(60)96                    acm_client.delete_certificate(CertificateArn=physical_resource_id)97    except Exception as e:98        logging.error('Exception: %s' % e, exc_info=True)99        reason = str(e)100        status = cfnresponse.FAILED101    finally:102        if event['RequestType'] == 'Delete':103            try:104                wait_message = 'waiting for events for request_id %s to propagate to cloudwatch...' % context.aws_request_id105                while not logs_client.filter_log_events(106                        logGroupName=context.log_group_name,107                        logStreamNames=[context.log_stream_name],108                        filterPattern='"%s"' % wait_message109                )['events']:110                    print(wait_message)111                    time.sleep(5)112            except Exception as e:113                logging.error('Exception: %s' % e, exc_info=True)114                time.sleep(120)...

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.