How to use _register_serialized_reference_replacement method in localstack

Best Python code snippet using localstack_python

transformer.py

Source:transformer.py

...26        """retrieve new enumeration value for a given scope key (e.g. for tokens such as <fn-name:1>"""27        current_counter = self.scoped_tokens.setdefault(scope, 1)28        self.scoped_tokens[scope] += 129        return current_counter30def _register_serialized_reference_replacement(31    transform_context: TransformContext, *, reference_value: str, replacement: str32):33    if '"' in reference_value:34        reference_value = reference_value.replace('"', '\\"')35    cache = transform_context._cache.setdefault("regexcache", set())36    cache_key = reference_value37    if cache_key not in cache:38        actual_replacement = f"<{replacement}:{transform_context.new_scope(replacement)}>"39        cache.add(cache_key)40        def _helper(bound_result, bound_replacement):41            def replace_val(s):42                SNAPSHOT_LOGGER.debug(43                    f"Replacing '{bound_result}' in snapshot with '{bound_replacement}'"44                )45                return s.replace(bound_result, bound_replacement, -1)46            return replace_val47        SNAPSHOT_LOGGER.debug(48            f"Registering reference replacement for value: '{reference_value}' -> '{actual_replacement}'"49        )50        transform_context.register_serialized_replacement(51            _helper(reference_value, actual_replacement)52        )53class Transformer(Protocol):54    def transform(self, input_data: dict, *, ctx: TransformContext) -> dict:55        ...56# Transformers57class ResponseMetaDataTransformer:58    def transform(self, input_data: dict, *, ctx: TransformContext) -> dict:59        for k, v in input_data.items():60            if k == "ResponseMetadata":61                metadata = v62                http_headers = metadata.get("HTTPHeaders")63                # TODO "x-amz-bucket-region"64                # TestS3.test_region_header_exists -> verifies bucket-region65                headers_to_collect = ["content_type"]66                simplified_headers = {}67                for h in headers_to_collect:68                    if http_headers.get(h):69                        simplified_headers[h] = http_headers[h]70                simplified_metadata = {71                    "HTTPHeaders": simplified_headers,72                }73                # HTTPStatusCode might be removed for marker skip_snapshot_verify74                if status_code := metadata.get("HTTPStatusCode"):75                    simplified_metadata["HTTPStatusCode"] = status_code76                input_data[k] = simplified_metadata77            elif isinstance(v, dict):78                input_data[k] = self.transform(v, ctx=ctx)79        return input_data80class JsonpathTransformer:81    def __init__(self, jsonpath: str, replacement: str, replace_reference: bool = True) -> None:82        self.jsonpath = jsonpath83        self.replacement = replacement84        self.replace_references = replace_reference85    def transform(self, input_data: dict, *, ctx: TransformContext) -> dict:86        pattern = parse(self.jsonpath)87        if self.replace_references:88            res = pattern.find(input_data)89            if not res:90                SNAPSHOT_LOGGER.debug(f"No match found for JsonPath '{self.jsonpath}'")91                return input_data92            for r in res:93                value_to_replace = r.value94                _register_serialized_reference_replacement(95                    ctx, reference_value=value_to_replace, replacement=self.replacement96                )97        else:98            original = copy.deepcopy(input_data)99            pattern.update(input_data, self.replacement)100            if original != input_data:101                SNAPSHOT_LOGGER.debug(102                    f"Replacing JsonPath '{self.jsonpath}' in snapshot with '{self.replacement}'"103                )104            else:105                SNAPSHOT_LOGGER.debug(f"No match found for JsonPath '{self.jsonpath}'")106        return input_data107    def _add_jsonpath_replacement(self, jsonpath, replacement):108        self.json_path_replacement_list.append((jsonpath, replacement))109class RegexTransformer:110    def __init__(self, regex: str | Pattern[str], replacement: str):111        self.regex = regex112        self.replacement = replacement113    def transform(self, input_data: dict, *, ctx: TransformContext) -> dict:114        compiled_regex = re.compile(self.regex) if isinstance(self.regex, str) else self.regex115        def _regex_replacer_helper(pattern: Pattern[str], repl: str):116            def replace_val(s):117                result = re.sub(pattern, repl, s)118                if result != s:119                    SNAPSHOT_LOGGER.debug(f"Replacing regex '{pattern.pattern}' with '{repl}'")120                else:121                    SNAPSHOT_LOGGER.debug(f"No match found for regex '{pattern.pattern}'")122                return result123            return replace_val124        ctx.register_serialized_replacement(125            _regex_replacer_helper(compiled_regex, self.replacement)126        )127        SNAPSHOT_LOGGER.debug(128            f"Registering regex pattern '{compiled_regex.pattern}' in snapshot with '{self.replacement}'"129        )130        return input_data131class KeyValueBasedTransformer:132    def __init__(133        self,134        match_fn: Callable[[str, str], Optional[str]],135        replacement: str,136        replace_reference: bool = True,137    ):138        self.match_fn = match_fn139        self.replacement = replacement140        self.replace_reference = replace_reference141    def transform(self, input_data: dict, *, ctx: TransformContext) -> dict:142        for k, v in input_data.items():143            if match_result := self.match_fn(k, v):144                if self.replace_reference:145                    _register_serialized_reference_replacement(146                        ctx, reference_value=match_result, replacement=self.replacement147                    )148                else:149                    SNAPSHOT_LOGGER.debug(150                        f"Replacing value for key '{k}' with '{self.replacement}'. (Original value: {str(v)})"151                    )152                    input_data[k] = self.replacement153            elif isinstance(v, list) and len(v) > 0 and isinstance(v[0], dict):154                for i in range(0, len(v)):155                    v[i] = self.transform(v[i], ctx=ctx)156            elif isinstance(v, dict):157                input_data[k] = self.transform(v, ctx=ctx)158        return input_data159class GenericTransformer:...

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.