Merge pull request #9 from networktocode-llc/lvrfrc87

WIP: Working on #5 and #6

Author: lvrfrc87

.travis.yml

@@ -4,7 +4,7 @@ stages:
   - "lint"
   - "test"
 
-if: "type IN (pull_request)" # Add in "branch" as an option if desired for branch testing as well
+if: "type IN (pull_request)"  # Add in "branch" as an option if desired for branch testing as well
 language: "python"
 services:
   - "docker"
@@ -30,7 +30,7 @@ jobs:
         - "pip install invoke toml"
       script:
         - "invoke black"
-        - "invoke bandit" # Bandit fails to function on > Py3.8 https://github.com/PyCQA/bandit/issues/639
+        - "invoke bandit"  # Bandit fails to function on > Py3.8 https://github.com/PyCQA/bandit/issues/639
         # - "invoke pydocstyle"
         - "invoke flake8"
         - "invoke yamllint"

netcompare/check_type.py

@@ -16,14 +16,13 @@ def init(*args):
         check_type = args[0]
         if check_type == "exact_match":
             return ExactMatchType(*args)
-        elif check_type == "tolerance":
+        if check_type == "tolerance":
             return ToleranceType(*args)
-        else:
-            raise NotImplementedError
+        raise NotImplementedError
 
     @staticmethod
     def extract_value_from_json_path(
-        value: Mapping, path: Mapping, exclude: List = list()
+        value: Mapping, path: Mapping, exclude: List = None
     ) -> Union[Mapping, List, int, str, bool]:
         """Return the value contained into a Mapping for a defined path."""
         return extract_values_from_output(value, path, exclude)

netcompare/evaluator.py

@@ -11,24 +11,7 @@
 
 
 def diff_generator(pre_result: Mapping, post_result: Mapping) -> Dict:
-    """
-    Generates diff between pre and post data based on check definition.
-
-    Args:
-        pre_result: pre data result.
-        post_result: post data result.
-
-    Return:
-        output: diff between pre and post data.
-
-    Example:
-        >>> from evaluator import diff_generator
-        >>> pre_data = {"result": [{"bgp_neigh": "10.17.254.2","state": "Idle"}]}
-        >>> post_data = {"result": [{"bgp_neigh": "10.17.254.2","state": "Up"}]}
-        >>> check_definition = {"check_type": "exact_match", "value_path": "result[*].[state]", "reference_key_path": "result[*].bgp_neigh"}
-        >>> print(diff_generator(check_definition, post_data, check_definition))
-        {'10.17.254.2': {'state': {'new_value': 'Up', 'old_value': 'Idle'}}}
-    """
+    """Generates diff between pre and post data based on check definition."""
     diff_result = DeepDiff(pre_result, post_result)
 
     result = diff_result.get("values_changed", {})
@@ -45,38 +28,7 @@ def diff_generator(pre_result: Mapping, post_result: Mapping) -> Dict:
 
 
 def get_diff_iterables_items(diff_result: Mapping) -> Mapping:
-    """
-    Return a dict with new and missing values where the values are in a list.
-
-    Args:
-        diff_result: dict retruned from a DeepDiff compare.
-
-    Return:
-        defaultdict
-
-    Example:
-        >>> diff_result = {
-            "iterable_item_added": {
-                "root['Ethernet3'][1]": {
-                    "hostname": "ios-xrv-unittest",
-                    "port": "Gi0/0/0/0"
-                },
-                "root['Ethernet3'][2]": {
-                    "hostname": "ios-xrv-unittest",
-                    "port": "Gi0/0/0/1"
-                }
-            }
-        }
-        >>> get_diff_iterables_items(diff_result)
-        defaultdict(functools.partial(<class 'collections.defaultdict'>, <class 'list'>),
-            {"['Ethernet3']": defaultdict(list,
-                {'new': [
-                        {'hostname': 'ios-xrv-unittest', 'port': 'Gi0/0/0/0'},
-                        {'hostname': 'ios-xrv-unittest', 'port': 'Gi0/0/0/1'}
-                    ]}
-            )}
-        )
-    """
+    """Return a dict with new and missing values where the values are in a list."""
     # DeepDiff iterable_items are returned when the source data is a list
     # and provided in the format: "root['Ethernet3'][1]"
     # or more generically: root['KEY']['KEY']['KEY']...[numeric_index]
