Merge branch 'master' into 58-write-wrapper-so-that-path-fspath-deprecation-warning-in-pytest-is-not-triggered

Author: bytinbit

mypy.ini

@@ -1,5 +1,5 @@
 [mypy]
-python_version = 3.6
+python_version = 3.7
 
 ### --strict
 # warn_unused_configs = True

nuts/base_tests/napalm_bgp_neighbors.py

@@ -7,7 +7,8 @@
 from nornir_napalm.plugins.tasks import napalm_get
 
 from nuts.context import NornirNutsContext
-from nuts.helpers.result import nuts_result_wrapper, NutsResult
+from nuts.helpers.filters import filter_hosts
+from nuts.helpers.result import map_host_to_nutsresult, NutsResult
 
 
 class BgpNeighborsContext(NornirNutsContext):
@@ -18,13 +19,10 @@ def nuts_arguments(self) -> Dict[str, List]:
         return {"getters": ["bgp_neighbors"]}
 
     def nornir_filter(self) -> F:
-        hosts = {entry["host"] for entry in self.nuts_parameters["test_data"]}
-        return F(name__any=hosts)
+        return filter_hosts(self.nuts_parameters["test_data"])
 
     def transform_result(self, general_result: AggregatedResult) -> Dict[str, NutsResult]:
-        return {
-            host: nuts_result_wrapper(result, self._transform_host_results) for host, result in general_result.items()
-        }
+        return map_host_to_nutsresult(general_result, self._transform_host_results)
 
     def _transform_host_results(self, single_result: MultiResult) -> dict:
         assert single_result[0].result is not None

nuts/base_tests/napalm_get_users.py

@@ -6,8 +6,10 @@
 from nornir.core.task import MultiResult, AggregatedResult
 from nornir_napalm.plugins.tasks import napalm_get
 
+
 from nuts.context import NornirNutsContext
-from nuts.helpers.result import nuts_result_wrapper, NutsResult
+from nuts.helpers.filters import filter_hosts
+from nuts.helpers.result import map_host_to_nutsresult, NutsResult
 
 
 class UsersContext(NornirNutsContext):
@@ -18,13 +20,10 @@ def nuts_arguments(self) -> Dict[str, List[str]]:
         return {"getters": ["users"]}
 
     def nornir_filter(self) -> F:
-        hosts = {entry["host"] for entry in self.nuts_parameters["test_data"]}
-        return F(name__any=hosts)
+        return filter_hosts(self.nuts_parameters["test_data"])
 
     def transform_result(self, general_result: AggregatedResult) -> Dict[str, NutsResult]:
-        return {
-            host: nuts_result_wrapper(result, self._transform_host_results) for host, result in general_result.items()
-        }
+        return map_host_to_nutsresult(general_result, self._transform_host_results)
 
     def _transform_host_results(self, single_result: MultiResult) -> Dict[str, Dict]:
         assert single_result[0].result is not None
@@ -52,4 +51,4 @@ def test_privilege_level(self, single_result, username, level):
 class TestNapalmOnlyDefinedUsersExist:
     @pytest.mark.nuts("host,usernames")
     def test_no_rogue_users(self, single_result, usernames):
-        assert list(single_result.result.keys()) == usernames
+        assert list(single_result.result) == usernames

nuts/base_tests/napalm_interfaces.py

@@ -6,9 +6,10 @@
 from nornir.core.task import MultiResult, AggregatedResult
 from nornir_napalm.plugins.tasks import napalm_get
 
+
 from nuts.context import NornirNutsContext
-from nuts.helpers.converters import InterfaceNameConverter
-from nuts.helpers.result import nuts_result_wrapper, NutsResult
+from nuts.helpers.filters import filter_hosts
+from nuts.helpers.result import NutsResult, map_host_to_nutsresult
 
 
 class InterfacesContext(NornirNutsContext):
@@ -19,13 +20,10 @@ def nuts_arguments(self) -> Dict[str, List[str]]:
         return {"getters": ["interfaces"]}
 
     def nornir_filter(self) -> F:
