move flatten funcs under utils

Author: lvrfrc87

netcompare/runner.py

@@ -2,7 +2,8 @@
 import re
 import jmespath
 from typing import Mapping, List, Generator, Union
-from .utils.jmspath.parsers import jmspath_value_parser, jmspath_refkey_parser, exclude_filter
+from .utils.jmspath.parsers import jmspath_value_parser, jmspath_refkey_parser
+from .utils.data.parsers import  exclude_filter, get_values
 
 
 def extract_values_from_output(value: Mapping, path: Mapping, exclude: List) -> Union[Mapping, List, int, str, bool]:
@@ -37,8 +38,7 @@ def extract_values_from_output(value: Mapping, path: Mapping, exclude: List) ->
         exclude_filter(wanted_value, exclude)
         filtered_value = wanted_value
 
-    pdb.set_trace()
-    filtered_value = get_meaningful_values(path, wanted_value)
+    filtered_value = get_values(path, wanted_value)
 
     if path and re.search(r"\$.*\$", path):
         wanted_reference_keys = jmespath.search(jmspath_refkey_parser(path), value)
@@ -48,67 +48,6 @@ def extract_values_from_output(value: Mapping, path: Mapping, exclude: List) ->
         return filtered_value
 
 
-def get_meaningful_values(path: Mapping, wanted_value):
-    if path:
-        # check if list of lists
-        if not any(isinstance(i, list) for i in wanted_value):
-            raise TypeError(
-                "Catching value must be defined as list in jmespath expression i.e. result[*].state -> result[*].[state]. You have {}'.".format(
-                    path
-                )
-            )
-        for element in wanted_value:
-            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
-                        )
-                    )
-                elif isinstance(item, list):
-                    wanted_value = flatten_list(wanted_value)
-                    break
-
-        filtered_value = associate_key_of_my_value(jmspath_value_parser(path), wanted_value)
-    else:
-        filtered_value = wanted_value
-    return filtered_value
-
-
-def flatten_list(my_list: List) -> List:
-    """
-    Flatten a multi level nested list and returns a list of lists.
-
-    Args:
-        my_list: nested list to be flattened.
-
-    Return:
-        [[-1, 0], [-1, 0], [-1, 0], ...]
-
-    Example:
-        >>> my_list = [[[[-1, 0], [-1, 0]]]]
-        >>> flatten_list(my_list)
-        [[-1, 0], [-1, 0]]
-    """
-    if not isinstance(my_list, list):
-        raise ValueError(f"Argument provided must be a list. You passed a {type(my_list)}")
-    if is_flat_list(my_list):
-        return my_list
-    return list(iter_flatten_list(my_list))
-
-
-def iter_flatten_list(my_list: List) -> Generator[List, None, None]:
-    """Recursively yield all flat lists within a given list."""
-    if is_flat_list(my_list):
-        yield my_list
-    else:
-        for item in my_list:
-            yield from iter_flatten_list(item)
-
-
-def is_flat_list(obj: List) -> bool:
-    """Return True is obj is a list that does not contain any lists as its first order elements."""
-    return isinstance(obj, list) and not any(isinstance(i, list) for i in obj)
 
 
 def associate_key_of_my_value(paths: Mapping, wanted_value: List) -> List:

netcompare/utils/data/__init__.py

