From 0f5c39c7385d9765acafe9d8ee25b7aff9561d89 Mon Sep 17 00:00:00 2001 From: Daniel Fangl Date: Wed, 5 Mar 2025 18:57:40 +0100 Subject: [PATCH] Initial implementation of making key value replacements dynamic --- localstack_snapshot/snapshots/transformer.py | 31 ++++++++---- .../snapshots/transformer_utility.py | 30 +++++++++++- tests/test_transformer.py | 47 +++++++++++++++++++ 3 files changed, 99 insertions(+), 9 deletions(-) diff --git a/localstack_snapshot/snapshots/transformer.py b/localstack_snapshot/snapshots/transformer.py index 4c484e6..a0a81c4 100644 --- a/localstack_snapshot/snapshots/transformer.py +++ b/localstack_snapshot/snapshots/transformer.py @@ -184,35 +184,36 @@ def replace_val(s): return input_data -class KeyValueBasedTransformer: +class KeyValueBasedTransformerFunctionReplacement: def __init__( self, match_fn: Callable[[str, Any], Optional[str]], - replacement: str, + replacement_function: [Callable[[str, Any], str]], replace_reference: bool = True, ): self.match_fn = match_fn - self.replacement = replacement + self.replacement_function = replacement_function self.replace_reference = replace_reference def transform(self, input_data: dict, *, ctx: TransformContext) -> dict: for k, v in input_data.items(): if (match_result := self.match_fn(k, v)) is not None: + replacement = self.replacement_function(k, v) if self.replace_reference: _register_serialized_reference_replacement( - ctx, reference_value=match_result, replacement=self.replacement + ctx, reference_value=match_result, replacement=replacement ) else: if isinstance(v, str): SNAPSHOT_LOGGER.debug( - f"Replacing value for key '{k}': Match result '{match_result:.200s}' with '{self.replacement}'. (Original value: {str(v)})" + f"Replacing value for key '{k}': Match result '{match_result:.200s}' with '{replacement}'. (Original value: {str(v)})" ) - input_data[k] = v.replace(match_result, self.replacement) + input_data[k] = v.replace(match_result, replacement) else: SNAPSHOT_LOGGER.debug( - f"Replacing value for key '{k}' with '{self.replacement}'. (Original value: {str(v)})" + f"Replacing value for key '{k}' with '{replacement}'. (Original value: {str(v)})" ) - input_data[k] = self.replacement + input_data[k] = replacement elif isinstance(v, list) and len(v) > 0: for i in range(0, len(v)): if isinstance(v[i], dict): @@ -223,6 +224,20 @@ def transform(self, input_data: dict, *, ctx: TransformContext) -> dict: return input_data +class KeyValueBasedTransformer(KeyValueBasedTransformerFunctionReplacement): + def __init__( + self, + match_fn: Callable[[str, Any], Optional[str]], + replacement: str, + replace_reference: bool = True, + ): + super().__init__( + match_fn=match_fn, + replacement_function=lambda k, v: replacement, + replace_reference=replace_reference, + ) + + class GenericTransformer: def __init__(self, fn: Callable[[dict, TransformContext], dict]): self.fn = fn diff --git a/localstack_snapshot/snapshots/transformer_utility.py b/localstack_snapshot/snapshots/transformer_utility.py index a04e36d..7a46651 100644 --- a/localstack_snapshot/snapshots/transformer_utility.py +++ b/localstack_snapshot/snapshots/transformer_utility.py @@ -1,9 +1,10 @@ from re import Pattern -from typing import Optional +from typing import Any, Callable, Optional from localstack_snapshot.snapshots.transformer import ( JsonpathTransformer, KeyValueBasedTransformer, + KeyValueBasedTransformerFunctionReplacement, RegexTransformer, TextTransformer, ) @@ -38,6 +39,33 @@ def key_value( replace_reference=reference_replacement, ) + @staticmethod + def key_value_replacement_function( + key: str, + replacement_function: Callable[[str, Any], str] = None, + reference_replacement: bool = True, + ): + """Creates a new KeyValueBasedTransformer. If the key matches, the value will be replaced. + + :param key: the name of the key which should be replaced + :param replacement_function: The function calculating the replacement. Will be passed the key and value of the replaced pair. + By default it is the key-name in lowercase, separated with hyphen + :param reference_replacement: if False, only the original value for this key will be replaced. + If True all references of this value will be replaced (using a regex pattern), for the entire test case. + In this case, the replaced value will be nummerated as well. + Default: True + + :return: KeyValueBasedTransformer + """ + replacement_function = replacement_function or ( + lambda x, y: _replace_camel_string_with_hyphen(key) + ) + return KeyValueBasedTransformerFunctionReplacement( + lambda k, v: v if k == key and (v is not None and v != "") else None, + replacement_function=replacement_function, + replace_reference=reference_replacement, + ) + @staticmethod def jsonpath(jsonpath: str, value_replacement: str, reference_replacement: bool = True): """Creates a new JsonpathTransformer. If the jsonpath matches, the value will be replaced. diff --git a/tests/test_transformer.py b/tests/test_transformer.py index 62a708a..bd417a6 100644 --- a/tests/test_transformer.py +++ b/tests/test_transformer.py @@ -50,6 +50,53 @@ def test_key_value_replacement(self): assert json.loads(tmp) == expected_key_value_reference + def test_key_value_replacement_custom_function(self): + input = { + "hello": "world", + "hello2": "again", + "path": {"to": {"anotherkey": "hi", "inside": {"hello": "inside"}}}, + } + + key_value = TransformerUtility.key_value_replacement_function( + "hello", + replacement_function=lambda k, v: f"placeholder({len(v)})", + reference_replacement=False, + ) + + expected_key_value = { + "hello": "placeholder(5)", + "hello2": "again", + "path": {"to": {"anotherkey": "hi", "inside": {"hello": "placeholder(6)"}}}, + } + + copied = copy.deepcopy(input) + ctx = TransformContext() + assert key_value.transform(copied, ctx=ctx) == expected_key_value + assert ctx.serialized_replacements == [] + + copied = copy.deepcopy(input) + key_value = TransformerUtility.key_value_replacement_function( + "hello", + replacement_function=lambda k, v: f"placeholder({len(v)})", + reference_replacement=True, + ) + # replacement counters are per replacement key, so it will start from 1 again. + expected_key_value_reference = { + "hello": "", + "hello2": "again", + "path": { + "to": {"anotherkey": "hi", "": {"hello": ""}} + }, + } + assert key_value.transform(copied, ctx=ctx) == copied + assert len(ctx.serialized_replacements) == 2 + + tmp = json.dumps(copied, default=str) + for sr in ctx.serialized_replacements: + tmp = sr(tmp) + + assert json.loads(tmp) == expected_key_value_reference + def test_key_value_replacement_with_falsy_value(self): input = { "hello": "world",