operator check implementation (#39)

lvrfrc87 · pszulczewski · web-flow · commit ce2ecbaa87b6 · 2022-03-07T09:27:40.000+01:00
* draft regex base function

* fix file import

* draft regex implementation

* fix logic into regex result

* remove assertion

* draft test for range check type

* add match/no-match logic

* commit save

* add assertion for check option

* add assertions for range type

* add assertion for strings

* commit save

* draft operator check type

* complete validate method for check_type operator

* work on oeprator_evaluator method

* drat operator all-same

* add all-same operator

* add contains operator

* add not-contains. Update contains logic

* add is-gt and is-lt operators

* add range operators

* add operator lib for DRY code

* fix some lint errors

* Update netcompare/operator.py

Not sure about this...if you look at the if statements, your suggestion would brake the logic...or am I missing something?

Co-authored-by: Patryk Szulczewski &lt;patryk@szulczewski.eu&gt;

* work on comments

* working on operator args

* fix tests

* improve validator switch logic

* fix pytests

Co-authored-by: Patryk Szulczewski &lt;patryk@szulczewski.eu&gt;
diff --git a/netcompare/check_types.py b/netcompare/check_types.py
@@ -3,8 +3,6 @@
 from typing import Mapping, Tuple, List, Dict, Any, Union
 from abc import ABC, abstractmethod
 import jmespath
-
-
 from .utils.jmespath_parsers import (
     jmespath_value_parser,
     jmespath_refkey_parser,
@@ -13,11 +11,10 @@
     keys_values_zipper,
 )
 from .utils.data_normalization import exclude_filter, flatten_list
-from .evaluators import diff_generator, parameter_evaluator, regex_evaluator
-
-# pylint: disable=arguments-differ
+from .evaluators import diff_generator, parameter_evaluator, regex_evaluator, operator_evaluator
 
 
+# pylint: disable=arguments-differ
 class CheckType(ABC):
     """Check Type Base Abstract Class."""
 
@@ -36,6 +33,8 @@ def init(check_type: str):
             return ParameterMatchType()
         if check_type == "regex":
             return RegexType()
+        if check_type == "operator":
+            return OperatorType()
 
         raise NotImplementedError
 
@@ -115,7 +114,7 @@ class ExactMatchType(CheckType):
     """Exact Match class docstring."""
 
     @staticmethod
-    def validate(**kwargs):
+    def validate(**kwargs) -> None:
         """Method to validate arguments."""
         # reference_data = getattr(kwargs, "reference_data")
 
@@ -130,7 +129,7 @@ class ToleranceType(CheckType):
     """Tolerance class docstring."""
 
     @staticmethod
-    def validate(**kwargs):
+    def validate(**kwargs) -> None:
         """Method to validate arguments."""
         # reference_data = getattr(kwargs, "reference_data")
         tolerance = kwargs.get("tolerance")
@@ -177,7 +176,7 @@ class ParameterMatchType(CheckType):
     """Parameter Match class implementation."""
 
     @staticmethod
-    def validate(**kwargs):
+    def validate(**kwargs) -> None:
         """Method to validate arguments."""
         mode_options = ["match", "no-match"]
         params = kwargs.get("params")
@@ -206,7 +205,7 @@ class RegexType(CheckType):
     """Regex Match class implementation."""
 
     @staticmethod
-    def validate(**kwargs):
+    def validate(**kwargs) -> None:
         """Method to validate arguments."""
         mode_options = ["match", "no-match"]
         regex = kwargs.get("regex")
@@ -228,3 +227,80 @@ def evaluate(self, value_to_compare: Mapping, regex: str, mode: str) -> Tuple[Ma
         self.validate(regex=regex, mode=mode)
         diff = regex_evaluator(value_to_compare, regex, mode)
         return diff, not diff
+
+
+class OperatorType(CheckType):
+    """Operator class implementation."""
+
+    @staticmethod
+    def validate(**kwargs) -> None:
+        """Validate operator parameters."""
+        in_operators = ("is-in", "not-in", "in-range", "not-range")
+        bool_operators = ("all-same",)
+        number_operators = ("is-gt", "is-lt")
+        # "equals" is redundant with check type "exact_match" an "parameter_match"
+        # equal_operators = ("is-equal", "not-equal")
+        string_operators = ("contains", "not-contains")
+        valid_options = (
+            in_operators,
+            bool_operators,
+            number_operators,
+            string_operators,
+            # equal_operators,
+        )
+
+        # Validate "params" argument is not None.
+        if not kwargs:
+            raise KeyError(f"'params' argument must be provided. You have {kwargs}. Read the docs for more info.")
+
+        params_key = kwargs["mode"]
+        params_value = kwargs["operator_data"]
+        # Validate "params" value is legal.
+        if all(params_key in operator for operator in valid_options):
+            raise ValueError(
+                f"'params' value must be one of the following: {[sub_element for element in valid_options for sub_element in element]}. You have: {params_key}"
+            )
+
+        if params_key in in_operators:
+            # "is-in", "not-in", "in-range", "not-range" requires an iterable
+            if not isinstance(params_value, (list, tuple)):
+                raise ValueError(
+                    f"Range check-option {in_operators} must have value of type list or tuple. i.e: dict(not-in=('Idle', 'Down'). You have: {params_value} of type {type(params_value)}You have: {params_value} of type {type(params_value)}"
+                )
+
+            # "in-range", "not-range" requires int or float where value at index 0 is lower than value at index 1
+            if params_key in ("in-range", "not-range"):
+                if not isinstance(params_value[0], (int, float)) and not isinstance(params_value[1], float, int):
+                    raise ValueError(
+                        f"Range check-option {params_key} must have value of type list or tuple with items of type float or int. i.e: dict(not-range=(70000000, 80000000). You have: {params_value} of type {type(params_value)}"
+                    )
+                if not params_value[0] < params_value[1]:
+                    raise ValueError(
+                        f"'range' and 'not-range' must have value at index 0 lower than value at index 1. i.e: dict(not-range=(70000000, 80000000). You have: {params_value} of type {type(params_value)}"
+                    )
+
+        # "is-gt","is-lt"  require either int() or float()
+        elif params_key in number_operators and not isinstance(params_value, (float, int)):
+            raise ValueError(
+                f"Check-option {number_operators} must have value of type float or int. i.e: dict(is-lt=50). You have: {params_value} of type {type(params_value)}"
+            )
+
+        # "contains", "not-contains" require string.
+        elif params_key in string_operators and not isinstance(params_value, str):
+            raise ValueError(
+                f"Range check-option {string_operators} must have value of type string. i.e: dict(contains='EVPN'). You have: {params_value} of type {type(params_value)}"
+            )
+
+        # "all-same" requires boolean True or False
+        elif params_key in bool_operators and not isinstance(params_value, bool):
+            raise ValueError(
+                f"Range check-option {bool_operators} must have value of type bool. i.e: dict(all-same=True). You have: {params_value} of type {type(params_value)}"
+            )
+
+    def evaluate(self, value_to_compare: Any, params: Any) -> Tuple[Mapping, bool]:
+        """Operator evaluator implementation."""
+        self.validate(**params)
+        # For naming consistency
+        reference_data = params
+        evaluation_result = operator_evaluator(reference_data, value_to_compare)
+        return evaluation_result, not evaluation_result
diff --git a/netcompare/evaluators.py b/netcompare/evaluators.py
@@ -3,6 +3,7 @@
 from typing import Any, Mapping, Dict
 from deepdiff import DeepDiff
 from .utils.diff_helpers import get_diff_iterables_items, fix_deepdiff_key_names
+from .operator import Operator
 
 
 def diff_generator(pre_result: Any, post_result: Any) -> Dict:
@@ -99,3 +100,14 @@ def regex_evaluator(values: Mapping, regex_expression: str, mode: str) -> Dict:
                     result.update(item)
 
     return result
+
+
+def operator_evaluator(referance_data: Mapping, value_to_compare: Mapping) -> Dict:
+    """Operator evaluator call."""
+    # referance_data
+    # {'mode': 'all-same', 'operator_data': True}
+    operator_mode = referance_data["mode"].replace("-", "_")
+    operator = Operator(referance_data["operator_data"], value_to_compare)
+
+    result = getattr(operator, operator_mode)()
+    return result
diff --git a/netcompare/operator.py b/netcompare/operator.py
@@ -0,0 +1,109 @@
+"""Operator diff."""
+import operator
+
+
+class Operator:
+    """Operator class implementation."""
+
+    def __init__(self, referance_data, value_to_compare) -> None:
+        """__init__ method."""
+        # [{'7.7.7.7': {'peerGroup': 'EVPN-OVERLAY-SPINE', 'vrf': 'default', 'state': 'Idle'}},
+        # {'10.1.0.0': {'peerGroup': 'IPv4-UNDERLAY-SPINE', 'vrf': 'default', 'state': 'Idle'}},
+        # {'10.2.0.0': {'peerGroup': 'IPv4-UNDERLAY-SPINE', 'vrf': 'default', 'state': 'Idle'}},
+        # {'10.64.207.255': {'peerGroup': 'IPv4-UNDERLAY-MLAG-PEER', 'vrf': 'default', 'state': 'Idle'}}]
+        self.referance_data = referance_data
+        self.value_to_compare = value_to_compare
+
+    def _loop_through_wrapper(self, call_ops):
+        """Wrappoer method for operator evaluation."""
+        ops = {
+            ">": operator.gt,
+            "<": operator.lt,
+            "is_in": operator.contains,
+            "not_in": operator.contains,
+            "contains": operator.contains,
+            "not_contains": operator.contains,
+        }
+
+        result = []
+        for item in self.value_to_compare:
+            for value in item.values():
+                for evaluated_value in value.values():
+                    # reverse operands (??? WHY ???) https://docs.python.org/3.8/library/operator.html#operator.contains
+                    if call_ops == "is_in":
+                        if ops[call_ops](self.referance_data, evaluated_value):
+                            result.append(item)
+                    elif call_ops == "not_contains":
+                        if not ops[call_ops](evaluated_value, self.referance_data):
+                            result.append(item)
+                    elif call_ops == "not_in":
+                        if not ops[call_ops](self.referance_data, evaluated_value):
+                            result.append(item)
+                    elif call_ops == "in_range":
+                        if self.referance_data[0] < evaluated_value < self.referance_data[1]:
+                            result.append(item)
+                    elif call_ops == "not_range":
+                        if not self.referance_data[0] < evaluated_value < self.referance_data[1]:
+                            result.append(item)
+                    # "<", ">", "contains"
+                    elif ops[call_ops](evaluated_value, self.referance_data):
+                        result.append(item)
+        if result:
+            return (True, result)
+        return (False, result)
+
+    def all_same(self):
+        """All same operator implementation."""
+        list_of_values = []
+        result = []
+
+        for item in self.value_to_compare:
+            for value in item.values():
+                # Create a list for compare valiues.
+                list_of_values.append(value)
+
+        for element in list_of_values:
+            if element != list_of_values[0]:
+                result.append(False)
+            else:
+                result.append(True)
+
+        if self.referance_data and not all(result):
+            return (False, self.value_to_compare)
+        if self.referance_data:
+            return (True, self.value_to_compare)
+        if not all(result):
+            return (True, self.value_to_compare)
+        return (False, self.value_to_compare)
+
+    def contains(self):
+        """Contains operator implementation."""
+        return self._loop_through_wrapper("contains")
+
+    def not_contains(self):
+        """Not contains operator implementation."""
+        return self._loop_through_wrapper("not_contains")
+
+    def is_gt(self):
+        """Is greather than operator implementation."""
+        return self._loop_through_wrapper(">")
+
+    def is_lt(self):
+        """Is lower than operator implementation."""
+        return self._loop_through_wrapper("<")
+
+    def is_in(self):
+        """Is in operator implementation."""
+        return self._loop_through_wrapper("is_in")
+
+    def not_in(self):
+        """Is not in operator implementation."""
+        return self._loop_through_wrapper("not_in")
+
+    def in_range(self):
+        """Is in range operator implementation."""
+        return self._loop_through_wrapper("in_range")
+
+    def not_range(self):
+        """Is not in range operator implementation."""
+        return self._loop_through_wrapper("not_range")
diff --git a/pyproject.toml b/pyproject.toml
@@ -57,7 +57,9 @@ no-docstring-rgx="^(_|test_|Meta$)"
 disable = """,
     line-too-long,
     bad-continuation,
-    E5110
+    E5110,
+    R0912,
+    R0801
     """
 
 [tool.pylint.miscellaneous]
diff --git a/tests/test_operators.py b/tests/test_operators.py
diff --git a/tests/test_type_checks.py b/tests/test_type_checks.py
