prep new jdiff compliance testing

Author: jeffkala

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

@@ -16,9 +16,7 @@
 )
 
 
-def extract_data_from_json(
-    data: Union[Mapping, List], path: str = "*", exclude: Optional[List] = None
-) -> Any:
+def extract_data_from_json(data: Union[Mapping, List], path: str = "*", exclude: Optional[List] = None) -> Any:
     """Return wanted data from outpdevice data based on the check path. See unit test for complete example.
 
     Get the wanted values to be evaluated if JMESPath expression is defined,
@@ -37,16 +35,12 @@ def extract_data_from_json(
     """
     if exclude and isinstance(data, Dict):
         if not isinstance(exclude, list):
-            raise ValueError(
-                f"Exclude list must be defined as a list. You have {type(exclude)}"
-            )
+            raise ValueError(f"Exclude list must be defined as a list. You have {type(exclude)}")
         # exclude unwanted elements
         exclude_filter(data, exclude)
 
     if not path:
-        warnings.warn(
-            "JMSPath cannot be empty string or type 'None'. Path argument reverted to default value '*'"
-        )
+        warnings.warn("JMSPath cannot be empty string or type 'None'. Path argument reverted to default value '*'")
         path = "*"
 
     if path == "*":
@@ -85,27 +79,19 @@ def extract_data_from_json(
     # Based on the expression or data we might have different data types
     # therefore we need to normalize.
     if re.search(r"\$.*\$", path):
-        paired_key_value = associate_key_of_my_value(
-            jmespath_value_parser(path), values
-        )
+        paired_key_value = associate_key_of_my_value(jmespath_value_parser(path), values)
         wanted_reference_keys = jmespath.search(jmespath_refkey_parser(path), data)
 
-        if isinstance(
-            wanted_reference_keys, dict
-        ):  # when wanted_reference_keys is dict() type
+        if isinstance(wanted_reference_keys, dict):  # when wanted_reference_keys is dict() type
             list_of_reference_keys = list(wanted_reference_keys.keys())
         elif any(
             isinstance(element, list) for element in wanted_reference_keys
         ):  # when wanted_reference_keys is a nested list
             list_of_reference_keys = flatten_list(wanted_reference_keys)[0]
-        elif isinstance(
-            wanted_reference_keys, list
-        ):  # when wanted_reference_keys is a list
+        elif isinstance(wanted_reference_keys, list):  # when wanted_reference_keys is a list
             list_of_reference_keys = wanted_reference_keys
         else:
-            raise ValueError(
-                "Reference Key normalization failure. Please verify data type returned."
-            )
+            raise ValueError("Reference Key normalization failure. Please verify data type returned.")
 
         normalized = keys_values_zipper(list_of_reference_keys, paired_key_value)
         # Data between pre and post may come in different order, so it needs to be sorted.

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

@@ -2,7 +2,8 @@
 
 import re
 from collections import defaultdict
-from functools import partial
+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_\./\\-]*)'"
@@ -63,12 +64,8 @@ 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
@@ -84,16 +81,10 @@ 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]
 
@@ -106,7 +97,45 @@ def _parse_index_element_string(index_element_string):
     if match:
         for inner_key in match[1::]:
             result[inner_key] = ""
-    return result
+    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 all_values_empty(input_dict):
+    """
+    Checks if all values in a dictionary are empty objects (empty string, list, or dictionary).
+
+    Args:
+        input_dict: The dictionary to check.
+
+    Returns:
+        True if all values are empty, False otherwise.
+    """
+    for value in input_dict.values():
+        if value:  # Empty objects evaluate to False in a boolean context
+            return False
+    return True
 
 
 def parse_diff(jdiff_evaluate_response, actual, intended, match_config):
@@ -116,67 +145,39 @@ def parse_diff(jdiff_evaluate_response, actual, intended, match_config):
 
     def process_diff(_map, extra_map, missing_map):
         for key, value in _map.items():
-            if (
-                isinstance(value, dict)
-                and "new_value" in value
-                and "old_value" in value
-            ):
+            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:
-                    new_key = _parse_index_element_string(key)
-                    new_key[key] = intended.get(match_config, {}).get(key)
-                    missing_map.update(new_key)
+                    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[key] = dict(value).get("new", {})
+                if dict(value).get("missing"):
+                    extra_map[key] = dict(value).get("missing", {})
             elif isinstance(value, dict):
                 extra_map[key] = {}
                 missing_map[key] = {}
                 process_diff(value, extra_map[key], missing_map[key])
         return extra_map, missing_map
 
-    extras, missings = process_diff(jdiff_evaluate_response, extra, missing)
-    return extras, missings
-
-
-# result = {'hostname': {'new_value': 'veos', 'old_value': 'veos-0'}, 'domain-name': 'missing'}
-# result = {'domain-name': 'missing'}
-# result = {'hostname': {'new_value': 'veos', 'old_value': 'veos-0'}, 'domain-name': 'missing', "index_element['openconfig-system:config']['ip name']": 'new'}
-# result = {'domain-name': 'missing','hostname': 'missing', "index_element['openconfig-system:config']['ip name']": 'new'}
-# result = {'servers': {'server': defaultdict(<class 'list'>, {'missing': [{'address': '1.us.pool.ntp.org', 'config': {'address': '1.us.pool.ntp.org'}, 'state': {'address': '1.us.pool.ntp.org'}}]})}}
-
-# '''
-# ```
-# from jdiff import CheckType
-
-# a = {
-#     "openconfig-system:ntp": {
-#         "servers": {
-#             "server": [
-#                 {
-#                     "address": "1.us.pool.ntp.org",
-#                     "config": {
-#                         "address": "1.us.pool.ntp.org"
-#                     },
-#                     "state": {
-#                         "address": "1.us.pool.ntp.org"
-#                     }
-#                 }
-#             ]
-#         }
-#     }
-# }
-
-# i = {
-#     "openconfig-system:ntp": {
-#         "servers": {
-#             "server": []
-#         }
-#     }
-# }
-
-# jdiff_param_match = CheckType.create("exact_match")
-# result, compliant = jdiff_param_match.evaluate(a, i)
-# ```
-# '''
+    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 distutils.util import strtobool

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