-        hosts = {entry["host"] for entry in self.nuts_parameters["test_data"]}
-        return F(name__any=hosts)
+        return filter_hosts(self.nuts_parameters["test_data"])
 
     def transform_result(self, general_result: AggregatedResult) -> Dict[str, NutsResult]:
-        return {
-            host: nuts_result_wrapper(result, self._transform_host_results) for host, result in general_result.items()
-        }
+        return map_host_to_nutsresult(general_result, self._transform_host_results)
 
     def _transform_host_results(self, single_result: MultiResult) -> dict:
         assert single_result[0].result is not None

nuts/base_tests/napalm_lldp_neighbors.py

@@ -8,7 +8,8 @@
 
 from nuts.context import NornirNutsContext
 from nuts.helpers.converters import InterfaceNameConverter
-from nuts.helpers.result import nuts_result_wrapper, NutsResult
+from nuts.helpers.filters import filter_hosts
+from nuts.helpers.result import NutsResult, map_host_to_nutsresult
 
 
 class LldpNeighborsContext(NornirNutsContext):
@@ -19,13 +20,10 @@ def nuts_arguments(self) -> Dict[str, List]:
         return {"getters": ["lldp_neighbors_detail"]}
 
     def nornir_filter(self) -> F:
-        hosts = {entry["host"] for entry in self.nuts_parameters["test_data"]}
-        return F(name__any=hosts)
+        return filter_hosts(self.nuts_parameters["test_data"])
 
     def transform_result(self, general_result: AggregatedResult) -> Dict[str, NutsResult]:
-        return {
-            host: nuts_result_wrapper(result, self._transform_host_results) for host, result in general_result.items()
-        }
+        return map_host_to_nutsresult(general_result, self._transform_host_results)
 
     def _transform_host_results(self, single_result: MultiResult) -> dict:
         assert single_result[0].result is not None

nuts/base_tests/napalm_network_instances.py

@@ -6,7 +6,8 @@
 from nornir.core.task import MultiResult, AggregatedResult
 from nornir_napalm.plugins.tasks import napalm_get
 
-from nuts.helpers.result import nuts_result_wrapper, NutsResult
+from nuts.helpers.filters import filter_hosts
+from nuts.helpers.result import nuts_result_wrapper, NutsResult, map_host_to_nutsresult
 from nuts.context import NornirNutsContext
 
 
@@ -18,15 +19,12 @@ def nuts_arguments(self) -> Dict[str, List[str]]:
         return {"getters": ["network_instances"]}
 
     def nornir_filter(self) -> F:
-        hosts = {entry["host"] for entry in self.nuts_parameters["test_data"]}
-        return F(name__any=hosts)
+        return filter_hosts(self.nuts_parameters["test_data"])
 
     def transform_result(self, general_result: AggregatedResult) -> Dict[str, NutsResult]:
-        return {
-            host: nuts_result_wrapper(result, self._transform_single_result) for host, result in general_result.items()
-        }
+        return map_host_to_nutsresult(general_result, self._transform_host_results)
 
-    def _transform_single_result(self, single_result: MultiResult) -> dict:
+    def _transform_host_results(self, single_result: MultiResult) -> dict:
         assert single_result[0].result is not None
         task_result = single_result[0].result
         network_instances = task_result["network_instances"]
@@ -45,7 +43,7 @@ def _extract_route_distinguisher(self, element: dict) -> str:
         return element["state"]["route_distinguisher"]
 
     def _extract_interfaces(self, element: dict) -> List[str]:
-        return list(element["interfaces"]["interface"].keys())
+        return list(element["interfaces"]["interface"])
 
 
 CONTEXT = NetworkInstancesContext

nuts/base_tests/napalm_ping.py

@@ -1,35 +1,80 @@
 """Let a device ping another device."""
 from enum import Enum
-from typing import Dict, List, Callable
+from typing import Dict, Callable, Any, List
 
 import pytest
 from nornir.core import Task
 from nornir.core.filter import F
