KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use is_deployable_resource method in localstack

Best Python code snippet using localstack_python
template_deployer.py
Source:template_deployer.py
...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):537        # get resource details538        for resource_id, resource in iteritems(next):539            stack_resource = describe_stack_resource(stack_name, resource_id)540            resource['__details__'] = stack_resource541        next = resources_to_deploy_next(resource_map, stack_name)542        if not next:543            return544        for resource_id, resource in iteritems(next):545            deploy_resource(resource_id, resource_map, stack_name=stack_name)546    LOG.warning('Unable to resolve all dependencies and deploy all resources ' +547        'after %s iterations. Remaining (%s): %s' % (iters, len(next), next))548# --------549# Util methods for analyzing resource dependencies550# --------551def is_deployable_resource(resource):552    resource_type = get_resource_type(resource)553    entry = RESOURCE_TO_FUNCTION.get(resource_type)554    if entry is None:555        LOG.warning('Unknown resource type "%s": %s' % (resource_type, resource))556    return entry and entry.get(ACTION_CREATE)557def is_deployed(resource_id, resources, stack_name):558    resource = resources[resource_id]559    resource_status = resource.get('__details__') or {}560    details = retrieve_resource_details(resource_id, resource_status, resources, stack_name)561    return bool(details)562def should_be_deployed(resource_id, resources, stack_name):563    """ Return whether the given resource is all of: (1) deployable, (2) not yet deployed,564        and (3) has no unresolved dependencies. """565    resource = resources[resource_id]566    if not is_deployable_resource(resource) or is_deployed(resource_id, resources, stack_name):567        return False568    res_deps = get_resource_dependencies(resource_id, resource, resources)569    return all_dependencies_satisfied(res_deps, stack_name, resources, resource_id)570def is_updateable(resource_id, resources, stack_name):571    """ Return whether the given resource can be updated or not """572    resource = resources[resource_id]573    if not is_deployable_resource(resource) or not is_deployed(resource_id, resources, stack_name):574        return False575    resource_type = get_resource_type(resource)576    return resource_type in UPDATEABLE_RESOURCES577def all_dependencies_satisfied(resources, stack_name, all_resources, depending_resource=None):578    for resource_id, resource in iteritems(resources):579        if is_deployable_resource(resource):580            if not is_deployed(resource_id, all_resources, stack_name):581                return False582    return True583def resources_to_deploy_next(resources, stack_name):584    result = {}585    for resource_id, resource in iteritems(resources):586        if should_be_deployed(resource_id, resources, stack_name):587            result[resource_id] = resource588    return result589def get_resource_dependencies(resource_id, resource, resources):590    result = {}591    dumped = json.dumps(common.json_safe(resource))592    for other_id, other in iteritems(resources):593        if resource != other:...
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.