support skipping last node by index if list

Author: bentsku

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)

tests/test_snapshots.py

@@ -191,6 +191,80 @@ 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_json_diff_format():
     path = ["Records", 1]