KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use retrieve_resource_details method in localstack
Best Python code snippet using localstack_python
template_deployer.py
Source:template_deployer.py 
...190 for res in resources['StackResources']:191 if res.get('LogicalResourceId') == logical_resource_id:192 result.append(res)193 return result194def retrieve_resource_details(resource_id, resource_status, resources, stack_name):195 resource = resources[resource_id]196 resource_id = resource_status.get('PhysicalResourceId') or resource_id197 resource_type = resource_status['ResourceType']198 if not resource:199 resource = {}200 resource_props = resource.get('Properties')201 try:202 if resource_type == 'AWS::Lambda::Function':203 resource_id = resource_props['FunctionName'] if resource else resource_id204 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(..).421 # We don't want to end up in an infinite/recursive deployment loop.422 return423 resource_map = template.get('Resources')424 if not resource_map:425 LOGGER.warning('CloudFormation template contains no Resources section')426 return427 next = resource_map428 iters = 10429 for i in range(0, iters):430 # get resource details431 for resource_id, resource in iteritems(next):432 stack_resources = describe_stack_resources(stack_name, resource_id)433 resource['__details__'] = stack_resources[0]434 next = resources_to_deploy_next(resource_map, stack_name)435 if not next:436 return437 for resource_id, resource in iteritems(next):438 deploy_resource(resource_id, resource_map, stack_name=stack_name)439 LOGGER.warning('Unable to resolve all dependencies and deploy all resources ' +440 'after %s iterations. Remaining (%s): %s' % (iters, len(next), next))441# --------442# Util methods for analyzing resource dependencies443# --------444def is_deployable_resource(resource):445 resource_type = get_resource_type(resource)446 entry = RESOURCE_TO_FUNCTION.get(resource_type)447 if entry is None:448 LOGGER.warning('Unknown resource type "%s"' % resource_type)449 return entry and entry.get(ACTION_CREATE)450def is_deployed(resource_id, resources, stack_name):451 resource = resources[resource_id]452 resource_status = resource['__details__']453 details = retrieve_resource_details(resource_id, resource_status, resources, stack_name)454 return bool(details)455def all_dependencies_satisfied(resources, stack_name, all_resources, depending_resource=None):456 for resource_id, resource in iteritems(resources):457 if is_deployable_resource(resource):458 if not is_deployed(resource_id, all_resources, stack_name):459 return False460 return True461def resources_to_deploy_next(resources, stack_name):462 result = {}463 for resource_id, resource in iteritems(resources):464 if is_deployable_resource(resource) and not is_deployed(resource_id, resources, stack_name):465 res_deps = get_resource_dependencies(resource_id, resource, resources)466 if all_dependencies_satisfied(res_deps, stack_name, resources, resource_id):467 result[resource_id] = resource...
34107_template_deployer.py
Source:34107_template_deployer.py
...190 for res in resources['StackResources']:191 if res.get('LogicalResourceId') == logical_resource_id:192 result.append(res)193 return result194def retrieve_resource_details(resource_id, resource_status, resources, stack_name):195 resource = resources[resource_id]196 resource_id = resource_status.get('PhysicalResourceId') or resource_id197 resource_type = resource_status['ResourceType']198 if not resource:199 resource = {}200 resource_props = resource.get('Properties')201 try:202 if resource_type == 'AWS::Lambda::Function':203 resource_id = resource_props['FunctionName'] if resource else resource_id204 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(..).421 # We don't want to end up in an infinite/recursive deployment loop.422 return423 resource_map = template.get('Resources')424 if not resource_map:425 LOGGER.warning('CloudFormation template contains no Resources section')426 return427 next = resource_map428 iters = 10429 for i in range(0, iters):430 # get resource details431 for resource_id, resource in iteritems(next):432 stack_resources = describe_stack_resources(stack_name, resource_id)433 resource['__details__'] = stack_resources[0]434 next = resources_to_deploy_next(resource_map, stack_name)435 if not next:436 return437 for resource_id, resource in iteritems(next):438 deploy_resource(resource_id, resource_map, stack_name=stack_name)439 LOGGER.warning('Unable to resolve all dependencies and deploy all resources ' +440 'after %s iterations. Remaining (%s): %s' % (iters, len(next), next))441# --------442# Util methods for analyzing resource dependencies443# --------444def is_deployable_resource(resource):445 resource_type = get_resource_type(resource)446 entry = RESOURCE_TO_FUNCTION.get(resource_type)447 if entry is None:448 LOGGER.warning('Unknown resource type "%s"' % resource_type)449 return entry and entry.get(ACTION_CREATE)450def is_deployed(resource_id, resources, stack_name):451 resource = resources[resource_id]452 resource_status = resource['__details__']453 details = retrieve_resource_details(resource_id, resource_status, resources, stack_name)454 return bool(details)455def all_dependencies_satisfied(resources, stack_name, all_resources, depending_resource=None):456 for resource_id, resource in iteritems(resources):457 if is_deployable_resource(resource):458 if not is_deployed(resource_id, all_resources, stack_name):459 return False460 return True461def resources_to_deploy_next(resources, stack_name):462 result = {}463 for resource_id, resource in iteritems(resources):464 if is_deployable_resource(resource) and not is_deployed(resource_id, resources, stack_name):465 res_deps = get_resource_dependencies(resource_id, resource, resources)466 if all_dependencies_satisfied(res_deps, stack_name, resources, resource_id):467 result[resource_id] = resource...
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?
