Unlock 30% off on Manual Testing Annual Plans this Holiday Season. 
Use Code: LCMYEARENDEXCL30
Use Code: LCMYEARENDEXCL30Check icon
Copied to Clipboard!
How to use resolve_refs_recursively method in localstack
Best Python code snippet using localstack_python
template_deployer.py
Source:template_deployer.py 
...197 topics = aws_stack.connect_to_service('sns').list_topics()['Topics']198 topic_arns = [t['TopicArn'] for t in topics if t['TopicArn'].endswith(':%s' % topic_name)]199 return topic_arns[0]200def get_role_arn(role_arn, **kwargs):201 role_arn = resolve_refs_recursively(kwargs.get('stack_name'), role_arn, kwargs.get('resources'))202 return aws_stack.role_arn(role_arn)203# ----------------204# CF TEMPLATE HANDLING205# ----------------206def parse_template(template):207 try:208 return json.loads(template)209 except Exception:210 return yaml.safe_load(template)211def template_to_json(template):212 template = parse_template(template)213 return json.dumps(template)214def get_resource_type(resource):215 res_type = resource.get('ResourceType') or resource.get('Type') or ''216 parts = res_type.split('::', 1)217 if len(parts) == 1:218 return None219 return parts[1]220def get_service_name(resource):221 parts = resource.get('Type', '').split('::')222 if len(parts) == 1:223 return None224 return parts[1].lower()225def get_resource_name(resource):226 res_type = get_resource_type(resource)227 properties = resource.get('Properties') or {}228 name = properties.get('Name')229 if name:230 return name231 # try to extract name from attributes232 if res_type == 'S3::Bucket':233 name = properties.get('BucketName')234 elif res_type == 'SQS::Queue':235 name = properties.get('QueueName')236 else:237 LOG.warning('Unable to extract name for resource type "%s"' % res_type)238 return name239def get_client(resource):240 resource_type = get_resource_type(resource)241 service = get_service_name(resource)242 resource_config = RESOURCE_TO_FUNCTION.get(resource_type)243 if resource_config is None:244 raise Exception('CloudFormation deployment for resource type %s not yet implemented' % resource_type)245 if ACTION_CREATE not in resource_config:246 # nothing to do for this resource247 return248 try:249 if resource_config[ACTION_CREATE].get('boto_client') == 'resource':250 return aws_stack.connect_to_resource(service)251 return aws_stack.connect_to_service(service)252 except Exception as e:253 LOG.warning('Unable to get client for "%s" API, skipping deployment: %s' % (service, e))254 return None255def describe_stack_resource(stack_name, logical_resource_id):256 client = aws_stack.connect_to_service('cloudformation')257 result = client.describe_stack_resource(StackName=stack_name, LogicalResourceId=logical_resource_id)258 return result['StackResourceDetail']259def retrieve_resource_details(resource_id, resource_status, resources, stack_name):260 resource = resources.get(resource_id)261 resource_id = resource_status.get('PhysicalResourceId') or resource_id262 if not resource:263 resource = {}264 resource_type = get_resource_type(resource)265 resource_props = resource.get('Properties')266 try:267 if resource_type == 'Lambda::Function':268 resource_id = resource_props['FunctionName'] if resource else resource_id269 return aws_stack.connect_to_service('lambda').get_function(FunctionName=resource_id)270 elif resource_type == 'Lambda::Version':271 name = resource_props['FunctionName']272 func_name = aws_stack.lambda_function_name(name)273 func_version = name.split(':')[7] if len(name.split(':')) > 7 else '$LATEST'274 versions = aws_stack.connect_to_service('lambda').list_versions_by_function(FunctionName=func_name)275 return ([v for v in versions['Versions'] if v['Version'] == func_version] or [None])[0]276 elif resource_type == 'Lambda::EventSourceMapping':277 resource_id = resource_props['FunctionName'] if resource else resource_id278 source_arn = resource_props.get('EventSourceArn')279 resource_id = resolve_refs_recursively(stack_name, resource_id, resources)280 source_arn = resolve_refs_recursively(stack_name, source_arn, resources)281 if not resource_id or not source_arn:282 raise Exception('ResourceNotFound')283 mappings = aws_stack.connect_to_service('lambda').list_event_source_mappings(284 FunctionName=resource_id, EventSourceArn=source_arn)285 mapping = list(filter(lambda m:286 m['EventSourceArn'] == source_arn and m['FunctionArn'] == aws_stack.lambda_function_arn(resource_id),287 mappings['EventSourceMappings']))288 if not mapping:289 raise Exception('ResourceNotFound')290 return mapping[0]291 elif resource_type == 'DynamoDB::Table':292 resource_id = resource_props['TableName'] if resource else resource_id293 return aws_stack.connect_to_service('dynamodb').describe_table(TableName=resource_id)294 elif resource_type == 'ApiGateway::RestApi':295 apis = aws_stack.connect_to_service('apigateway').get_rest_apis()['items']296 resource_id = resource_props['Name'] if resource else resource_id297 result = list(filter(lambda api: api['name'] == resource_id, apis))298 return result[0] if result else None299 elif resource_type == 'ApiGateway::Resource':300 api_id = resource_props['RestApiId'] if resource else resource_id301 api_id = resolve_refs_recursively(stack_name, api_id, resources)302 parent_id = resolve_refs_recursively(stack_name, resource_props['ParentId'], resources)303 if not api_id or not parent_id:304 return None305 api_resources = aws_stack.connect_to_service('apigateway').get_resources(restApiId=api_id)['items']306 target_resource = list(filter(lambda res:307 res.get('parentId') == parent_id and res['pathPart'] == resource_props['PathPart'], api_resources))308 if not target_resource:309 return None310 path = aws_stack.get_apigateway_path_for_resource(api_id,311 target_resource[0]['id'], resources=api_resources)312 result = list(filter(lambda res: res['path'] == path, api_resources))313 return result[0] if result else None314 elif resource_type == 'ApiGateway::Deployment':315 api_id = resource_props['RestApiId'] if resource else resource_id316 api_id = resolve_refs_recursively(stack_name, api_id, resources)317 if not api_id:318 return None319 result = aws_stack.connect_to_service('apigateway').get_deployments(restApiId=api_id)['items']320 # TODO possibly filter results by stage name or other criteria321 return result[0] if result else None322 elif resource_type == 'ApiGateway::Method':323 api_id = resolve_refs_recursively(stack_name, resource_props['RestApiId'], resources)324 res_id = resolve_refs_recursively(stack_name, resource_props['ResourceId'], resources)325 if not api_id or not res_id:326 return None327 res_obj = aws_stack.connect_to_service('apigateway').get_resource(restApiId=api_id, resourceId=res_id)328 match = [v for (k, v) in res_obj['resourceMethods'].items()329 if resource_props['HttpMethod'] in (v.get('httpMethod'), k)]330 return match or None331 elif resource_type == 'SQS::Queue':332 sqs_client = aws_stack.connect_to_service('sqs')333 queues = sqs_client.list_queues()334 result = list(filter(lambda item:335 # TODO possibly find a better way to compare resource_id with queue URLs336 item.endswith('/%s' % resource_id), queues.get('QueueUrls', [])))337 if not result:338 return None339 result = sqs_client.get_queue_attributes(QueueUrl=result[0], AttributeNames=['All'])['Attributes']340 result['Arn'] = result['QueueArn']341 return result342 elif resource_type == 'SNS::Topic':343 topics = aws_stack.connect_to_service('sns').list_topics()344 result = list(filter(lambda item: item['TopicArn'] == resource_id, topics.get('Topics', [])))345 return result[0] if result else None346 elif resource_type == 'S3::Bucket':347 bucket_name = resource_props.get('BucketName') or resource_id348 return aws_stack.connect_to_service('s3').get_bucket_location(Bucket=bucket_name)349 elif resource_type == 'Logs::LogGroup':350 # TODO implement351 raise Exception('ResourceNotFound')352 elif resource_type == 'Kinesis::Stream':353 stream_name = resolve_refs_recursively(stack_name, resource_props['Name'], resources)354 result = aws_stack.connect_to_service('kinesis').describe_stream(StreamName=stream_name)355 return result356 elif resource_type == 'StepFunctions::StateMachine':357 sm_name = resource_props.get('StateMachineName') or resource_id358 sm_name = resolve_refs_recursively(stack_name, sm_name, resources)359 sfn_client = aws_stack.connect_to_service('stepfunctions')360 state_machines = sfn_client.list_state_machines()['stateMachines']361 sm_arn = [m['stateMachineArn'] for m in state_machines if m['name'] == sm_name]362 if not sm_arn:363 return None364 result = sfn_client.describe_state_machine(stateMachineArn=sm_arn[0])365 return result366 if is_deployable_resource(resource):367 LOG.warning('Unexpected resource type %s when resolving references of resource %s: %s' %368 (resource_type, resource_id, resource))369 except Exception as e:370 # we expect this to be a "not found" exception371 markers = ['NoSuchBucket', 'ResourceNotFound', '404']372 if not list(filter(lambda marker, e=e: marker in str(e), markers)):373 LOG.warning('Unexpected error retrieving details for resource %s: %s %s - %s %s' %374 (resource_type, e, traceback.format_exc(), resource, resource_status))375 return None376def extract_resource_attribute(resource_type, resource, attribute):377 LOG.debug('Extract resource attribute: %s %s' % (resource_type, attribute))378 # extract resource specific attributes379 if resource_type == 'Lambda::Function':380 actual_attribute = 'FunctionArn' if attribute == 'Arn' else attribute381 return resource['Configuration'][actual_attribute]382 elif resource_type == 'DynamoDB::Table':383 actual_attribute = 'LatestStreamArn' if attribute == 'StreamArn' else attribute384 value = resource['Table'].get(actual_attribute)385 return value386 elif resource_type == 'ApiGateway::RestApi':387 if attribute == 'PhysicalResourceId':388 return resource['id']389 if attribute == 'RootResourceId':390 resources = aws_stack.connect_to_service('apigateway').get_resources(restApiId=resource['id'])['items']391 for res in resources:392 if res['path'] == '/' and not res.get('parentId'):393 return res['id']394 elif resource_type == 'ApiGateway::Resource':395 if attribute == 'PhysicalResourceId':396 return resource['id']397 return resource.get(attribute)398def resolve_ref(stack_name, ref, resources, attribute):399 if ref == 'AWS::Region':400 return DEFAULT_REGION401 resource_status = {}402 if stack_name:403 resource_status = describe_stack_resource(stack_name, ref)404 attr_value = resource_status.get(attribute)405 if attr_value not in [None, '']:406 return attr_value407 elif ref in resources:408 resource_status = resources[ref]['__details__']409 # fetch resource details410 resource = resources.get(ref)411 resource_new = retrieve_resource_details(ref, resource_status, resources, stack_name)412 if not resource_new:413 return414 resource_type = get_resource_type(resource)415 result = extract_resource_attribute(resource_type, resource_new, attribute)416 if not result:417 LOG.warning('Unable to extract reference attribute %s from resource: %s' % (attribute, resource_new))418 return result419def resolve_refs_recursively(stack_name, value, resources):420 if isinstance(value, dict):421 if len(value) == 1 and 'Ref' in value:422 return resolve_ref(stack_name, value['Ref'],423 resources, attribute='PhysicalResourceId')424 elif len(value) == 1 and 'Fn::GetAtt' in value:425 return resolve_ref(stack_name, value['Fn::GetAtt'][0],426 resources, attribute=value['Fn::GetAtt'][1])427 else:428 for key, val in iteritems(value):429 value[key] = resolve_refs_recursively(stack_name, val, resources)430 if len(value) == 1 and 'Fn::Join' in value:431 return value['Fn::Join'][0].join(value['Fn::Join'][1])432 if isinstance(value, list):433 for i in range(0, len(value)):434 value[i] = resolve_refs_recursively(stack_name, value[i], resources)435 return value436def update_resource(resource_id, resources, stack_name):437 resource = resources[resource_id]438 resource_type = get_resource_type(resource)439 if resource_type not in UPDATEABLE_RESOURCES:440 LOG.warning('Unable to update resource type "%s", id "%s"' % (resource_type, resource_id))441 return442 props = resource['Properties']443 if resource_type == 'Lambda::Function':444 client = aws_stack.connect_to_service('lambda')445 keys = ('FunctionName', 'Role', 'Handler', 'Description', 'Timeout', 'MemorySize', 'Environment', 'Runtime')446 update_props = dict([(k, props[k]) for k in keys if k in props])447 return client.update_function_configuration(**update_props)448def deploy_resource(resource_id, resources, stack_name):449 resource = resources[resource_id]450 client = get_client(resource)451 if not client:452 return False453 resource_type = get_resource_type(resource)454 func_details = RESOURCE_TO_FUNCTION.get(resource_type)455 if not func_details:456 LOG.warning('Resource type not yet implemented: %s' % resource_type)457 return458 LOG.debug('Deploying resource type "%s" id "%s"' % (resource_type, resource_id))459 func_details = func_details[ACTION_CREATE]460 function = getattr(client, func_details['function'])461 params = dict(func_details['parameters'])462 defaults = func_details.get('defaults', {})463 if 'Properties' not in resource:464 resource['Properties'] = {}465 resource_props = resource['Properties']466 for param_key, prop_keys in iteritems(dict(params)):467 params.pop(param_key, None)468 if not isinstance(prop_keys, list):469 prop_keys = [prop_keys]470 for prop_key in prop_keys:471 if prop_key == PLACEHOLDER_RESOURCE_NAME:472 params[param_key] = resource_id473 resource_name = get_resource_name(resource)474 if resource_name:475 params[param_key] = resource_name476 else:477 # try to obtain physical resource name from stack resources478 try:479 return resolve_ref(stack_name, resource_id, resources,480 attribute='PhysicalResourceId')481 except Exception as e:482 LOG.debug('Unable to extract physical id for resource %s: %s' % (resource_id, e))483 else:484 if callable(prop_key):485 prop_value = prop_key(resource_props, stack_name=stack_name, resources=resources)486 else:487 prop_value = resource_props.get(prop_key)488 if prop_value is not None:489 params[param_key] = prop_value490 tmp_value = params.get(param_key)491 if tmp_value is not None:492 params[param_key] = resolve_refs_recursively(stack_name, tmp_value, resources)493 break494 # hack: convert to boolean495 if params.get(param_key) in ['True', 'False']:496 params[param_key] = params.get(param_key) == 'True'497 # assign default value if empty498 params = common.merge_recursive(defaults, params)499 # invoke function500 try:501 LOG.debug('Request for creating resource type "%s": %s' % (resource_type, params))502 result = function(**params)503 except Exception as e:504 LOG.warning('Error calling %s with params: %s for resource: %s' % (function, params, resource))505 raise e506 # some resources have attached/nested resources which we need to create recursively now507 if resource_type == 'ApiGateway::Method':508 integration = resource_props.get('Integration')509 if integration:510 api_id = resolve_refs_recursively(stack_name, resource_props['RestApiId'], resources)511 res_id = resolve_refs_recursively(stack_name, resource_props['ResourceId'], resources)512 uri = integration.get('Uri')513 if uri:514 uri = resolve_refs_recursively(stack_name, uri, resources)515 aws_stack.connect_to_service('apigateway').put_integration(restApiId=api_id, resourceId=res_id,516 httpMethod=resource_props['HttpMethod'], type=integration['Type'],517 integrationHttpMethod=integration['IntegrationHttpMethod'], uri=uri518 )519 elif resource_type == 'SNS::Topic':520 subscriptions = resource_props.get('Subscription', [])521 for subscription in subscriptions:522 endpoint = resolve_refs_recursively(stack_name, subscription['Endpoint'], resources)523 topic_arn = retrieve_topic_arn(params['Name'])524 aws_stack.connect_to_service('sns').subscribe(525 TopicArn=topic_arn, Protocol=subscription['Protocol'], Endpoint=endpoint)526 return result527def deploy_template(template, stack_name):528 if isinstance(template, string_types):529 template = parse_template(template)530 resource_map = template.get('Resources')531 if not resource_map:532 LOG.warning('CloudFormation template contains no Resources section')533 return534 next = resource_map535 iters = 10536 for i in range(0, iters):...
34107_template_deployer.py
Source:34107_template_deployer.py
...204 return aws_stack.connect_to_service('lambda').get_function(FunctionName=resource_id)205 if resource_type == 'AWS::Lambda::EventSourceMapping':206 resource_id = resource_props['FunctionName'] if resource else resource_id207 source_arn = resource_props.get('EventSourceArn')208 resource_id = resolve_refs_recursively(stack_name, resource_id, resources)209 source_arn = resolve_refs_recursively(stack_name, source_arn, resources)210 if not resource_id or not source_arn:211 raise Exception('ResourceNotFound')212 mappings = aws_stack.connect_to_service('lambda').list_event_source_mappings(213 FunctionName=resource_id, EventSourceArn=source_arn)214 mapping = list(filter(lambda m:215 m['EventSourceArn'] == source_arn and m['FunctionArn'] == aws_stack.lambda_function_arn(resource_id),216 mappings['EventSourceMappings']))217 if not mapping:218 raise Exception('ResourceNotFound')219 return mapping[0]220 if resource_type == 'AWS::DynamoDB::Table':221 resource_id = resource_props['TableName'] if resource else resource_id222 return aws_stack.connect_to_service('dynamodb').describe_table(TableName=resource_id)223 if resource_type == 'AWS::ApiGateway::RestApi':224 apis = aws_stack.connect_to_service('apigateway').get_rest_apis()['items']225 resource_id = resource_props['Name'] if resource else resource_id226 result = list(filter(lambda api: api['name'] == resource_id, apis))227 return result[0] if result else None228 if resource_type == 'AWS::ApiGateway::Resource':229 api_id = resource_props['RestApiId'] if resource else resource_id230 api_id = resolve_refs_recursively(stack_name, api_id, resources)231 parent_id = resolve_refs_recursively(stack_name, resource_props['ParentId'], resources)232 if not api_id or not parent_id:233 return None234 api_resources = aws_stack.connect_to_service('apigateway').get_resources(restApiId=api_id)['items']235 target_resource = list(filter(lambda res:236 res.get('parentId') == parent_id and res['pathPart'] == resource_props['PathPart'], api_resources))237 if not target_resource:238 return None239 path = aws_stack.get_apigateway_path_for_resource(api_id,240 target_resource[0]['id'], resources=api_resources)241 result = list(filter(lambda res: res['path'] == path, api_resources))242 return result[0] if result else None243 if resource_type == 'AWS::ApiGateway::Deployment':244 api_id = resource_props['RestApiId'] if resource else resource_id245 api_id = resolve_refs_recursively(stack_name, api_id, resources)246 if not api_id:247 return None248 result = aws_stack.connect_to_service('apigateway').get_deployments(restApiId=api_id)['items']249 # TODO possibly filter results by stage name or other criteria250 return result[0] if result else None251 if resource_type == 'AWS::ApiGateway::Method':252 api_id = resolve_refs_recursively(stack_name, resource_props['RestApiId'], resources)253 res_id = resolve_refs_recursively(stack_name, resource_props['ResourceId'], resources)254 if not api_id or not res_id:255 return None256 return aws_stack.connect_to_service('apigateway').get_method(restApiId=api_id,257 resourceId=res_id, httpMethod=resource_props['HttpMethod'])258 if resource_type == 'AWS::SQS::Queue':259 queues = aws_stack.connect_to_service('sqs').list_queues()260 result = list(filter(lambda item:261 # TODO possibly find a better way to compare resource_id with queue URLs262 item.endswith('/%s' % resource_id), queues['QueueUrls']))263 return result264 if resource_type == 'AWS::S3::Bucket':265 return aws_stack.connect_to_service('s3').get_bucket_location(Bucket=resource_id)266 if resource_type == 'AWS::Logs::LogGroup':267 # TODO implement268 raise Exception('ResourceNotFound')269 if resource_type == 'AWS::Kinesis::Stream':270 stream_name = resolve_refs_recursively(stack_name, resource_props['Name'], resources)271 result = aws_stack.connect_to_service('kinesis').describe_stream(StreamName=stream_name)272 return result273 if is_deployable_resource(resource):274 LOGGER.warning('Unexpected resource type %s when resolving references' % resource_type)275 except Exception as e:276 # we expect this to be a "not found" exception277 markers = ['NoSuchBucket', 'ResourceNotFound', '404']278 if not list(filter(lambda marker, e=e: marker in str(e), markers)):279 LOGGER.warning('Unexpected error retrieving details for resource %s: %s %s - %s %s' %280 (resource_type, e, traceback.format_exc(), resource, resource_status))281 return None282def extract_resource_attribute(resource_type, resource, attribute):283 LOGGER.debug('Extract resource attribute: %s %s' % (resource_type, attribute))284 # extract resource specific attributes285 if resource_type == 'Lambda::Function':286 actual_attribute = 'FunctionArn' if attribute == 'Arn' else attribute287 return resource['Configuration'][actual_attribute]288 elif resource_type == 'DynamoDB::Table':289 actual_attribute = 'LatestStreamArn' if attribute == 'StreamArn' else attribute290 value = resource['Table'].get(actual_attribute)291 return value292 elif resource_type == 'ApiGateway::RestApi':293 if attribute == 'PhysicalResourceId':294 return resource['id']295 if attribute == 'RootResourceId':296 resources = aws_stack.connect_to_service('apigateway').get_resources(restApiId=resource['id'])['items']297 for res in resources:298 if res['path'] == '/' and not res.get('parentId'):299 return res['id']300 elif resource_type == 'ApiGateway::Resource':301 if attribute == 'PhysicalResourceId':302 return resource['id']303 return resource.get(attribute)304def resolve_ref(stack_name, ref, resources, attribute):305 LOGGER.debug('Resolving ref %s - %s' % (ref, attribute))306 if ref == 'AWS::Region':307 return DEFAULT_REGION308 resource_status = describe_stack_resources(stack_name, ref)[0]309 attr_value = resource_status.get(attribute)310 if attr_value not in [None, '']:311 return attr_value312 # fetch resource details313 resource = resources.get(ref)314 resource_new = retrieve_resource_details(ref, resource_status, resources, stack_name)315 if not resource_new:316 return317 resource_type = get_resource_type(resource)318 result = extract_resource_attribute(resource_type, resource_new, attribute)319 if not result:320 LOGGER.warning('Unable to extract reference attribute %s from resource: %s' % (attribute, resource_new))321 return result322def resolve_refs_recursively(stack_name, value, resources):323 if isinstance(value, dict):324 if len(value) == 1 and 'Ref' in value:325 return resolve_ref(stack_name, value['Ref'],326 resources, attribute='PhysicalResourceId')327 elif len(value) == 1 and 'Fn::GetAtt' in value:328 return resolve_ref(stack_name, value['Fn::GetAtt'][0],329 resources, attribute=value['Fn::GetAtt'][1])330 else:331 for key, val in iteritems(value):332 value[key] = resolve_refs_recursively(stack_name, val, resources)333 if len(value) == 1 and 'Fn::Join' in value:334 return value['Fn::Join'][0].join(value['Fn::Join'][1])335 if isinstance(value, list):336 for i in range(0, len(value)):337 value[i] = resolve_refs_recursively(stack_name, value[i], resources)338 return value339def set_status_deployed(resource_id, resource, stack_name):340 # TODO341 pass342 # client = aws_stack.connect_to_service('cloudformation')343 # template = {344 # # TODO update deployment status345 # MARKER_DONT_REDEPLOY_STACK: {}346 # }347 # TODO: instead of calling update_stack, introduce a backdoor API method to348 # update the deployment status of individual resources. The problem with349 # using the code below is that it sets the status to UPDATE_COMPLETE which may350 # be undesirable (if the stack has just been created we expect CREATE_COMPLETE).351 # client.update_stack(StackName=stack_name, TemplateBody=json.dumps(template), UsePreviousTemplate=True)352def deploy_resource(resource_id, resources, stack_name):353 resource = resources[resource_id]354 client = get_client(resource)355 if not client:356 return False357 resource_type = get_resource_type(resource)358 func_details = RESOURCE_TO_FUNCTION.get(resource_type)359 if not func_details:360 LOGGER.warning('Resource type not yet implemented: %s' % resource['Type'])361 return362 LOGGER.debug('Deploying resource type "%s" id "%s"' % (resource_type, resource_id))363 func_details = func_details[ACTION_CREATE]364 function = getattr(client, func_details['function'])365 params = dict(func_details['parameters'])366 defaults = func_details.get('defaults', {})367 if 'Properties' not in resource:368 resource['Properties'] = {}369 resource_props = resource['Properties']370 for param_key, prop_keys in iteritems(dict(params)):371 params.pop(param_key, None)372 if not isinstance(prop_keys, list):373 prop_keys = [prop_keys]374 for prop_key in prop_keys:375 if prop_key == PLACEHOLDER_RESOURCE_NAME:376 # obtain physical resource name from stack resources377 params[param_key] = resolve_ref(stack_name, resource_id, resources,378 attribute='PhysicalResourceId')379 else:380 prop_value = resource_props.get(prop_key)381 if prop_value is not None:382 params[param_key] = prop_value383 tmp_value = params.get(param_key)384 if tmp_value is not None:385 params[param_key] = resolve_refs_recursively(stack_name, tmp_value, resources)386 break387 # hack: convert to boolean388 if params.get(param_key) in ['True', 'False']:389 params[param_key] = params.get(param_key) == 'True'390 # assign default value if empty391 params = common.merge_recursive(defaults, params)392 # invoke function393 try:394 result = function(**params)395 except Exception as e:396 LOGGER.warning('Error calling %s with params: %s for resource: %s' % (function, params, resource))397 raise e398 # some resources have attached/nested resources which we need to create recursively now399 if resource_type == 'ApiGateway::Method':400 integration = resource_props.get('Integration')401 if integration:402 api_id = resolve_refs_recursively(stack_name, resource_props['RestApiId'], resources)403 res_id = resolve_refs_recursively(stack_name, resource_props['ResourceId'], resources)404 uri = integration.get('Uri')405 if uri:406 uri = resolve_refs_recursively(stack_name, uri, resources)407 aws_stack.connect_to_service('apigateway').put_integration(restApiId=api_id, resourceId=res_id,408 httpMethod=resource_props['HttpMethod'], type=integration['Type'],409 integrationHttpMethod=integration['IntegrationHttpMethod'], uri=uri410 )411 # update status412 set_status_deployed(resource_id, resource, stack_name)413 return result414def deploy_template(template, stack_name):415 if isinstance(template, string_types):416 template = parse_template(template)417 if MARKER_DONT_REDEPLOY_STACK in template:418 # If we are currently deploying, then bail. This can occur if419 # deploy_template(..) method calls boto's update_stack(..) (to update the420 # state of resources) which itself triggers another call to deploy_template(..)....
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
