Support jsonpath-ng 1.9 and fix skip paths with special characters

.github/workflows/build.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: [ "3.10", "3.11", "3.12" ]
+        python-version: [ "3.11", "3.12" ]
     steps:
       - name: Checkout
         uses: actions/checkout@v3

localstack_snapshot/snapshots/prototype.py

@@ -11,7 +11,7 @@
 from typing import Dict, List, Optional
 
 from deepdiff import DeepDiff
-from jsonpath_ng import DatumInContext
+from jsonpath_ng import Fields, Index
 from jsonpath_ng.ext import parse
 
 from localstack_snapshot.snapshots.transformer import (
@@ -377,16 +377,6 @@ def replace_value(self, pattern: Pattern[str], value: str):
     def _remove_skip_verification_paths(self, tmp: Dict):
         """Removes all keys from the dict, that match the given json-paths in self.skip_verification_path"""
 
-        def build_full_path_nodes(field_match: DatumInContext):
-            """Traverse the matched Datum to build the path field by field"""
-            full_path_nodes = [str(field_match.path).replace("'", "")]
-            next_node = field_match
-            while next_node.context is not None:
-                full_path_nodes.append(str(next_node.context.path))
-                next_node = next_node.context
-
-            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):
@@ -407,34 +397,22 @@ def _remove_placeholder(_tmp):
         for path in self.skip_verification_paths:
             matches = parse(path).find(tmp) or []
             for m in matches:
-                full_path = build_full_path_nodes(m)
-                helper = tmp
-                if len(full_path) > 1:
-                    for p in full_path[:-1]:
-                        if isinstance(helper, list) and p.lstrip("[").rstrip("]").isnumeric():
-                            helper = helper[int(p.lstrip("[").rstrip("]"))]
-                        elif isinstance(helper, dict):
-                            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("]"))
+                # the context holds a reference to the container of the matched value, we can mutate it directly
+                parent = m.context.value if m.context is not None else None
+                if isinstance(parent, dict) and isinstance(m.path, Fields):
+                    for key in m.path.fields:
+                        parent.pop(key, None)  # might have been deleted already
+                elif isinstance(parent, list) and isinstance(m.path, Index):
+                    for index in m.path.indices:
                         # 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] = _SKIP_PLACEHOLDER_VALUE
+                        parent[index] = _SKIP_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),
-                        )
+                elif isinstance(parent, (dict, list)):
+                    SNAPSHOT_LOGGER.warning(
+                        "Snapshot skip path '%s' was not applied as it was invalid for that snapshot",
+                        path,
+                    )
 
         if has_placeholder:
             _remove_placeholder(tmp)

pyproject.toml

@@ -10,16 +10,15 @@ authors = [
 version = "0.3.5"
 description = "Extracted snapshot testing lib for LocalStack"
 dependencies = [
-    "jsonpath-ng>1.6",
+    "jsonpath-ng>=1.9",
     "deepdiff",
 ]
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 license = {file = "LICENSE"}
 classifiers = [
     "License :: OSI Approved :: Apache Software License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
 ]
@@ -53,8 +52,8 @@ include = '((localstack_snapshot)/.*\.py$|tests/.*\.py$)'
 #extend_exclude = '()'
 
 [tool.ruff]
-# Always generate Python 3.10-compatible code.
-target-version = "py310"
+# Always generate Python 3.11-compatible code.
+target-version = "py311"
 line-length = 110
 exclude = [
     ".venv*",

tests/test_snapshots.py

@@ -1,3 +1,4 @@
+import copy
 import io
 from enum import Enum
 
@@ -268,48 +269,44 @@ 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):
+    # skipping mutates the recorded and observed state, so every assertion needs a new session
+
+    @staticmethod
+    def _new_session(recorded: dict, observed: dict) -> SnapshotSession:
         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"]},
-        )
+        sm.recorded_state = {"key_a": copy.deepcopy(recorded)}
+        sm.match("key_a", copy.deepcopy(observed))
+        return sm
+
+    def test_list_as_last_node_in_skip_verification_path(self):
+        recorded = {"aaa": ["item1", "item2", "item3"]}
+        observed = {"aaa": ["item1", "different-value"]}
 
         with pytest.raises(Exception) as ctx:  # asserts it fail without skipping
