-
Notifications
You must be signed in to change notification settings - Fork 7
Expand file tree
/
Copy pathevaluators.py
More file actions
117 lines (88 loc) · 4.4 KB
/
Copy pathevaluators.py
File metadata and controls
117 lines (88 loc) · 4.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
"""Evaluators."""
import re
from typing import Any, Dict, List, Mapping, Tuple
from deepdiff import DeepDiff
from .operator import Operator
from .utils.diff_helpers import fix_deepdiff_key_names, get_diff_iterables_items
def diff_generator(pre_result: Any, post_result: Any) -> Dict:
"""Generates diff between pre and post data based on check definition.
Args:
pre_result: dataset to compare
post_result: dataset to compare
Returns:
dict: differences between two datasets with the following keys:
- "values_changed": Item values that have changed
- "missing": Item keys that have been removed
- "new": Item keys that have been added
"""
diff_result = DeepDiff(pre_result, post_result)
result = diff_result.get("values_changed", {})
if diff_result.get("dictionary_item_removed"):
result.update({k: "missing" for k in diff_result["dictionary_item_removed"]})
if diff_result.get("dictionary_item_added"):
result.update({k: "new" for k in diff_result["dictionary_item_added"]})
iterables_items = get_diff_iterables_items(diff_result)
if iterables_items:
result.update(iterables_items)
return fix_deepdiff_key_names(result)
def parameter_evaluator(values: List[Dict], parameters: Mapping, mode: str) -> Dict:
"""Parameter Match evaluator engine.
Args:
values: List of items what we will check the parameters against
parameters: Dict with the keys and reference values to check
mode: "match" or "no-match" to define the evaluation mode
Example:
values: [{'7.7.7.7': {'peerAddress': '7.7.7.7', 'localAsn': '65130.1100', 'linkType': 'external'}}]
parameters: {'localAsn': '65130.1100', 'linkType': 'external'}
Returns:
Dictionary with all the items that have some value not matching the expectations from parameters
"""
if not isinstance(values, list):
raise TypeError("'values' must be of type List.")
result = {}
for index, value in enumerate(values):
# value: {'7.7.7.7': {'peerAddress': '7.7.7.7', 'localAsn': '65130.1101', 'linkType': 'externals
if not isinstance(value, dict):
raise TypeError(f"'value' ({value}) must be of type Dict, and it's {type(value)}")
result_item = {}
# When data has been normalized with $key$, get inner key and value
if len(value) == 1:
# inner_key: '7.7.7.7'
inner_key = list(value.keys())[0]
# inner_value: [{'peerAddress': '7.7.7.7', 'localAsn': '65130.1101', 'linkType': 'externals'}]
value = list(value.values())[0]
else:
inner_key = index
for parameter_key, parameter_value in parameters.items():
if mode == "match" and value[parameter_key] != parameter_value:
result_item[parameter_key] = value[parameter_key]
elif mode == "no-match" and value[parameter_key] == parameter_value:
result_item[parameter_key] = value[parameter_key]
if result_item:
result[inner_key] = result_item
return result
def regex_evaluator(values: List[Dict[Any, Dict]], regex_expression: str, mode: str) -> Dict:
"""Regex Match evaluator engine."""
# values: [{'7.7.7.7': {'peerGroup': 'EVPN-OVERLAY-SPINE'}}]
# parameter: {'regex': '.*UNDERLAY.*', 'mode': 'match'}
result = {}
if not isinstance(values, list):
raise TypeError("Something went wrong during JMSPath parsing. 'values' must be of type List.")
for item in values:
for founded_value in item.values():
for value in founded_value.values():
match_result = re.search(regex_expression, value)
# Fail if there is no regex match for "match" mode
if mode == "match" and not match_result:
result.update(item)
# Fail if there is regex match for "no-match" mode.
elif mode == "no-match" and match_result:
result.update(item)
return result
def operator_evaluator(reference_data: Mapping, value_to_compare: Mapping) -> Tuple[Dict, bool]:
"""Operator evaluator call."""
# reference_data
# {'mode': 'all-same', 'operator_data': True}
operator_mode = reference_data["mode"].replace("-", "_")
operator = Operator(reference_data["operator_data"], value_to_compare)
return getattr(operator, operator_mode)()