-from nornir.core.task import Result, MultiResult, AggregatedResult
+from nornir.core.task import Result, AggregatedResult
 from nornir_napalm.plugins.tasks import napalm_ping
 
 from nuts.context import NornirNutsContext
-from nuts.helpers.result import nuts_result_wrapper, NutsResult
+from nuts.helpers.filters import filter_hosts
+from nuts.helpers.result import (
+    NutsResult,
+    map_host_to_dest_to_nutsresult,
+)
+
+
+class Ping(Enum):
+    FAIL = 0
+    SUCCESS = 1
+    FLAPPING = 2
 
 
 class PingContext(NornirNutsContext):
     def nuts_task(self) -> Callable:
-        return napalm_ping_multi_host
-
-    def nuts_arguments(self) -> dict:
-        args = super().nuts_arguments()
-        args["destinations_per_host"] = _destinations_per_host(self.nuts_parameters["test_data"])
-        return args
+        return self.napalm_ping_multi_dests
 
     def nornir_filter(self) -> F:
-        hosts = {entry["host"] for entry in self.nuts_parameters["test_data"]}
-        return F(name__any=hosts)
+        return filter_hosts(self.nuts_parameters["test_data"])
 
     def transform_result(self, general_result: AggregatedResult) -> Dict[str, Dict[str, NutsResult]]:
-        test_data = self.nuts_parameters["test_data"]
-        return {
-            host: _parse_ping_results(host, task_results, test_data) for host, task_results in general_result.items()
-        }
+        return map_host_to_dest_to_nutsresult(general_result, self._transform_single_entry)
+
+    def _transform_single_entry(self, single_result: Result) -> Ping:
+        assert single_result.host is not None
+        assert single_result.result is not None
+        max_drop = self._allowed_max_drop_for_destination(single_result.host.name, single_result.destination)  # type: ignore[attr-defined] # see below
+        return _map_result_to_enum(single_result.result, max_drop)
+
+    def _allowed_max_drop_for_destination(self, host: str, dest: str) -> int:
+        """
+        Matches the host-destination pair from a nornir task to the
+        host-destination pair from test_data and retrieves its
+        max_drop value that has been defined for that pair.
+
+        :param host: host entry from the nornir task
+        :param dest: destination that was pinged by a host from the nornir task
+        :return: max_drop value from test_data
+        """
+        test_data: List[Dict[str, Any]] = self.nuts_parameters["test_data"]
+        for entry in test_data:
+            if entry["host"] == host and entry["destination"] == dest:
+                return entry["max_drop"]
+        return 0
+
+    def napalm_ping_multi_dests(self, task: Task, **kwargs) -> Result:
+        """
+        One host pings all destinations as defined in the test bundle.
+
+        Note: The destination is not included in the nornir result if the ping fails.
+        Therefore we cannot know which destination was not reachable,
+        so we must patch the destination onto the result object to know later which
+        host-destination pair actually failed.
+
+        :param task: nornir task for ping
+        :param kwargs: arguments from the test bundle for the napalm ping task, such as
+        count, ttl, timeout
+        :return: all pinged destinations per host
+        """
+        destinations_per_hosts = [
+            entry["destination"] for entry in self.nuts_parameters["test_data"] if entry["host"] == task.host.name
+        ]
+        for destination in destinations_per_hosts:
+            result = task.run(task=napalm_ping, dest=destination, **kwargs)
+            result[0].destination = destination  # type: ignore[attr-defined]
+        return Result(host=task.host, result="All pings executed")
 
 
 CONTEXT = PingContext
