Skip to content

Support dot notation for list indices in skip verification paths #28

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
bentsku wants to merge 1 commit into from
Closed

Support dot notation for list indices in skip verification paths #28

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 33 additions & 2 deletions localstack_snapshot/snapshots/prototype.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from typing import Dict, List, Optional

from deepdiff import DeepDiff
from jsonpath_ng import Fields, Index
from jsonpath_ng import DatumInContext, Fields, Index, JSONPath
from jsonpath_ng.ext import parse

from localstack_snapshot.snapshots.transformer import (
Expand All @@ -30,6 +30,37 @@
_SKIP_PLACEHOLDER_VALUE = "$__to_be_skipped__$"


class _FieldOrIndex(Fields):
"""
Numeric field, which is used as a list index if the datum is a list, e.g. `$..aaa.1`.
jsonpath-ng only matches `Fields` against dicts, and `$.a.1` and `$.a.'1'` are parsed identically,
so the decision needs to happen when matching the datum.
"""

def find(self, datum):
datum = DatumInContext.wrap(datum)
if isinstance(datum.value, list):
return Index(int(self.fields[0])).find(datum)
return super().find(datum)


def _parse_skip_verification_path(path: str) -> JSONPath:
"""
Parses the jsonpath, and allows the dot notation for list indices.
Note: consecutive indices like `$..aaa.2.0` cannot be parsed, as `2.0` is lexed as a float. Use `$..aaa.2[0]` instead.
"""

def _replace_numeric_fields(node):
if isinstance(node, Fields) and len(node.fields) == 1 and node.fields[0].isdecimal():
return _FieldOrIndex(*node.fields)
for attr, value in vars(node).items():
if isinstance(value, JSONPath):
setattr(node, attr, _replace_numeric_fields(value))
return node

return _replace_numeric_fields(parse(path))


class SnapshotMatchResult:
def __init__(self, a: dict, b: dict, key: str = ""):
self.a = a
Expand Down Expand Up @@ -395,7 +426,7 @@ def _remove_placeholder(_tmp):
has_placeholder = False

for path in self.skip_verification_paths:
matches = parse(path).find(tmp) or []
matches = _parse_skip_verification_path(path).find(tmp) or []
for m in matches:
# 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
Expand Down
200 changes: 93 additions & 107 deletions tests/test_snapshots.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import copy
import io
from enum import Enum

Expand Down Expand Up @@ -268,136 +269,121 @@ 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

@pytest.mark.parametrize(
"skip_path",
[["$..aaa[1]", "$..aaa[2]"], ["$..aaa.1", "$..aaa.2"]],
ids=["brackets", "dot"],
)
def test_list_as_last_node_in_skip_verification_path(self, skip_path):
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"]},
]
}
@pytest.mark.parametrize(
"skip_path",
[
[
"$..aaa[0].aab[1]",
"$..aaa[0].bbb",
"$..aaa[1].aab[2]",
"$..aaa[2].aab[0]",
],
[
"$..aaa.0..aab.1",
"$..aaa.0..bbb",
"$..aaa.1..aab.2",
"$..aaa.2..aab.0",
],
],
ids=["brackets", "dot"],
)
def test_list_as_last_node_in_skip_verification_path_complex(self, skip_path):
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 = [
"$..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"}]},
)
self._new_session(recorded, observed)._assert_all(skip_verification_paths=skip_path)

@pytest.mark.parametrize(
"skip_path", [["$..aaa[1].aab"], ["$..aaa.1.aab"]], ids=["brackets", "dot"]
)
def test_list_as_mid_node_in_skip_verification_path(self, skip_path):
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)

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",
],
],
]
},
)
@pytest.mark.parametrize(
"skip_path",
[
["$..aaa[2][0]"],
["$..aaa.2[0]"],
# these 2 will actually skip almost everything, as they will match every first element of any list inside `aaa`
["$..aaa..[0]"],
["$..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"]]]}

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[2][0]"]
sm._assert_all(skip_verification_paths=skip_path)
self._new_session(recorded, observed)._assert_all(skip_verification_paths=skip_path)

skip_path = ["$..aaa.2[0]"]
sm._assert_all(skip_verification_paths=skip_path)
@pytest.mark.parametrize("skip_path", [["$..aaa.10"], ["$..aaa.'10'"], ["$..'10'"]])
def test_numeric_dict_key_in_skip_verification_path(self, skip_path):
recorded = {"aaa": {"10": "value", "1": "same"}, "bbb": ["same", "same"]}
observed = {"aaa": {"10": "bad-value", "1": "same"}, "bbb": ["same", "same"]}

# 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
self._new_session(recorded, observed)._assert_all()
ctx.match("Parity snapshot failed")

self._new_session(recorded, observed)._assert_all(skip_verification_paths=skip_path)

skip_path = ["$..aaa..0"]
sm._assert_all(skip_verification_paths=skip_path)
def test_dot_index_out_of_range_in_skip_verification_path(self):
recorded = {"aaa": ["item1"]}
observed = {"aaa": ["bad-value"]}

with pytest.raises(Exception) as ctx: # asserts the skip path does not match anything
self._new_session(recorded, observed)._assert_all(skip_verification_paths=["$..aaa.5"])
ctx.match("Parity snapshot failed")

@pytest.mark.parametrize(
"key",
Expand Down
Loading