diff --git a/localstack_snapshot/snapshots/prototype.py b/localstack_snapshot/snapshots/prototype.py index 533f285..3a579fb 100644 --- a/localstack_snapshot/snapshots/prototype.py +++ b/localstack_snapshot/snapshots/prototype.py @@ -25,6 +25,8 @@ SNAPSHOT_LOGGER = logging.getLogger(__name__) SNAPSHOT_LOGGER.setLevel(logging.DEBUG if os.environ.get("DEBUG_SNAPSHOT") else logging.WARNING) +_PLACEHOLDER_VALUE = "$__marker__$" + class SnapshotMatchResult: def __init__(self, a: dict, b: dict, key: str = ""): @@ -218,7 +220,7 @@ def _assert_all( self.skip_verification_paths = skip_verification_paths or [] if skip_verification_paths: SNAPSHOT_LOGGER.warning( - f"Snapshot verification disabled for paths: {skip_verification_paths}" + "Snapshot verification disabled for paths: %s", skip_verification_paths ) if self.update: @@ -306,7 +308,7 @@ def _transform(self, tmp: dict) -> dict: try: replaced_tmp[key] = json.loads(dumped_value) except JSONDecodeError: - SNAPSHOT_LOGGER.error(f"could not decode json-string:\n{tmp}") + SNAPSHOT_LOGGER.error("could not decode json-string:\n%s", tmp) return {} return replaced_tmp @@ -365,6 +367,21 @@ def build_full_path_nodes(field_match: DatumInContext): return full_path_nodes[::-1][1:] # reverse the list and remove Root()/$ + def _remove_placeholder(_tmp): + """Traverse the object and remove any values in a list that would be equal to the placeholder""" + if isinstance(_tmp, dict): + for k, v in _tmp.items(): + if isinstance(v, dict): + _remove_placeholder(v) + elif isinstance(v, list): + _tmp[k] = _remove_placeholder(v) + elif isinstance(_tmp, list): + return [_remove_placeholder(item) for item in _tmp if item != _PLACEHOLDER_VALUE] + + return _tmp + + has_placeholder = False + for path in self.skip_verification_paths: matches = parse(path).find(tmp) or [] for m in matches: @@ -378,7 +395,24 @@ def build_full_path_nodes(field_match: DatumInContext): helper = helper.get(p, None) if not helper: continue + if ( isinstance(helper, dict) and full_path[-1] in helper.keys() ): # might have been deleted already del helper[full_path[-1]] + elif isinstance(helper, list): + try: + index = int(full_path[-1].lstrip("[").rstrip("]")) + # we need to set a placeholder value as the skips are based on index + # if we are to pop the values, the next skip index will have shifted and won't be correct + helper[index] = _PLACEHOLDER_VALUE + has_placeholder = True + except ValueError: + SNAPSHOT_LOGGER.warning( + "Snapshot skip path '%s' was not applied as it was invalid for that snapshot", + path, + exc_info=SNAPSHOT_LOGGER.isEnabledFor(logging.DEBUG), + ) + + if has_placeholder: + _remove_placeholder(tmp) diff --git a/tests/test_snapshots.py b/tests/test_snapshots.py index e4436fe..d74e366 100644 --- a/tests/test_snapshots.py +++ b/tests/test_snapshots.py @@ -191,6 +191,137 @@ def test_non_homogeneous_list(self): sm.match("key1", [{"key2": "value1"}, "value2", 3]) sm._assert_all() + def test_list_as_last_node_in_skip_verification_path(self): + sm = SnapshotSession(scope_key="A", verify=True, base_file_path="", update=False) + sm.recorded_state = {"key_a": {"aaa": ["item1", "item2", "item3"]}} + sm.match( + "key_a", + {"aaa": ["item1", "different-value"]}, + ) + + with pytest.raises(Exception) as ctx: # asserts it fail without skipping + sm._assert_all() + ctx.match("Parity snapshot failed") + + skip_path = ["$..aaa[1]", "$..aaa[2]"] + sm._assert_all(skip_verification_paths=skip_path) + + skip_path = ["$..aaa.1", "$..aaa.2"] + sm._assert_all(skip_verification_paths=skip_path) + + def test_list_as_last_node_in_skip_verification_path_complex(self): + sm = SnapshotSession(scope_key="A", verify=True, base_file_path="", update=False) + sm.recorded_state = { + "key_a": { + "aaa": [ + {"aab": ["aac", "aad"]}, + {"aab": ["aac", "aad"]}, + {"aab": ["aac", "aad"]}, + ] + } + } + sm.match( + "key_a", + { + "aaa": [ + {"aab": ["aac", "bad-value"], "bbb": "value"}, + {"aab": ["aac", "aad", "bad-value"]}, + {"aab": ["bad-value", "aad"]}, + ] + }, + ) + + with pytest.raises(Exception) as ctx: # asserts it fail without skipping + sm._assert_all() + ctx.match("Parity snapshot failed") + + skip_path = [ + "$..aaa[0].aab[1]", + "$..aaa[0].bbb", + "$..aaa[1].aab[2]", + "$..aaa[2].aab[0]", + ] + sm._assert_all(skip_verification_paths=skip_path) + + skip_path = [ + "$..aaa.0..aab.1", + "$..aaa.0..bbb", + "$..aaa.1..aab.2", + "$..aaa.2..aab.0", + ] + sm._assert_all(skip_verification_paths=skip_path) + + def test_list_as_mid_node_in_skip_verification_path(self): + sm = SnapshotSession(scope_key="A", verify=True, base_file_path="", update=False) + sm.recorded_state = {"key_a": {"aaa": [{"aab": "value1"}, {"aab": "value2"}]}} + sm.match( + "key_a", + {"aaa": [{"aab": "value1"}, {"aab": "bad-value"}]}, + ) + + with pytest.raises(Exception) as ctx: # asserts it fail without skipping + sm._assert_all() + ctx.match("Parity snapshot failed") + + skip_path = ["$..aaa[1].aab"] + sm._assert_all(skip_verification_paths=skip_path) + + skip_path = ["$..aaa.1.aab"] + sm._assert_all(skip_verification_paths=skip_path) + + def test_list_as_last_node_in_skip_verification_path_nested(self): + sm = SnapshotSession(scope_key="A", verify=True, base_file_path="", update=False) + sm.recorded_state = { + "key_a": { + "aaa": [ + "bbb", + "ccc", + [ + "ddd", + "eee", + [ + "fff", + "ggg", + ], + ], + ] + } + } + sm.match( + "key_a", + { + "aaa": [ + "bbb", + "ccc", + [ + "bad-value", + "eee", + [ + "fff", + "ggg", + ], + ], + ] + }, + ) + + with pytest.raises(Exception) as ctx: # asserts it fail without skipping + sm._assert_all() + ctx.match("Parity snapshot failed") + + skip_path = ["$..aaa[2][0]"] + sm._assert_all(skip_verification_paths=skip_path) + + skip_path = ["$..aaa.2[0]"] + sm._assert_all(skip_verification_paths=skip_path) + + # these 2 will actually skip almost everything, as they will match every first element of any list inside `aaa` + skip_path = ["$..aaa..[0]"] + sm._assert_all(skip_verification_paths=skip_path) + + skip_path = ["$..aaa..0"] + sm._assert_all(skip_verification_paths=skip_path) + def test_json_diff_format(): path = ["Records", 1]