@@ -104,12 +56,10 @@ def get_diff_iterables_items(diff_result: Mapping) -> Mapping:
 
 def fix_deepdiff_key_names(obj: Mapping) -> Mapping:
     """Return a dict based on the provided dict object where the brackets and quotes are removed from the string."""
-    # sample keys:
-    # root[2]['10.64.207.255']['accepted_prefixes']
-    # root['Ethernet1'][0]['port']
     pattern = r"'([A-Za-z0-9_\./\\-]*)'"
 
-    result = dict()
+    result = {}
+    # root[2]['10.64.207.255']['accepted_prefixes']
     for key, value in obj.items():
         key_parts = re.findall(pattern, key)
         partial_res = group_value(key_parts, value)
@@ -118,39 +68,14 @@ def fix_deepdiff_key_names(obj: Mapping) -> Mapping:
 
 
 def group_value(tree_list: List, value: Mapping) -> Mapping:
-    """
-    Build dictionary based on value's key and reference key.
-
-    Args:
-        tree_list: list of value's key and reference key
-        value: value results
-
-    Return:
-        value: Mapping of the changed values associated to reference key
-
-    Example:
-        >>> tree_list = ['10.17.254.2', 'state']
-        >>> value = {'new_value': 'Up', 'old_value': 'Idle'}
-        {'10.17.254.2': {'state': {'new_value': 'Up', 'old_value': 'Idle'}}}
-    """
+    """Build dictionary based on value's key and reference key."""
     if tree_list:
         return {tree_list[0]: group_value(tree_list[1:], value)}
     return value
 
 
 def dict_merger(original_dict: Mapping, merged_dict: Mapping):
-    """
-    Merge dictionaries to build final result.
-
-    Args:
-        original_dict: empty dictionary to be merged
-        merged_dict: dictionary containing returned key/value and associated reference key
-
-    Example:
-        >>> original_dict = {}
-        >>> merged_dict = {'10.17.254.2': {'state': {'new_value': 'Up', 'old_value': 'Idle'}}}
-        {'10.17.254.2': {'state': {'new_value': 'Up', 'old_value': 'Idle'}}}
-    """
+    """Merge dictionaries to build final result."""
     for key in merged_dict.keys():
         if key in original_dict and isinstance(original_dict[key], dict) and isinstance(merged_dict[key], dict):
             dict_merger(original_dict[key], merged_dict[key])

netcompare/runner.py

@@ -1,32 +1,15 @@
-#!/ur/bin/env python3
+"""Library wrapper for output parsing."""
 import re
-import jmespath
 from typing import Mapping, List, Union
+import jmespath
 from .utils.jmspath_parsers import jmspath_value_parser, jmspath_refkey_parser
 from .utils.filter_parsers import exclude_filter
 from .utils.refkey import keys_cleaner, keys_values_zipper, associate_key_of_my_value
 from .utils.flatten import flatten_list
 
 
 def extract_values_from_output(value: Mapping, path: Mapping, exclude: List) -> Union[Mapping, List, int, str, bool]:
-    """Return data from output depending on the check path. See unit text for complete example.
-
-    Args:
-        path: "result[0].vrfs.default.peerList[*].[$peerAddress$,prefixesReceived]",
-        value: {
-            "jsonrpc": "2.0",
-            "id": "EapiExplorer-1",
-            "result": [
-                {
-                "vrfs": {
-                    "default": {
-                    "peerList": [
-                        { ...
-        exclude: ["interfaceStatistics", "interfaceCounters"]
-
-    Return:
-        [{'7.7.7.7': {'prefixesReceived': 101}}, {'10.1.0.0': {'prefixesReceived': 120}}, ...
-    """
+    """Return data from output depending on the check path. See unit text for complete example."""
     # Get the wanted values to be evaluated if jmspath expression is defined, otherwise
     # use the entire output if jmespath is not defined in check. This cover the "raw" diff type.
     if path and not exclude:
@@ -48,11 +31,9 @@ def extract_values_from_output(value: Mapping, path: Mapping, exclude: List) ->
             for item in element:
                 if isinstance(item, dict):
                     raise TypeError(
-                        'Must be list of lists i.e. [["Idle", 75759616], ["Idle", 75759620]]. You have {}\'.'.format(
-                            wanted_value
-                        )
+                        f'Must be list of lists i.e. [["Idle", 75759616], ["Idle", 75759620]]. You have {wanted_value}\'.'
                     )
-                elif isinstance(item, list):
+                if isinstance(item, list):
                     wanted_value = flatten_list(wanted_value)
                     break
 
@@ -64,5 +45,5 @@ def extract_values_from_output(value: Mapping, path: Mapping, exclude: List) ->
         wanted_reference_keys = jmespath.search(jmspath_refkey_parser(path), value)
         list_of_reference_keys = keys_cleaner(wanted_reference_keys)
         return keys_values_zipper(list_of_reference_keys, paired_key_value)
-    else:
-        return paired_key_value
+
+    return paired_key_value

netcompare/utils/filter_parsers.py

@@ -1,3 +1,4 @@
+"""Filtering parsing."""
 from typing import Mapping, List
 
 
@@ -36,5 +37,5 @@ def exclude_filter(data: Mapping, exclude: List):
 
     elif isinstance(data, list):
         for element in data:
-            if isinstance(element, dict) or isinstance(element, list):
+            if isinstance(element, (dict, list)):
                 exclude_filter(element, exclude)

netcompare/utils/flatten.py

@@ -1,3 +1,4 @@
+"""Flatten multi-nested list."""
 from typing import List, Generator
 
 

netcompare/utils/jmspath_parsers.py

@@ -1,3 +1,4 @@
+"""JMSPath expresion parsers."""
 import re
 
 
@@ -19,7 +20,8 @@ def jmspath_value_parser(path: str):
     if regex_normalized_value:
         normalized_value = regex_match_value.group().split("$")[1]
         return path.replace(regex_normalized_value.group(), normalized_value)
-    else: return path
+
+    return path
 
 
 def jmspath_refkey_parser(path: str):
