Skip to content

support skipping last node by index if parent node is a list #16

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’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

support skipping last node by index if parent node is a list #16

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
38 changes: 36 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 @@ -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 = ""):
Expand Down Expand Up @@ -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
Copy link
Contributor Author

@bentsku bentsku Jul 2, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

see https://docs.astral.sh/ruff/rules/logging-f-string/
note: could also be added to the project rules as we did for LocalStack

)

if self.update:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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)
131 changes: 131 additions & 0 deletions tests/test_snapshots.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down