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

Best Python code snippet using localstack_python
template_deployer.py
Source:template_deployer.py
...167        if resource_type == "Parameter":168            return resource_props169        message = (170            f"Unexpected resource type {resource_type} when resolving "171            f"references of resource {resource_id}: {dump_resource_as_json(resource)}"172        )173        log_not_available_message(resource_type=resource_type, message=message)174    except DependencyNotYetSatisfied:175        return176    except Exception as e:177        check_not_found_exception(e, resource_type, resource, resource_status)178    return None179def check_not_found_exception(e, resource_type, resource, resource_status=None):180    # we expect this to be a "not found" exception181    markers = [182        "NoSuchBucket",183        "ResourceNotFound",184        "NoSuchEntity",185        "NotFoundException",186        "404",187        "not found",188        "not exist",189    ]190    if not list(filter(lambda marker, e=e: marker in str(e), markers)):191        LOG.warning(192            "Unexpected error retrieving details for resource type %s: Exception: %s - %s - status: %s",193            resource_type,194            e,195            resource,196            resource_status,197        )198        return False199    return True200def extract_resource_attribute(201    resource_type,202    resource_state,203    attribute,204    resource_id=None,205    resource=None,206    stack=None,207):208    LOG.debug("Extract resource attribute: %s %s", resource_type, attribute)209    is_ref_attribute = attribute in ["PhysicalResourceId", "Ref"]210    is_ref_attr_or_arn = is_ref_attribute or attribute == "Arn"211    resource = resource or {}212    if not resource and stack.resources:213        resource = stack.resources[resource_id]214    if not resource_state:215        resource_state = retrieve_resource_details(resource_id, {}, stack=stack)216        if not resource_state:217            raise DependencyNotYetSatisfied(218                resource_ids=resource_id,219                message='Unable to fetch details for resource "%s" (attribute "%s")'220                % (resource_id, attribute),221            )222    if isinstance(resource_state, GenericBaseModel):223        if hasattr(resource_state, "get_cfn_attribute"):224            try:225                return resource_state.get_cfn_attribute(attribute)226            except Exception:227                pass228        raise Exception(229            'Unable to extract attribute "%s" from "%s" model class %s'230            % (attribute, resource_type, type(resource_state))231        )232    # extract resource specific attributes233    # TODO: remove the code below - move into resource model classes!234    resource_props = resource.get("Properties", {})235    if resource_type == "Parameter":236        result = None237        param_value = resource_props.get(238            "Value",239            resource.get("Value", resource_props.get("Properties", {}).get("Value")),240        )241        if is_ref_attr_or_arn:242            result = param_value243        elif isinstance(param_value, dict):244            result = param_value.get(attribute)245        if result is not None:246            return result247        return ""248    elif resource_type == "Lambda::Function":249        func_configs = resource_state.get("Configuration") or {}250        if is_ref_attr_or_arn:251            func_arn = func_configs.get("FunctionArn")252            if func_arn:253                return resolve_refs_recursively(stack, func_arn)254            func_name = resolve_refs_recursively(stack, func_configs.get("FunctionName"))255            return aws_stack.lambda_function_arn(func_name)256        else:257            return func_configs.get(attribute)258    elif resource_type == "Lambda::Version":259        if resource_state.get("Version"):260            return "%s:%s" % (261                resource_state.get("FunctionArn"),262                resource_state.get("Version").split(":")[-1],263            )264    elif resource_type == "DynamoDB::Table":265        actual_attribute = "LatestStreamArn" if attribute == "StreamArn" else attribute266        value = resource_state.get("Table", {}).get(actual_attribute)267        if value:268            return value269    elif resource_type == "ApiGateway::RestApi":270        if is_ref_attribute:271            result = resource_state.get("id")272            if result:273                return result274        if attribute == "RootResourceId":275            api_id = resource_state["id"]276            resources = aws_stack.connect_to_service("apigateway").get_resources(restApiId=api_id)[277                "items"278            ]279            for res in resources:280                if res["path"] == "/" and not res.get("parentId"):281                    return res["id"]282    elif resource_type == "ApiGateway::Resource":283        if is_ref_attribute:284            return resource_state.get("id")285    elif resource_type == "ApiGateway::Deployment":286        if is_ref_attribute:287            return resource_state.get("id")288    elif resource_type == "S3::Bucket":289        if attribute == "WebsiteURL":290            bucket_name = resource_props.get("BucketName")291            return f"http://{bucket_name}.{S3_STATIC_WEBSITE_HOSTNAME}"292        if is_ref_attr_or_arn:293            bucket_name = resource_props.get("BucketName")294            bucket_name = resolve_refs_recursively(stack, bucket_name)295            if attribute == "Arn":296                return aws_stack.s3_bucket_arn(bucket_name)297            return bucket_name298    elif resource_type == "Elasticsearch::Domain":299        if attribute == "DomainEndpoint":300            domain_status = resource_state.get("DomainStatus", {})301            result = domain_status.get("Endpoint")302            if result:303                return result304        if attribute in ["Arn", "DomainArn"]:305            domain_name = resource_props.get("DomainName") or resource_state.get("DomainName")306            return aws_stack.es_domain_arn(domain_name)307    elif resource_type == "StepFunctions::StateMachine":308        if is_ref_attr_or_arn:309            return resource_state["stateMachineArn"]310    elif resource_type == "SNS::Topic":311        if is_ref_attribute and resource_state.get("TopicArn"):312            topic_arn = resource_state.get("TopicArn")313            return resolve_refs_recursively(stack, topic_arn)314    elif resource_type == "SQS::Queue":315        if is_ref_attr_or_arn:316            if attribute == "Arn" and resource_state.get("QueueArn"):317                return resolve_refs_recursively(stack, resource_state.get("QueueArn"))318            return aws_stack.get_sqs_queue_url(resource_props.get("QueueName"))319    attribute_lower = first_char_to_lower(attribute)320    result = resource_state.get(attribute) or resource_state.get(attribute_lower)321    if result is None and isinstance(resource, dict):322        result = resource_props.get(attribute) or resource_props.get(attribute_lower)323        if result is None:324            result = get_attr_from_model_instance(325                resource,326                attribute,327                resource_type=resource_type,328                resource_id=resource_id,329            )330    if is_ref_attribute:331        for attr in ["Id", "PhysicalResourceId", "Ref"]:332            if result is None:333                for obj in [resource_state, resource]:334                    result = result or obj.get(attr)335    return result336def canonical_resource_type(resource_type):337    if "::" in resource_type and not resource_type.startswith("AWS::"):338        resource_type = "AWS::%s" % resource_type339    return resource_type340def get_attr_from_model_instance(resource, attribute, resource_type, resource_id=None):341    resource_type = canonical_resource_type(resource_type)342    model_class = RESOURCE_MODELS.get(resource_type)343    if not model_class:344        if resource_type not in ["AWS::Parameter", "Parameter"]:345            LOG.debug('Unable to find model class for resource type "%s"', resource_type)346        return347    try:348        inst = model_class(resource_name=resource_id, resource_json=resource)349        return inst.get_cfn_attribute(attribute)350    except Exception as e:351        LOG.debug("Failed to retrieve model attribute: %s", attribute, exc_info=e)352def resolve_ref(stack, ref, attribute):353    stack_name = stack.stack_name354    resources = stack.resources355    if ref == "AWS::Region":356        return aws_stack.get_region()357    if ref == "AWS::Partition":358        return "aws"359    if ref == "AWS::StackName":360        return stack_name361    if ref == "AWS::StackId":362        # TODO return proper stack id!363        return stack_name364    if ref == "AWS::AccountId":365        return TEST_AWS_ACCOUNT_ID366    if ref == "AWS::NoValue":367        return PLACEHOLDER_AWS_NO_VALUE368    if ref == "AWS::NotificationARNs":369        # TODO!370        return {}371    if ref == "AWS::URLSuffix":372        return AWS_URL_SUFFIX373    is_ref_attribute = attribute in ["Ref", "PhysicalResourceId", "Arn"]374    if is_ref_attribute:375        # extract the Properties here, as we only want to recurse over the resource props...376        resource_props = resources.get(ref, {}).get("Properties")377        resolve_refs_recursively(stack, resource_props)378        return determine_resource_physical_id(379            resource_id=ref,380            attribute=attribute,381            stack=stack,382        )383    if resources.get(ref):384        if isinstance(resources[ref].get(attribute), (str, int, float, bool, dict)):385            return resources[ref][attribute]386    # fetch resource details387    resource_new = retrieve_resource_details(ref, {}, stack=stack)388    if not resource_new:389        raise DependencyNotYetSatisfied(390            resource_ids=ref,391            message='Unable to fetch details for resource "%s" (resolving attribute "%s")'392            % (ref, attribute),393        )394    resource = resources.get(ref)395    resource_type = get_resource_type(resource)396    result = extract_resource_attribute(397        resource_type, resource_new, attribute, resource_id=ref, resource=resource, stack=stack398    )399    if result is None:400        LOG.warning(401            'Unable to extract reference attribute "%s" from resource: %s %s',402            attribute,403            resource_new,404            resource,405        )406    return result407# Using a @prevent_stack_overflow decorator here to avoid infinite recursion408# in case we load stack exports that have circular dependencies (see issue 3438)409# TODO: Potentially think about a better approach in the future410@prevent_stack_overflow(match_parameters=True)411def resolve_refs_recursively(stack, value):412    result = _resolve_refs_recursively(stack, value)413    # localstack specific patches414    if isinstance(result, str):415        # we're trying to filter constructed API urls here (e.g. via Join in the template)416        api_match = REGEX_OUTPUT_APIGATEWAY.match(result)417        if api_match:418            prefix = api_match[1]419            host = api_match[2]420            path = api_match[3]421            port = config.service_port("apigateway")422            return f"{prefix}{host}:{port}/{path}"423        # basic dynamic reference support424        # see: https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/dynamic-references.html425        # technically there are more restrictions for each of these services but checking each of these426        # isn't really necessary for the current level of emulation427        dynamic_ref_match = REGEX_DYNAMIC_REF.match(result)428        if dynamic_ref_match:429            service_name = dynamic_ref_match[1]430            reference_key = dynamic_ref_match[2]431            # only these 3 services are supported for dynamic references right now432            if service_name == "ssm":433                ssm_client = aws_stack.connect_to_service("ssm")434                return ssm_client.get_parameter(Name=reference_key)["Parameter"]["Value"]435            elif service_name == "ssm-secure":436                ssm_client = aws_stack.connect_to_service("ssm")437                return ssm_client.get_parameter(Name=reference_key, WithDecryption=True)[438                    "Parameter"439                ]["Value"]440            elif service_name == "secretsmanager":441                # reference key needs to be parsed further442                # because {{resolve:secretsmanager:secret-id:secret-string:json-key:version-stage:version-id}}443                # we match for "secret-id:secret-string:json-key:version-stage:version-id"444                # where445                #   secret-id can either be the secret name or the full ARN of the secret446                #   secret-string *must* be SecretString447                #   all other values are optional448                secret_id = reference_key449                [json_key, version_stage, version_id] = [None, None, None]450                if "SecretString" in reference_key:451                    parts = reference_key.split(":SecretString:")452                    secret_id = parts[0]453                    [json_key, version_stage, version_id] = parts[1].split(":")454                kwargs = {}  # optional args for get_secret_value455                if version_id:456                    kwargs["VersionId"] = version_id457                if version_stage:458                    kwargs["VersionStage"] = version_stage459                secretsmanager_client = aws_stack.connect_to_service("secretsmanager")460                secret_value = secretsmanager_client.get_secret_value(SecretId=secret_id, **kwargs)[461                    "SecretString"462                ]463                if json_key:464                    return json.loads(secret_value)[json_key]465                else:466                    return secret_value467            else:468                LOG.warning(f"Unsupported service for dynamic parameter: {service_name=}")469    return result470@prevent_stack_overflow(match_parameters=True)471# TODO: move Stack model into separate file and add type hints here472def _resolve_refs_recursively(stack, value):473    if isinstance(value, dict):474        keys_list = list(value.keys())475        stripped_fn_lower = keys_list[0].lower().split("::")[-1] if len(keys_list) == 1 else None476        # process special operators477        if keys_list == ["Ref"]:478            ref = resolve_ref(stack, value["Ref"], attribute="Ref")479            if ref is None:480                resources = stack.resources481                msg = 'Unable to resolve Ref for resource "%s" (yet)' % value["Ref"]482                LOG.debug("%s - %s", msg, resources.get(value["Ref"]) or set(resources.keys()))483                raise DependencyNotYetSatisfied(resource_ids=value["Ref"], message=msg)484            ref = resolve_refs_recursively(stack, ref)485            return ref486        if stripped_fn_lower == "getatt":487            attr_ref = value[keys_list[0]]488            attr_ref = attr_ref.split(".") if isinstance(attr_ref, str) else attr_ref489            return resolve_ref(stack, attr_ref[0], attribute=attr_ref[1])490        if stripped_fn_lower == "join":491            join_values = value[keys_list[0]][1]492            join_values = [resolve_refs_recursively(stack, v) for v in join_values]493            none_values = [v for v in join_values if v is None]494            if none_values:495                raise Exception(496                    "Cannot resolve CF fn::Join %s due to null values: %s" % (value, join_values)497                )498            return value[keys_list[0]][0].join([str(v) for v in join_values])499        if stripped_fn_lower == "sub":500            item_to_sub = value[keys_list[0]]501            attr_refs = {r: {"Ref": r} for r in STATIC_REFS}502            if not isinstance(item_to_sub, list):503                item_to_sub = [item_to_sub, {}]504            result = item_to_sub[0]505            item_to_sub[1].update(attr_refs)506            for key, val in item_to_sub[1].items():507                val = resolve_refs_recursively(stack, val)508                result = result.replace("${%s}" % key, val)509            # resolve placeholders510            result = resolve_placeholders_in_string(result, stack=stack)511            return result512        if stripped_fn_lower == "findinmap":513            attr = resolve_refs_recursively(stack, value[keys_list[0]][1])514            result = resolve_ref(stack, value[keys_list[0]][0], attribute=attr)515            if not result:516                resources = stack.resources517                raise Exception(518                    f"Cannot resolve fn::FindInMap: {value[keys_list[0]]} {list(resources.keys())}"519                )520            key = value[keys_list[0]][2]521            if not isinstance(key, str):522                key = resolve_refs_recursively(stack, key)523            return result.get(key)524        if stripped_fn_lower == "importvalue":525            import_value_key = resolve_refs_recursively(stack, value[keys_list[0]])526            stack_export = stack.exports_map.get(import_value_key) or {}527            if not stack_export.get("Value"):528                LOG.info(529                    'Unable to find export "%s" in stack "%s", existing export names: %s',530                    import_value_key,531                    stack.stack_name,532                    list(stack.exports_map.keys()),533                )534                return None535            return stack_export["Value"]536        if stripped_fn_lower == "if":537            condition, option1, option2 = value[keys_list[0]]538            condition = evaluate_condition(stack, condition)539            return resolve_refs_recursively(stack, option1 if condition else option2)540        if stripped_fn_lower == "condition":541            result = evaluate_condition(stack, value[keys_list[0]])542            return result543        if stripped_fn_lower == "not":544            condition = value[keys_list[0]][0]545            condition = resolve_refs_recursively(stack, condition)546            return not condition547        if stripped_fn_lower in ["and", "or"]:548            conditions = value[keys_list[0]]549            results = [resolve_refs_recursively(stack, cond) for cond in conditions]550            result = all(results) if stripped_fn_lower == "and" else any(results)551            return result552        if stripped_fn_lower == "equals":553            operand1, operand2 = value[keys_list[0]]554            operand1 = resolve_refs_recursively(stack, operand1)555            operand2 = resolve_refs_recursively(stack, operand2)556            return str(operand1) == str(operand2)557        if stripped_fn_lower == "select":558            index, values = value[keys_list[0]]559            index = resolve_refs_recursively(stack, index)560            values = resolve_refs_recursively(stack, values)561            return values[index]562        if stripped_fn_lower == "split":563            delimiter, string = value[keys_list[0]]564            delimiter = resolve_refs_recursively(stack, delimiter)565            string = resolve_refs_recursively(stack, string)566            return string.split(delimiter)567        if stripped_fn_lower == "getazs":568            region = resolve_refs_recursively(stack, value["Fn::GetAZs"]) or aws_stack.get_region()569            azs = []570            for az in ("a", "b", "c", "d"):571                azs.append("%s%s" % (region, az))572            return azs573        if stripped_fn_lower == "base64":574            value_to_encode = value[keys_list[0]]575            value_to_encode = resolve_refs_recursively(stack, value_to_encode)576            return to_str(base64.b64encode(to_bytes(value_to_encode)))577        for key, val in dict(value).items():578            value[key] = resolve_refs_recursively(stack, val)579    if isinstance(value, list):580        for i in range(len(value)):581            value[i] = resolve_refs_recursively(stack, value[i])582    return value583def resolve_placeholders_in_string(result, stack):584    resources = stack.resources585    def _replace(match):586        parts = match.group(1).split(".")587        if len(parts) >= 2:588            resource_name, _, attr_name = match.group(1).partition(".")589            resolved = resolve_ref(stack, resource_name.strip(), attribute=attr_name.strip())590            if resolved is None:591                raise DependencyNotYetSatisfied(592                    resource_ids=resource_name,593                    message="Unable to resolve attribute ref %s" % match.group(1),594                )595            return resolved596        if len(parts) == 1 and parts[0] in resources:597            resource_json = resources[parts[0]]598            resource_type = get_resource_type(resource_json)599            result = extract_resource_attribute(600                resource_type,601                resource_json.get(KEY_RESOURCE_STATE, {}),602                "Ref",603                stack=stack,604                resource_id=parts[0],605            )606            if result is None:607                raise DependencyNotYetSatisfied(608                    resource_ids=parts[0],609                    message="Unable to resolve attribute ref %s" % match.group(1),610                )611            # make sure we resolve any functions/placeholders in the extracted string612            result = resolve_refs_recursively(stack, result)613            # make sure we convert the result to string614            result = "" if result is None else str(result)615            return result616        # TODO raise exception here?617        return match.group(0)618    regex = r"\$\{([^\}]+)\}"619    result = re.sub(regex, _replace, result)620    return result621def evaluate_condition(stack, condition):622    condition = resolve_refs_recursively(stack, condition)623    condition = resolve_ref(stack, condition, attribute="Ref")624    condition = resolve_refs_recursively(stack, condition)625    return condition626def evaluate_resource_condition(stack, resource):627    condition = resource.get("Condition")628    if condition:629        condition = evaluate_condition(stack, condition)630        if condition is False or condition in FALSE_STRINGS or is_none_or_empty_value(condition):631            return False632    return True633def get_stack_parameter(stack_name, parameter):634    try:635        client = aws_stack.connect_to_service("cloudformation")636        stack = client.describe_stacks(StackName=stack_name)["Stacks"]637    except Exception:638        return None639    stack = stack and stack[0]640    if not stack:641        return None642    result = [p["ParameterValue"] for p in stack["Parameters"] if p["ParameterKey"] == parameter]643    return (result or [None])[0]644def update_resource(resource_id, stack):645    resources = stack.resources646    stack_name = stack.stack_name647    resource = resources[resource_id]648    resource_type = get_resource_type(resource)649    if resource_type not in UPDATEABLE_RESOURCES:650        LOG.warning('Unable to update resource type "%s", id "%s"', resource_type, resource_id)651        return652    LOG.info("Updating resource %s of type %s", resource_id, resource_type)653    instance = get_resource_model_instance(resource_id, stack=stack)654    if instance:655        result = instance.update_resource(resource, stack_name=stack_name, resources=resources)656        instance.fetch_and_update_state(stack_name=stack_name, resources=resources)657        return result658def get_resource_model_instance(resource_id: str, stack) -> Optional[GenericBaseModel]:659    """Obtain a typed resource entity instance representing the given stack resource."""660    resource = stack.resources[resource_id]661    resource_type = get_resource_type(resource)662    canonical_type = canonical_resource_type(resource_type)663    resource_class = RESOURCE_MODELS.get(canonical_type)664    if not resource_class:665        return None666    instance = resource_class(resource)667    return instance668def fix_account_id_in_arns(params):669    def fix_ids(o, **kwargs):670        if isinstance(o, dict):671            for k, v in o.items():672                if is_string(v, exclude_binary=True):673                    o[k] = aws_stack.fix_account_id_in_arns(v)674        elif is_string(o, exclude_binary=True):675            o = aws_stack.fix_account_id_in_arns(o)676        return o677    result = recurse_object(params, fix_ids)678    return result679def convert_data_types(func_details, params):680    """Convert data types in the "params" object, with the type defs681    specified in the 'types' attribute of "func_details"."""682    types = func_details.get("types") or {}683    attr_names = types.keys() or []684    def cast(_obj, _type):685        if _type == bool:686            return _obj in ["True", "true", True]687        if _type == str:688            if isinstance(_obj, bool):689                return str(_obj).lower()690            return str(_obj)691        if _type in (int, float):692            return _type(_obj)693        return _obj694    def fix_types(o, **kwargs):695        if isinstance(o, dict):696            for k, v in o.items():697                if k in attr_names:698                    o[k] = cast(v, types[k])699        return o700    result = recurse_object(params, fix_types)701    return result702def log_not_available_message(resource_type: str, message: str):703    LOG.warning(704        f"{message}. To find out if {resource_type} is supported in LocalStack Pro, "705        "please check out our docs at https://docs.localstack.cloud/aws/cloudformation"706    )707def dump_resource_as_json(resource: Dict) -> str:708    return str(run_safe(lambda: json.dumps(json_safe(resource))) or resource)709# TODO remove this method710def prepare_template_body(req_data):711    return template_preparer.prepare_template_body(req_data)712def deploy_resource(stack, resource_id):713    result = execute_resource_action(resource_id, stack, ACTION_CREATE)714    return result715def delete_resource(stack, resource_id):716    return execute_resource_action(resource_id, stack, ACTION_DELETE)717def execute_resource_action(resource_id: str, stack, action_name: str):718    stack_name = stack.stack_name719    resources = stack.resources720    resource = resources[resource_id]721    resource_type = get_resource_type(resource)722    func_details = get_deployment_config(resource_type)723    if not func_details or action_name not in func_details:724        if resource_type in ["Parameter"]:725            return726        log_not_available_message(727            resource_type=resource_type,728            message=f"Action {action_name} for resource type {resource_type} not available",729        )730        return731    LOG.debug(732        'Running action "%s" for resource type "%s" id "%s"',733        action_name,734        resource_type,735        resource_id,736    )737    func_details = func_details[action_name]738    func_details = func_details if isinstance(func_details, list) else [func_details]739    results = []740    for func in func_details:741        if callable(func["function"]):742            result = func["function"](resource_id, resources, resource_type, func, stack_name)743            results.append(result)744            continue745        client = get_client(resource, func)746        if client:747            result = configure_resource_via_sdk(748                stack,749                resource_id,750                resource_type,751                func,752                action_name,753            )754            results.append(result)755    return (results or [None])[0]756def configure_resource_via_sdk(stack, resource_id, resource_type, func_details, action_name):757    resources = stack.resources758    stack_name = stack.stack_name759    resource = resources[resource_id]760    if resource_type == "EC2::Instance":761        if action_name == "create":762            func_details["boto_client"] = "resource"763    client = get_client(resource, func_details)764    function = getattr(client, func_details["function"])765    params = func_details.get("parameters") or lambda_get_params()766    defaults = func_details.get("defaults", {})767    resource_props = resource["Properties"] = resource.get("Properties", {})768    resource_props = dict(resource_props)769    resource_state = resource.get(KEY_RESOURCE_STATE, {})770    if callable(params):771        params = params(772            resource_props,773            stack_name=stack_name,774            resources=resources,775            resource_id=resource_id,776        )777    else:778        # it could be a list like ['param1', 'param2', {'apiCallParamName': 'cfResourcePropName'}]779        if isinstance(params, list):780            _params = {}781            for param in params:782                if isinstance(param, dict):783                    _params.update(param)784                else:785                    _params[param] = param786            params = _params787        params = dict(params)788        for param_key, prop_keys in dict(params).items():789            params.pop(param_key, None)790            if not isinstance(prop_keys, list):791                prop_keys = [prop_keys]792            for prop_key in prop_keys:793                if prop_key == PLACEHOLDER_RESOURCE_NAME:794                    params[param_key] = PLACEHOLDER_RESOURCE_NAME795                else:796                    if callable(prop_key):797                        prop_value = prop_key(798                            resource_props,799                            stack_name=stack_name,800                            resources=resources,801                            resource_id=resource_id,802                        )803                    else:804                        prop_value = resource_props.get(805                            prop_key,806                            resource.get(prop_key, resource_state.get(prop_key)),807                        )808                    if prop_value is not None:809                        params[param_key] = prop_value810                        break811    # replace PLACEHOLDER_RESOURCE_NAME in params812    resource_name_holder = {}813    def fix_placeholders(o, **kwargs):814        if isinstance(o, dict):815            for k, v in o.items():816                if v == PLACEHOLDER_RESOURCE_NAME:817                    if "value" not in resource_name_holder:818                        resource_name_holder["value"] = get_resource_name(resource) or resource_id819                    o[k] = resource_name_holder["value"]820        return o821    recurse_object(params, fix_placeholders)822    # assign default values if empty823    params = merge_recursive(defaults, params)824    # this is an indicator that we should skip this resource deployment, and return825    if params is None:826        return827    # convert refs828    for param_key, param_value in dict(params).items():829        if param_value is not None:830            params[param_key] = resolve_refs_recursively(stack, param_value)831    # convert any moto account IDs (123456789012) in ARNs to our format (000000000000)832    params = fix_account_id_in_arns(params)833    # convert data types (e.g., boolean strings to bool)834    params = convert_data_types(func_details, params)835    # remove None values, as they usually raise boto3 errors836    params = remove_none_values(params)837    # convert boolean strings838    #  (TODO: we should find a more reliable mechanism than this opportunistic/probabilistic approach!)839    params_before_conversion = copy.deepcopy(params)840    for param_key, param_value in dict(params).items():841        # Convert to boolean (TODO: do this recursively?)842        if str(param_value).lower() in ["true", "false"]:843            params[param_key] = str(param_value).lower() == "true"844    # invoke function845    try:846        LOG.debug(847            'Request for resource type "%s" in region %s: %s %s',848            resource_type,849            aws_stack.get_region(),850            func_details["function"],851            params,852        )853        try:854            result = function(**params)855        except botocore.exceptions.ParamValidationError as e:856            LOG.debug(f"Trying original parameters: {params_before_conversion}")857            if "type: <class 'bool'>" not in str(e):858                raise859            result = function(**params_before_conversion)860    except Exception as e:861        if action_name == "delete" and check_not_found_exception(e, resource_type, resource):862            return863        LOG.warning("Error calling %s with params: %s for resource: %s", function, params, resource)864        raise e865    return result866def get_action_name_for_resource_change(res_change):867    return {"Add": "CREATE", "Remove": "DELETE", "Modify": "UPDATE"}.get(res_change)868# TODO: this shouldn't be called for stack parameters869def determine_resource_physical_id(resource_id, stack=None, attribute=None):870    resources = stack.resources871    stack_name = stack.stack_name872    resource = resources.get(resource_id, {})873    if not resource:874        return875    resource_type = get_resource_type(resource)876    resource_type = re.sub("^AWS::", "", resource_type)877    resource_props = resource.get("Properties", {})878    # determine result from resource class879    canonical_type = canonical_resource_type(resource_type)880    resource_class = RESOURCE_MODELS.get(canonical_type)881    if resource_class:882        resource_inst = resource_class(resource)883        resource_inst.fetch_state_if_missing(stack_name=stack_name, resources=resources)884        result = resource_inst.get_physical_resource_id(attribute=attribute)885        if result:886            return result887    # TODO: put logic into resource-specific model classes!888    if resource_type == "ApiGateway::RestApi":889        result = resource_props.get("id")890        if result:891            return result892    elif resource_type == "ApiGateway::Stage":893        return resource_props.get("StageName")894    elif resource_type == "AppSync::DataSource":895        return resource_props.get("DataSourceArn")896    elif resource_type == "KinesisFirehose::DeliveryStream":897        return aws_stack.firehose_stream_arn(resource_props.get("DeliveryStreamName"))898    elif resource_type == "StepFunctions::StateMachine":899        return aws_stack.state_machine_arn(900            resource_props.get("StateMachineName")901        )  # returns ARN in AWS902    elif resource_type == "S3::Bucket":903        if attribute == "Arn":904            return aws_stack.s3_bucket_arn(resource_props.get("BucketName"))905        return resource_props.get("BucketName")  # Note: "Ref" returns bucket name in AWS906    elif resource_type == "IAM::Role":907        if attribute == "Arn":908            return aws_stack.role_arn(resource_props.get("RoleName"))909        return resource_props.get("RoleName")910    elif resource_type == "IAM::Policy":911        if attribute == "Arn":912            return aws_stack.policy_arn(resource_props.get("PolicyName"))913        return resource_props.get("PolicyName")914    elif resource_type == "DynamoDB::Table":915        table_name = resource_props.get("TableName")916        if table_name:917            return table_name918    elif resource_type == "Logs::LogGroup":919        return resource_props.get("LogGroupName")920    elif resource_type == "ApiGateway::Model":921        model_name = resource_props.get("Name")922        if model_name:923            return model_name924    res_id = resource.get("PhysicalResourceId")925    if res_id and attribute in [None, "Ref", "PhysicalResourceId"]:926        return res_id927    result = extract_resource_attribute(928        resource_type,929        {},930        attribute or "PhysicalResourceId",931        resource_id=resource_id,932        resource=resource,933        stack=stack,934    )935    if result is not None:936        # note that value could be an empty string here (in case of Parameter values)937        return result938    LOG.info(939        'Unable to determine PhysicalResourceId for "%s" resource, ID "%s"',940        resource_type,941        resource_id,942    )943def update_resource_details(stack, resource_id, details, action=None):944    resource = stack.resources.get(resource_id, {})945    if not resource or not details:946        return947    # TODO: we need to rethink this method - this should be encapsulated in the resource model classes.948    #   Also, instead of actively updating the PhysicalResourceId attributes below, they should be949    #   determined and returned by the resource model classes upon request.950    resource_type = resource.get("Type") or ""951    resource_type = re.sub("^AWS::", "", resource_type)952    resource_props = resource.get("Properties", {})953    if resource_type == "ApiGateway::RestApi":954        resource_props["id"] = details["id"]955    if resource_type == "KMS::Key":956        resource["PhysicalResourceId"] = details["KeyMetadata"]["KeyId"]957    if resource_type == "EC2::Instance":958        if details and isinstance(details, list) and hasattr(details[0], "id"):959            resource["PhysicalResourceId"] = details[0].id960        if isinstance(details, dict) and details.get("InstanceId"):961            resource["PhysicalResourceId"] = details["InstanceId"]962    if resource_type == "EC2::SecurityGroup":963        resource["PhysicalResourceId"] = details["GroupId"]964    if resource_type == "IAM::InstanceProfile":965        resource["PhysicalResourceId"] = details["InstanceProfile"]["InstanceProfileName"]966    if resource_type == "StepFunctions::Activity":967        resource["PhysicalResourceId"] = details["activityArn"]968    if resource_type == "ApiGateway::Model":969        resource["PhysicalResourceId"] = details["id"]970    if resource_type == "EC2::VPC":971        resource["PhysicalResourceId"] = details["Vpc"]["VpcId"]972    if resource_type == "EC2::Subnet":973        resource["PhysicalResourceId"] = details["Subnet"]["SubnetId"]974    if resource_type == "EC2::RouteTable":975        resource["PhysicalResourceId"] = details["RouteTable"]["RouteTableId"]976    if resource_type == "EC2::Route":977        resource["PhysicalResourceId"] = generate_route_id(978            resource_props["RouteTableId"],979            resource_props.get("DestinationCidrBlock", ""),980            resource_props.get("DestinationIpv6CidrBlock"),981        )982def add_default_resource_props(983    resource,984    stack_name,985    resource_name=None,986    resource_id=None,987    update=False,988    existing_resources=None,989):990    """Apply some fixes to resource props which otherwise cause deployments to fail"""991    res_type = resource["Type"]992    canonical_type = canonical_resource_type(res_type)993    resource_class = RESOURCE_MODELS.get(canonical_type)994    if resource_class is not None:995        resource_class.add_defaults(resource, stack_name)996# -----------------------997# MAIN TEMPLATE DEPLOYER998# -----------------------999class TemplateDeployer(object):1000    def __init__(self, stack):1001        self.stack = stack1002    @property1003    def resources(self):1004        return self.stack.resources1005    @property1006    def stack_name(self):1007        return self.stack.stack_name1008    # ------------------1009    # MAIN ENTRY POINTS1010    # ------------------1011    def deploy_stack(self):1012        self.stack.set_stack_status("CREATE_IN_PROGRESS")1013        try:1014            self.apply_changes(1015                self.stack,1016                self.stack,1017                stack_name=self.stack.stack_name,1018                initialize=True,1019                action="CREATE",1020            )1021        except Exception as e:1022            LOG.info("Unable to create stack %s: %s", self.stack.stack_name, e)1023            self.stack.set_stack_status("CREATE_FAILED")1024            raise1025    def apply_change_set(self, change_set):1026        action = "CREATE"1027        change_set.stack.set_stack_status("%s_IN_PROGRESS" % action)1028        try:1029            self.apply_changes(1030                change_set.stack,1031                change_set,1032                stack_name=change_set.stack_name,1033                action=action,1034            )1035        except Exception as e:1036            LOG.info(1037                "Unable to apply change set %s: %s", change_set.metadata.get("ChangeSetName"), e1038            )1039            change_set.metadata["Status"] = "%s_FAILED" % action1040            self.stack.set_stack_status("%s_FAILED" % action)1041            raise1042    def update_stack(self, new_stack):1043        self.stack.set_stack_status("UPDATE_IN_PROGRESS")1044        # apply changes1045        self.apply_changes(self.stack, new_stack, stack_name=self.stack.stack_name, action="UPDATE")1046    def delete_stack(self):1047        if not self.stack:1048            return1049        self.stack.set_stack_status("DELETE_IN_PROGRESS")1050        stack_resources = list(self.stack.resources.values())1051        resources = {r["LogicalResourceId"]: clone_safe(r) for r in stack_resources}1052        for key, resource in resources.items():1053            resource["Properties"] = resource.get("Properties", clone_safe(resource))1054            resource["ResourceType"] = resource.get("ResourceType") or resource.get("Type")1055        for resource_id, resource in resources.items():1056            # TODO: cache condition value in resource details on deployment and use cached value here1057            if evaluate_resource_condition(self, resource):1058                delete_resource(self, resource_id)1059                self.stack.set_resource_status(resource_id, "DELETE_COMPLETE")1060        # update status1061        self.stack.set_stack_status("DELETE_COMPLETE")1062    # ----------------------------1063    # DEPENDENCY RESOLUTION UTILS1064    # ----------------------------1065    def is_deployable_resource(self, resource):1066        resource_type = get_resource_type(resource)1067        entry = get_deployment_config(resource_type)1068        if entry is None and resource_type not in ["Parameter", None]:1069            resource_str = dump_resource_as_json(resource)1070            LOG.warning(f'Unable to deploy resource type "{resource_type}": {resource_str}')1071        return bool(entry and entry.get(ACTION_CREATE))1072    def is_deployed(self, resource):1073        resource_status = {}1074        resource_id = resource["LogicalResourceId"]1075        details = retrieve_resource_details(resource_id, resource_status, stack=self.stack)1076        return bool(details)1077    def is_updateable(self, resource):1078        """Return whether the given resource can be updated or not."""1079        if not self.is_deployable_resource(resource) or not self.is_deployed(resource):1080            return False1081        resource_type = get_resource_type(resource)1082        return resource_type in UPDATEABLE_RESOURCES1083    def all_resource_dependencies_satisfied(self, 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.