@@ -33,13 +35,13 @@ def jmspath_refkey_parser(path: str):
     """
     splitted_jmspath = path.split(".")
 
-    for n, i in enumerate(splitted_jmspath):
-        regex_match_anchor = re.search(r"\$.*\$", i)
+    for number, element in enumerate(splitted_jmspath):
+        regex_match_anchor = re.search(r"\$.*\$", element)
 
         if regex_match_anchor:
-            splitted_jmspath[n] = regex_match_anchor.group().replace("$", "")
+            splitted_jmspath[number] = regex_match_anchor.group().replace("$", "")
 
-        if regex_match_anchor and not i.startswith("[") and not i.endswith("]"):
-            splitted_jmspath = splitted_jmspath[: n + 1]
+        if regex_match_anchor and not element.startswith("[") and not element.endswith("]"):
+            splitted_jmspath = splitted_jmspath[: number + 1]
 
     return ".".join(splitted_jmspath)

netcompare/utils/refkey.py

@@ -1,56 +1,31 @@
+"""Reference key utilities."""
 from typing import Mapping, List
 
 
 def keys_cleaner(wanted_reference_keys: Mapping) -> List[Mapping]:
-    """
-    Get every required reference key from output.
-
-    Args:
-        wanted_reference_keys: {'10.1.0.0': {'address_family': {'ipv4': ...
-
-    Return:
-        ['10.1.0.0', '10.2.0.0', '10.64.207.255', '7.7.7.7']
-
-    Example:
-        >>> from runner import keys_cleaner
-        >>> wanted_reference_keys = {'10.1.0.0': {'address_family': 'ipv4'}}
-        >>> keys_cleaner(wanted_reference_keys)
-        ['10.1.0.0', '10.2.0.0', '10.64.207.255', '7.7.7.7']
-    """
+    """Get every required reference key from output."""
     if isinstance(wanted_reference_keys, list):
         return wanted_reference_keys
 
-    elif isinstance(wanted_reference_keys, dict):
-        my_keys_list = list()
+    if isinstance(wanted_reference_keys, dict):
+        my_keys_list = []
 
-        if isinstance(wanted_reference_keys, dict): 
+        if isinstance(wanted_reference_keys, dict):
             for key in wanted_reference_keys.keys():
                 my_keys_list.append(key)
         else:
-            raise TypeError(f'Must be a dictionary. You have type:{type(wanted_reference_keys)} output:{wanted_reference_keys}\'.')
-                    
-        return my_keys_list
-
+            raise TypeError(
+                f"Must be a dictionary. You have type:{type(wanted_reference_keys)} output:{wanted_reference_keys}'."
+            )
 
-def keys_values_zipper(list_of_reference_keys: List, wanted_value_with_key: List) -> List:
-    """
-    Build dictionary zipping keys with relative values.
+        return my_keys_list
 
-    Args:
-        list_of_reference_keys: ['10.1.0.0', '10.2.0.0', '10.64.207.255', '7.7.7.7']
-        wanted_value_with_key: [{'is_enabled': True, 'is_up': False}, ...
+    return None
 
-    Return:
-        [{'10.1.0.0': {'is_enabled': True, 'is_up': False}}, , ...
 
-    Example:
-        >>> from runner import keys_values_zipper
-        >>> list_of_reference_keys = ['10.1.0.0']
-        >>> wanted_value_with_key = [{'is_enabled': True, 'is_up': False}]
-        >>> keys_values_zipper(list_of_reference_keys, wanted_value_with_key)
-        [{'10.1.0.0': {'is_enabled': True, 'is_up': False}}]
-    """
-    final_result = list()
+def keys_values_zipper(list_of_reference_keys: List, wanted_value_with_key: List) -> List:
+    """Build dictionary zipping keys with relative values."""
+    final_result = []
 
     if len(list_of_reference_keys) != len(wanted_value_with_key):
         raise ValueError("Keys len != from Values len")
@@ -62,23 +37,7 @@ def keys_values_zipper(list_of_reference_keys: List, wanted_value_with_key: List
 
 
 def associate_key_of_my_value(paths: str, wanted_value: List) -> List:
-    """
-    Associate each key defined in path to every value found in output.
-
-    Args:
-        paths: {"path": "global.peers.*.[is_enabled,is_up]"}
-        wanted_value: [[True, False], [True, False], [True, False], [True, False]]
-
-    Return:
-        [{'is_enabled': True, 'is_up': False}, ...
-
-    Example:
-        >>> from runner import associate_key_of_my_value
-        >>> path = {"path": "global.peers.*.[is_enabled,is_up]"}
-        >>> wanted_value = [[True, False], [True, False], [True, False], [True, False]]
-        {'is_enabled': True, 'is_up': False}, {'is_enabled': True, 'is_up': False}, ...
-    """
-
+    """Associate each key defined in path to every value found in output."""
     # global.peers.*.[is_enabled,is_up] / result.[*].state
     find_the_key_of_my_values = paths.split(".")[-1]
 
@@ -90,10 +49,10 @@ def associate_key_of_my_value(paths: str, wanted_value: List) -> List:
     else:
         my_key_value_list = [find_the_key_of_my_values]
 
-    final_list = list()
+    final_list = []
 
     for items in wanted_value:
-        temp_dict = dict()
+        temp_dict = {}
 
         if len(items) != len(my_key_value_list):
             raise ValueError("Key's value len != from value len")