@@ -38,52 +83,29 @@ def transform_result(self, general_result: AggregatedResult) -> Dict[str, Dict[s
 @pytest.mark.usefixtures("check_nuts_result")
 class TestNapalmPing:
     @pytest.fixture
-    def single_result(self, nornir_nuts_ctx: NornirNutsContext, host: str, destination: str) -> NutsResult:
-        transformed_result = nornir_nuts_ctx.transformed_result()
-        assert host in transformed_result, f"Host {host} not found in aggregated result."
-        assert destination in transformed_result[host], f"Destination {destination} not found in result."
-        return transformed_result[host][destination]
+    def single_result(self, nuts_ctx: NornirNutsContext, host: str, destination: str) -> NutsResult:
+        assert host in nuts_ctx.transformed_result, f"Host {host} not found in aggregated result."
+        assert destination in nuts_ctx.transformed_result[host], f"Destination {destination} not found in result."
+        return nuts_ctx.transformed_result[host][destination]
 
     @pytest.mark.nuts("host,destination,expected")
     def test_ping(self, single_result, expected):
         assert single_result.result.name == expected
 
 
-class Ping(Enum):
-    FAIL = 0
-    SUCCESS = 1
-    FLAPPING = 2
-
-
-def napalm_ping_multi_host(task: Task, destinations_per_host, **kwargs) -> Result:
-    destinations = destinations_per_host(task.host.name)
-    for destination in destinations:
-        result = task.run(task=napalm_ping, dest=destination, **kwargs)
-        result[0].destination = destination  # type: ignore[attr-defined]
-    return Result(host=task.host, result="All pings executed")
-
-
-def _destinations_per_host(test_data):
-    return lambda host_name: [entry["destination"] for entry in test_data if entry["host"] == host_name]
-
-
-def _parse_ping_results(host: str, task_results: MultiResult, test_data: List[dict]) -> dict:
-    maxdrop_per_destination = {
-        entry["destination"]: entry.get("max_drop", 0) for entry in test_data if entry["host"] == host
-    }
-    return {
-        ping_task.destination: nuts_result_wrapper(  # type: ignore[attr-defined]
-            ping_task, _get_transform_single_entry(maxdrop_per_destination[ping_task.destination])  # type: ignore[attr-defined]
-        )
-        for ping_task in task_results[1:]
-    }
-
-
-def _get_transform_single_entry(max_drop):
-    return lambda ping_task: _map_result_to_enum(ping_task.result, max_drop)
+def _map_result_to_enum(result: Dict[str, Dict[Any, Any]], max_drop: int) -> Ping:
+    """
+    Evaluates the ping that has been conducted with nornir and matches it
+    to a Ping-Enum which can be either FAIL, SUCCESS or FLAPPING.
 
+    FAIL: Packet loss equals probes sent.
+    SUCCESS: Packet loss is below or equal max_drop.
+    FLAPPING: Everything else.
 
-def _map_result_to_enum(result: dict, max_drop: int) -> Ping:
+    :param result: a single nornir Result
+    :param max_drop: max_drop threshold
+    :return: evaluated ping result
+    """
     if result["success"]["packet_loss"] == result["success"]["probes_sent"]:
         return Ping.FAIL
     if result["success"]["packet_loss"] <= max_drop:

nuts/base_tests/netmiko_cdp_neighbors.py

@@ -6,7 +6,8 @@
 from nornir.core.task import MultiResult, AggregatedResult
 from nornir_netmiko import netmiko_send_command
 
-from nuts.helpers.result import nuts_result_wrapper, NutsResult
+from nuts.helpers.filters import filter_hosts
+from nuts.helpers.result import NutsResult, map_host_to_nutsresult
 from nuts.context import NornirNutsContext
 
 
@@ -18,17 +19,14 @@ def nuts_arguments(self) -> dict:
         return {"command_string": "show cdp neighbors detail", "use_textfsm": True}
 
     def nornir_filter(self) -> F:
-        hosts = {entry["host"] for entry in self.nuts_parameters["test_data"]}
-        return F(name__any=hosts)
+        return filter_hosts(self.nuts_parameters["test_data"])
 
     def _transform_host_results(self, host_results: MultiResult) -> dict:
         assert host_results[0].result is not None
         return {neighbor["destination_host"]: neighbor for neighbor in host_results[0].result}
 
     def transform_result(self, general_result: AggregatedResult) -> Dict[str, NutsResult]:
-        return {
-            host: nuts_result_wrapper(result, self._transform_host_results) for host, result in general_result.items()
-        }
+        return map_host_to_nutsresult(general_result, self._transform_host_results)
 
 
 CONTEXT = CdpNeighborsContext