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:...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!