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

Best Python code snippet using localstack_python
template_deployer.py
Source:template_deployer.py
...169        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:...
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.