True diff'ing function for missing and extra elements.

jdiff/__init__.py

@@ -1,4 +1,5 @@
 """Pre/Post Check library."""
+
 from .check_types import CheckType
 from .extract_data import extract_data_from_json
 

jdiff/check_types.py

@@ -1,4 +1,5 @@
 """CheckType Implementation."""
+
 from typing import List, Tuple, Dict, Any, Union
 from abc import ABC, abstractmethod
 from .evaluators import diff_generator, parameter_evaluator, regex_evaluator, operator_evaluator

jdiff/evaluators.py

@@ -1,4 +1,5 @@
 """Evaluators."""
+
 import re
 from typing import Any, Mapping, Dict, Tuple, List
 from deepdiff import DeepDiff

jdiff/extract_data.py

@@ -1,13 +1,16 @@
 """Extract data from JSON. Based on custom JMSPath implementation."""
+
 import re
 import warnings
-from typing import Mapping, List, Dict, Any, Union, Optional
+from typing import Any, Dict, List, Mapping, Optional, Union
+
 import jmespath
+
 from .utils.data_normalization import exclude_filter, flatten_list
 from .utils.jmespath_parsers import (
-    jmespath_value_parser,
-    jmespath_refkey_parser,
     associate_key_of_my_value,
+    jmespath_refkey_parser,
+    jmespath_value_parser,
     keys_values_zipper,
     multi_reference_keys,
 )