@@ -0,0 +1,69 @@
+from typing import Mapping, List
+from ..jmspath.parsers import jmspath_value_parser
+from ..list.flatten import flatten_list
+from ...runner import associate_key_of_my_value
+
+def exclude_filter(data: Mapping, exclude: List):
+    """
+    Recusively look through all dict keys and pop out the one defined in "exclude".
+
+    Update in place existing dictionary. Look into unit test for example.
+
+    Args:
+        data: {
+                "interfaces": {
+                "Management1": {
+                    "name": "Management1",
+                    "interfaceStatus": "connected",
+                    "autoNegotiate": "success",
+                    "interfaceStatistics": {
+                        "inBitsRate": 3403.4362520883615,
+                        "inPktsRate": 3.7424095978179257,
+                        "outBitsRate": 16249.69114419833,
+                        "updateInterval": 300,
+                        "outPktsRate": 2.1111866059750692
+                    },...
+        exclude: ["interfaceStatistics", "interfaceCounters"]
+    """
+    if isinstance(data, dict):
+        for exclude_element in exclude:
+            try:
+                data.pop(exclude_element)
+            except KeyError:
+                pass
+
+        for key in data:
+            if isinstance(data[key], dict) or isinstance(data[key], list):
+                exclude_filter(data[key], exclude)
+
+    elif isinstance(data, list):
+        for element in data:
+            if isinstance(element, dict) or isinstance(element, list):
+                exclude_filter(element, exclude)
+
+
+def get_values(path: Mapping, wanted_value):
+    if path:
+        # check if list of lists
+        if not any(isinstance(i, list) for i in wanted_value):
+            raise TypeError(
+                "Catching value must be defined as list in jmespath expression i.e. result[*].state -> result[*].[state]. You have {}'.".format(
+                    path
+                )
+            )
+        for element in wanted_value:
+            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
+                        )
+                    )
+                elif isinstance(item, list):
+                    wanted_value = flatten_list(wanted_value)
+                    break
+
+        filtered_value = associate_key_of_my_value(jmspath_value_parser(path), wanted_value)
+    else:
+        filtered_value = wanted_value
+    return filtered_value

netcompare/utils/data/parsers.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+
+import pytest
+import sys
+from ..parsers import exclude_filter
+sys.path.append("..")
+
+
+assertion_failed_message = """Test output is different from expected output.
+output: {output}
+expected output: {expected_output}
+"""
+
+exclude_filter_case_1 = (
+    ["interfaceStatistics"],
+    {
+        "interfaces": {
+        "Management1": {
+            "name": "Management1",
+            "interfaceStatus": "connected",
+            "autoNegotiate": "success",
+            "interfaceStatistics": {
+                "inBitsRate": 3403.4362520883615,
+                "inPktsRate": 3.7424095978179257,
+                "outBitsRate": 16249.69114419833,
+                "updateInterval": 300,
+                "outPktsRate": 2.1111866059750692
+                }
+            }
+        }
+    },
+    {
+        "interfaces": {
+        "Management1": {
+            "name": "Management1",
+            "interfaceStatus": "connected",
+            "autoNegotiate": "success"
+            }
+        }
+    }
+)
+
+exclude_filter_tests = [
+    exclude_filter_case_1,
+]
+
+
+@pytest.mark.parametrize("exclude, data, expected_output", exclude_filter_tests)
+def test_exclude_filter(exclude, data, expected_output):
+    exclude_filter(data, exclude)
+    assert expected_output == data, assertion_failed_message.format(output=data, expected_output=expected_output)

netcompare/utils/data/tests/__init__.py

@@ -46,41 +46,3 @@ def jmspath_refkey_parser(path: Mapping):
 
     return ".".join(splitted_jmspath)
 
-
-def exclude_filter(data: Mapping, exclude: List):
-    """
-    Recusively look through all dict keys and pop out the one defined in "exclude".
-
-    Update in place existing dictionary. Look into unit test for example.
-
-    Args:
-        data: {
-                "interfaces": {
-                "Management1": {
-                    "name": "Management1",
-                    "interfaceStatus": "connected",
-                    "autoNegotiate": "success",
-                    "interfaceStatistics": {
-                        "inBitsRate": 3403.4362520883615,
-                        "inPktsRate": 3.7424095978179257,
-                        "outBitsRate": 16249.69114419833,
-                        "updateInterval": 300,
-                        "outPktsRate": 2.1111866059750692
-                    },...
-        exclude: ["interfaceStatistics", "interfaceCounters"]
-    """
-    if isinstance(data, dict):
-        for exclude_element in exclude:
-            try:
-                data.pop(exclude_element)
-            except KeyError:
-                pass
-
-        for key in data:
-            if isinstance(data[key], dict) or isinstance(data[key], list):
-                exclude_filter(data[key], exclude)
-
-    elif isinstance(data, list):
-        for element in data:
-            if isinstance(element, dict) or isinstance(element, list):
-                exclude_filter(element, exclude)