-            sm._assert_all()
+            self._new_session(recorded, observed)._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)
+        self._new_session(recorded, observed)._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"]},
-                ]
-            }
+        recorded = {
+            "aaa": [
+                {"aab": ["aac", "aad"]},
+                {"aab": ["aac", "aad"]},
+                {"aab": ["aac", "aad"]},
+            ]
+        }
+        observed = {
+            "aaa": [
+                {"aab": ["aac", "bad-value"], "bbb": "value"},
+                {"aab": ["aac", "aad", "bad-value"]},
+                {"aab": ["bad-value", "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()
+            self._new_session(recorded, observed)._assert_all()
         ctx.match("Parity snapshot failed")
 
         skip_path = [
@@ -318,86 +315,75 @@ def test_list_as_last_node_in_skip_verification_path_complex(self):
             "$..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)
+        self._new_session(recorded, observed)._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"}]},
-        )
+        recorded = {"aaa": [{"aab": "value1"}, {"aab": "value2"}]}
+        observed = {"aaa": [{"aab": "value1"}, {"aab": "bad-value"}]}
 
         with pytest.raises(Exception) as ctx:  # asserts it fail without skipping
-            sm._assert_all()
+            self._new_session(recorded, observed)._assert_all()
         ctx.match("Parity snapshot failed")
 
         skip_path = ["$..aaa[1].aab"]
-        sm._assert_all(skip_verification_paths=skip_path)
+        self._new_session(recorded, observed)._assert_all(skip_verification_paths=skip_path)
+
+    @pytest.mark.parametrize(
+        "skip_path",
+        [
+            ["$..aaa[2][0]"],
+            # this will actually skip almost everything, as it will match every first element of any list inside `aaa`
+            ["$..aaa..[0]"],
+        ],
+    )
+    def test_list_as_last_node_in_skip_verification_path_nested(self, skip_path):
+        recorded = {"aaa": ["bbb", "ccc", ["ddd", "eee", ["fff", "ggg"]]]}
+        observed = {"aaa": ["bbb", "ccc", ["bad-value", "eee", ["fff", "ggg"]]]}
 
-        skip_path = ["$..aaa.1.aab"]
-        sm._assert_all(skip_verification_paths=skip_path)
+        with pytest.raises(Exception) as ctx:  # asserts it fail without skipping
+            self._new_session(recorded, observed)._assert_all()
+        ctx.match("Parity snapshot failed")
 
-    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",
-                        ],
-                    ],
-                ]
+        self._new_session(recorded, observed)._assert_all(skip_verification_paths=skip_path)
+
+    @pytest.mark.parametrize(
+        "key",
+        ["with space", "Key%", "00", "10", "a.b", "aws:SourceArn", "Ünïcode", "x-amz-id"],
+    )
+    def test_special_characters_in_skip_verification_path(self, key):
+        def _new_session():
+            # use a new session for every assertion, as skipping mutates the recorded and observed state
+            sm = SnapshotSession(scope_key="A", verify=True, base_file_path="", update=False)
+            sm.recorded_state = {
+                "key_a": {key: {"Inner": [{key: "value", "other": "same"}]}, "aaa": "same"}
             }
-        }
-        sm.match(
-            "key_a",
-            {
-                "aaa": [
-                    "bbb",
-                    "ccc",
-                    [
-                        "bad-value",
-                        "eee",
-                        [
-                            "fff",
-                            "ggg",
-                        ],
-                    ],
-                ]
-            },
-        )
+            sm.match(
+                "key_a", {key: {"Inner": [{key: "bad-value", "other": "same"}]}, "aaa": "same"}
+            )
+            return sm
 
         with pytest.raises(Exception) as ctx:  # asserts it fail without skipping
-            sm._assert_all()
+            _new_session()._assert_all()
         ctx.match("Parity snapshot failed")
 
-        skip_path = ["$..aaa[2][0]"]
-        sm._assert_all(skip_verification_paths=skip_path)
+        # the key is used as an intermediate node as well as the last node
+        _new_session()._assert_all(skip_verification_paths=[f"$..'{key}'.Inner[0].'{key}'"])
+        _new_session()._assert_all(skip_verification_paths=[f"$..Inner[*].'{key}'"])
+        _new_session()._assert_all(skip_verification_paths=[f"$..'{key}'"])
 
-        skip_path = ["$..aaa.2[0]"]
-        sm._assert_all(skip_verification_paths=skip_path)
+    def test_skip_verification_path_with_filter(self):
+        def _new_session():
+            sm = SnapshotSession(scope_key="A", verify=True, base_file_path="", update=False)
+            sm.recorded_state = {"key_a": {"aaa": [{"id": "1", "v": "a"}, {"id": "2", "v": "b"}]}}
+            sm.match("key_a", {"aaa": [{"id": "1", "v": "a"}, {"id": "2", "v": "bad-value"}]})
+            return sm
 
-        # 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)
+        with pytest.raises(Exception) as ctx:  # asserts it fail without skipping
+            _new_session()._assert_all()
+        ctx.match("Parity snapshot failed")
 
-        skip_path = ["$..aaa..0"]
-        sm._assert_all(skip_verification_paths=skip_path)
+        _new_session()._assert_all(skip_verification_paths=["$..aaa[?(@.id == '2')].v"])
+        _new_session()._assert_all(skip_verification_paths=["$..aaa[?(@.id == '2')]"])
 
 
 def test_json_diff_format():