Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use _resolve_refs_recursively method in localstack

Best Python code snippet using localstack_python

template_deployer.py

Source:template_deployer.py

...408# 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":...

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.