Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import pytest
from kopf.structs.diffs import reduce, Diff, DiffItem, DiffOperation
DIFF = Diff([
DiffItem(DiffOperation.ADD , ('key1',), None, 'new1'),
DiffItem(DiffOperation.CHANGE, ('key2',), 'old2', 'new2'),
DiffItem(DiffOperation.ADD , ('key2', 'suba'), 'olda', 'newa'),
DiffItem(DiffOperation.REMOVE, ('key2', 'subb'), 'oldb', 'newb'),
DiffItem(DiffOperation.REMOVE, ('key3',), 'old3', None),
])
@pytest.mark.parametrize('diff', [
[['op', ['key', 'sub'], 'old', 'new']],
[['op', ('key', 'sub'), 'old', 'new']],
[('op', ['key', 'sub'], 'old', 'new')],
[('op', ('key', 'sub'), 'old', 'new')],
(['op', ['key', 'sub'], 'old', 'new'],),
(['op', ('key', 'sub'), 'old', 'new'],),
(('op', ['key', 'sub'], 'old', 'new'),),
(('op', ('key', 'sub'), 'old', 'new'),),
], ids=[
'lll-diff', 'llt-diff', 'ltl-diff', 'ltt-diff',
import pytest
from kopf.structs.diffs import reduce, Diff, DiffItem, DiffOperation
DIFF = Diff([
DiffItem(DiffOperation.ADD , ('key1',), None, 'new1'),
DiffItem(DiffOperation.CHANGE, ('key2',), 'old2', 'new2'),
DiffItem(DiffOperation.ADD , ('key2', 'suba'), 'olda', 'newa'),
DiffItem(DiffOperation.REMOVE, ('key2', 'subb'), 'oldb', 'newb'),
DiffItem(DiffOperation.REMOVE, ('key3',), 'old3', None),
])
@pytest.mark.parametrize('diff', [
[['op', ['key', 'sub'], 'old', 'new']],
[['op', ('key', 'sub'), 'old', 'new']],
[('op', ['key', 'sub'], 'old', 'new')],
[('op', ('key', 'sub'), 'old', 'new')],
(['op', ['key', 'sub'], 'old', 'new'],),
(['op', ('key', 'sub'), 'old', 'new'],),
(('op', ['key', 'sub'], 'old', 'new'),),
(('op', ('key', 'sub'), 'old', 'new'),),
def reduce_iter(
d: Diff,
path: dicts.FieldPath,
) -> Iterator[DiffItem]:
for op, field, old, new in d:
# As-is diff (i.e. a root field).
if not path:
yield DiffItem(op, tuple(field), old, new)
# The diff-field is longer than the path: get "spec.struct" when "spec.struct.field" is set.
# Retranslate the diff with the field prefix shrinked.
elif tuple(field[:len(path)]) == tuple(path):
yield DiffItem(op, tuple(field[len(path):]), old, new)
# The diff-field is shorter than the path: get "spec.struct" when "spec={...}" is added.
# Generate a new diff, with new ops, for the resolved sub-field.
elif tuple(field) == tuple(path[:len(field)]):
tail = path[len(field):]
old_tail = dicts.resolve(old, tail, default=None, assume_empty=True)
new_tail = dicts.resolve(new, tail, default=None, assume_empty=True)
yield from diff_iter(old_tail, new_tail)
and the ``old`` & ``new`` values (`None` for addition/removal).
List values are treated as a whole, and not recursed into.
Therefore, an addition/removal of a list item is considered
as a change of the whole value.
If the deep diff for lists/sets is needed, see the libraries:
* https://dictdiffer.readthedocs.io/en/latest/
* https://github.com/seperman/deepdiff
* https://python-json-patch.readthedocs.io/en/latest/tutorial.html
"""
if a == b: # incl. cases when both are None
pass
elif a is None:
yield DiffItem(DiffOperation.ADD, path, a, b)
elif b is None:
yield DiffItem(DiffOperation.REMOVE, path, a, b)
elif type(a) != type(b):
yield DiffItem(DiffOperation.CHANGE, path, a, b)
elif isinstance(a, collections.abc.Mapping):
a_keys = frozenset(a.keys())
b_keys = frozenset(b.keys())
for key in b_keys - a_keys:
yield from diff_iter(None, b[key], path=path+(key,))
for key in a_keys - b_keys:
yield from diff_iter(a[key], None, path=path+(key,))
for key in a_keys & b_keys:
yield from diff_iter(a[key], b[key], path=path+(key,))
else:
yield DiffItem(DiffOperation.CHANGE, path, a, b)
List values are treated as a whole, and not recursed into.
Therefore, an addition/removal of a list item is considered
as a change of the whole value.
If the deep diff for lists/sets is needed, see the libraries:
* https://dictdiffer.readthedocs.io/en/latest/
* https://github.com/seperman/deepdiff
* https://python-json-patch.readthedocs.io/en/latest/tutorial.html
"""
if a == b: # incl. cases when both are None
pass
elif a is None:
yield DiffItem(DiffOperation.ADD, path, a, b)
elif b is None:
yield DiffItem(DiffOperation.REMOVE, path, a, b)
elif type(a) != type(b):
yield DiffItem(DiffOperation.CHANGE, path, a, b)
elif isinstance(a, collections.abc.Mapping):
a_keys = frozenset(a.keys())
b_keys = frozenset(b.keys())
for key in b_keys - a_keys:
yield from diff_iter(None, b[key], path=path+(key,))
for key in a_keys - b_keys:
yield from diff_iter(a[key], None, path=path+(key,))
for key in a_keys & b_keys:
yield from diff_iter(a[key], b[key], path=path+(key,))
else:
yield DiffItem(DiffOperation.CHANGE, path, a, b)
yield DiffItem(DiffOperation.ADD, path, a, b)
elif b is None:
yield DiffItem(DiffOperation.REMOVE, path, a, b)
elif type(a) != type(b):
yield DiffItem(DiffOperation.CHANGE, path, a, b)
elif isinstance(a, collections.abc.Mapping):
a_keys = frozenset(a.keys())
b_keys = frozenset(b.keys())
for key in b_keys - a_keys:
yield from diff_iter(None, b[key], path=path+(key,))
for key in a_keys - b_keys:
yield from diff_iter(a[key], None, path=path+(key,))
for key in a_keys & b_keys:
yield from diff_iter(a[key], b[key], path=path+(key,))
else:
yield DiffItem(DiffOperation.CHANGE, path, a, b)