netcompare/utils/data/tests/test_parsers.py

@@ -2,7 +2,7 @@
 
 import pytest
 import sys
-from ..parsers import jmspath_value_parser, jmspath_refkey_parser, exclude_filter
+from ..parsers import jmspath_value_parser, jmspath_refkey_parser
 sys.path.append("..")
 
 
@@ -45,34 +45,6 @@
     "result[0].vrfs",
 )
 
-exclude_filter_case_1 = (
-    ["interfaceStatistics"],
-    {
-        "interfaces": {
-        "Management1": {
-            "name": "Management1",
-            "interfaceStatus": "connected",
-            "autoNegotiate": "success",
-            "interfaceStatistics": {
-                "inBitsRate": 3403.4362520883615,
-                "inPktsRate": 3.7424095978179257,
-                "outBitsRate": 16249.69114419833,
-                "updateInterval": 300,
-                "outPktsRate": 2.1111866059750692
-                }
-            }
-        }
-    },
-    {
-        "interfaces": {
-        "Management1": {
-            "name": "Management1",
-            "interfaceStatus": "connected",
-            "autoNegotiate": "success"
-            }
-        }
-    }
-)
 
 value_parser_tests = [
     value_parser_case_1,
@@ -88,10 +60,6 @@
     keyref_parser_case_4,
 ]
 
-exclude_filter_tests = [
-    exclude_filter_case_1,
-]
-
 
 @pytest.mark.parametrize("path, expected_output", value_parser_tests)
 def test_value_parser(path, expected_output):
@@ -102,8 +70,3 @@ def test_value_parser(path, expected_output):
 def test_keyref_parser(path, expected_output):
     output = jmspath_refkey_parser(path)
     assert expected_output == output, assertion_failed_message.format(output=output, expected_output=expected_output)
-
-@pytest.mark.parametrize("exclude, data, expected_output", exclude_filter_tests)
-def test_exclude_filter(exclude, data, expected_output):
-    exclude_filter(data, exclude)
-    assert expected_output == data, assertion_failed_message.format(output=data, expected_output=expected_output)

netcompare/utils/jmspath/parsers.py

@@ -0,0 +1,37 @@
+from typing import Mapping, List, Generator, Union
+
+def flatten_list(my_list: List) -> List:
+    """
+    Flatten a multi level nested list and returns a list of lists.
+
+    Args:
+        my_list: nested list to be flattened.
+
+    Return:
+        [[-1, 0], [-1, 0], [-1, 0], ...]
+
+    Example:
+        >>> my_list = [[[[-1, 0], [-1, 0]]]]
+        >>> flatten_list(my_list)
+        [[[[-1, 0], [-1, 0]]]]
+    """
+    def iter_flatten_list(my_list: List) -> Generator[List, None, None]:
+        """Recursively yield all flat lists within a given list."""
+        if is_flat_list(my_list):
+            yield my_list
+        else:
+            for item in my_list:
+                yield from iter_flatten_list(item)
+
+
+    def is_flat_list(obj: List) -> bool:
+        """Return True is obj is a list that does not contain any lists as its first order elements."""
+        return isinstance(obj, list) and not any(isinstance(i, list) for i in obj)
+
+    if not isinstance(my_list, list):
+        raise ValueError(f"Argument provided must be a list. You passed a {type(my_list)}")
+    if is_flat_list(my_list):
+        return my_list
+    return list(iter_flatten_list(my_list))
+
+