@@ -48,7 +51,10 @@ def extract_data_from_json(data: Union[Mapping, List], path: str = "*", exclude:
     if len(re.findall(r"\$.*?\$", path)) > 1:
         clean_path = path.replace("$", "")
         values = jmespath.search(f"{clean_path}{' | []' * (path.count('*') - 1)}", data)
-        return keys_values_zipper(multi_reference_keys(path, data), associate_key_of_my_value(clean_path, values))
+        return keys_values_zipper(
+            multi_reference_keys(path, data),
+            associate_key_of_my_value(clean_path, values),
+        )
 
     values = jmespath.search(jmespath_value_parser(path), data)
 

jdiff/operator.py

@@ -1,4 +1,5 @@
 """Operator diff."""
+
 import operator
 from typing import Any, List, Tuple
 

jdiff/utils/data_normalization.py

@@ -1,4 +1,5 @@
 """Data Normalization utilities."""
+
 from typing import List, Generator, Union, Dict
 
 

jdiff/utils/diff_helpers.py

@@ -1,8 +1,10 @@
 """Diff helpers."""
+
 import re
 from collections import defaultdict
-from functools import partial
-from typing import Mapping, Dict, List, DefaultDict
+from functools import partial, reduce
+from operator import getitem
+from typing import DefaultDict, Dict, List, Mapping
 
 REGEX_PATTERN_RELEVANT_KEYS = r"'([A-Za-z0-9_\./\\-]*)'"
 
@@ -62,8 +64,12 @@ def fix_deepdiff_key_names(obj: Mapping) -> Dict:
     result = {}  # type: Dict
     for key, value in obj.items():
         key_parts = re.findall(REGEX_PATTERN_RELEVANT_KEYS, key)
-        if not key_parts:  # If key parts can't be find, keep original key so data is not lost.
-            key_parts = [key.replace("root", "index_element")]  # replace root from DeepDiff with more meaningful name.
+        if (
+            not key_parts
+        ):  # If key parts can't be find, keep original key so data is not lost.
+            key_parts = [
+                key.replace("root", "index_element")
+            ]  # replace root from DeepDiff with more meaningful name.
         partial_res = group_value(key_parts, value)
         dict_merger(result, partial_res)
     return result
@@ -79,9 +85,97 @@ def group_value(tree_list: List, value: Dict) -> Dict:
 def dict_merger(original_dict: Dict, dict_to_merge: Dict):
     """Function to merge a dictionary (dict_to_merge) recursively into the original_dict."""
     for key in dict_to_merge.keys():
-        if key in original_dict and isinstance(original_dict[key], dict) and isinstance(dict_to_merge[key], dict):
+        if (
+            key in original_dict
+            and isinstance(original_dict[key], dict)
+            and isinstance(dict_to_merge[key], dict)
+        ):
             dict_merger(original_dict[key], dict_to_merge[key])
         elif key in original_dict.keys():
-            original_dict[key + "_dup!"] = dict_to_merge[key]  # avoid overwriting existing keys.
+            original_dict[key + "_dup!"] = dict_to_merge[
+                key
+            ]  # avoid overwriting existing keys.
         else:
             original_dict[key] = dict_to_merge[key]
+
+
+def _parse_index_element_string(index_element_string):
+    """Build out dictionary from the index element string."""
+    result = {}
+    pattern = r"\[\'(.*?)\'\]"
+    match = re.findall(pattern, index_element_string)
+    if match:
+        for inner_key in match[1::]:
+            result[inner_key] = ""
+    return match, result
+
+
+def set_nested_value(data, keys, value):
+    """
+    Recursively sets a value in a nested dictionary, given a list of keys.
+
+    Args:
+        data (dict): The nested dictionary to modify.
+        keys (list): A list of keys to access the target value.
+        value: The value to set.
+
+    Returns:
+        None: The function modifies the dictionary in place.  Returns None.
+    """
+    if not keys:
+        return  # Should not happen, but good to have.
+    if len(keys) == 1:
+        data[keys[0]] = value
+    else:
+        if keys[0] not in data:
+            data[keys[0]] = {}  # Create the nested dictionary if it doesn't exist
+        set_nested_value(data[keys[0]], keys[1:], value)
+
+
+def parse_diff(jdiff_evaluate_response, actual, intended, match_config):
+    """Parse jdiff evaluate result into missing and extra dictionaries."""
+    extra = {}
+    missing = {}
+
+    def process_diff(_map, extra_map, missing_map, previous_key=None):
+        for key, value in _map.items():
+            if (
+                isinstance(value, dict)
+                and "new_value" in value
+                and "old_value" in value
+            ):
+                extra_map[key] = value["old_value"]
+                missing_map[key] = value["new_value"]
+            elif isinstance(value, str):
+                if "missing" in value:
+                    extra_map[key] = actual.get(match_config, {}).get(key)
+                if "new" in value:
+                    key_chain, _ = _parse_index_element_string(key)
+                    new_value = reduce(getitem, key_chain, intended)
+                    set_nested_value(missing_map, key_chain[1::], new_value)
+            elif isinstance(value, defaultdict):
+                if dict(value).get("new"):
+                    missing[previous_key][key] = dict(value).get("new", {})
+                if dict(value).get("missing"):
+                    extra_map[previous_key][key] = dict(value).get("missing", {})
+            elif isinstance(value, dict):
+                extra_map[key] = {}
+                missing_map[key] = {}
+                process_diff(value, extra_map[key], missing_map[key], previous_key=key)
+        return extra_map, missing_map
+
+    extras, missing = process_diff(jdiff_evaluate_response, extra, missing)
+    # Don't like this, but with less the performant way of doing it right now it works to clear out
+    # Any empty dicts that are left over from the diff.
+    # This is a bit of a hack, but it works for now.
+    final_extras = extras.copy()
+    final_missing = missing.copy()
+    for key, value in extras.items():
+        if isinstance(value, dict):
+            if not value:
+                del final_extras[key]
+    for key, value in missing.items():
+        if isinstance(value, dict):
+            if not value:
+                del final_missing[key]
+    return final_extras, final_missing

jdiff/utils/jmespath_parsers.py

@@ -5,6 +5,7 @@
 From one expression defined in jdiff, we will derive two expressions: one expression that traverse the json output and get the
 evaluated bit of it, the second will target the reference key relative to the value to evaluate. More on README.md
 """
+
 import re
 from typing import Mapping, List, Union
 

tasks.py

@@ -1,4 +1,5 @@
 """Tasks for use with Invoke."""
+
 import os
 import sys
 from invoke import task

tests/test_diff_generator.py

@@ -1,4 +1,5 @@
 """Diff generator tests."""
+
 import pytest
 from jdiff.evaluators import diff_generator
 from jdiff import extract_data_from_json