Merge branch '57-execute-nuts-without-need-of-an-actual-network' into 'master'

Generalise NutsContext and add offline showcase

Closes #57 and #60

See merge request ins/nettowel/nettowel-nuts!27

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_get_users.py

@@ -51,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_network_instances.py

@@ -43,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

@@ -83,11 +83,10 @@ def napalm_ping_multi_dests(self, task: Task, **kwargs) -> Result:
 @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):

nuts/base_tests/netmiko_iperf.py

@@ -76,11 +76,10 @@ def teardown(self) -> None:
 @pytest.mark.usefixtures("check_nuts_result")
 class TestNetmikoIperf:
     @pytest.fixture
-    def single_result(self, nornir_nuts_ctx, host, destination):
-        result = nornir_nuts_ctx.transformed_result()
-        assert host in result, f"Host {host} not found in aggregated result."
-        assert destination in result[host], f"Destination {destination} not found in result."
-        return result[host][destination]
+    def single_result(self, nuts_ctx: NornirNutsContext, host, destination):
+        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,min_expected")
     def test_iperf(self, single_result, min_expected):

nuts/context.py

@@ -1,10 +1,16 @@
 """Provide necessary information that is needed for a specific test."""
-from typing import Any, Callable, Optional
+import pathlib
+from typing import Any, Callable, Optional, Dict, Union
 
+from nornir import InitNornir
 from nornir.core import Nornir
 from nornir.core.task import AggregatedResult
 
 from nuts.helpers.errors import NutsSetupError
+from nuts.helpers.result import NutsResult
+
+
+_TransformedResult = Dict[str, Any]
 
 
 class NutsContext:
@@ -16,6 +22,11 @@ class NutsContext:
 
     def __init__(self, nuts_parameters: Any = None):
         self.nuts_parameters = nuts_parameters or {}
+        self._cached_result: Optional[_TransformedResult] = None
+
+    def initialize(self) -> None:
+        """Initialize dependencies for this context after it has been created."""
+        pass
 
     def nuts_arguments(self) -> dict:
         """
@@ -30,6 +41,30 @@ def nuts_arguments(self) -> dict:
         test_execution = self.nuts_parameters.get("test_execution", None)
         return {**(test_execution if test_execution is not None else {})}
 
+    def general_result(self) -> Any:
+        """
+        :return: raw, unprocessed result
+        """
+        raise NotImplementedError
+
+    def transform_result(self, general_result: Any) -> _TransformedResult:
+        """
+        :param general_result: raw result
+        :return: processed result ready to be passed to a test
+        """
+        raise NotImplementedError
+
+    @property
+    def transformed_result(self) -> _TransformedResult:
+        """
+        The (processed) results of the network task, ready to be passed on to a test's fixture.
+        The results are cached, so that general_result does not need to be called multiple times as it might
+        access the network.
+        """
+        if self._cached_result is None:
+            self._cached_result = self.transform_result(self.general_result())
+        return self._cached_result
+
 
 class NornirNutsContext(NutsContext):
     """
@@ -41,11 +76,20 @@ class NornirNutsContext(NutsContext):
 
     """
 
+    #: The path to a nornir configuration file.
+    #: https://nornir.readthedocs.io/en/stable/configuration/index.html
+    NORNIR_CONFIG_FILE = pathlib.Path("nr-config.yaml")
+
     def __init__(self, nuts_parameters: Any = None):
         super().__init__(nuts_parameters)
-        self._transformed_result = None
         self.nornir: Optional[Nornir] = None
 
+    def initialize(self) -> None:
+        self.nornir = InitNornir(
+            config_file=str(self.NORNIR_CONFIG_FILE),
+            logging={"enabled": False},
+        )
+
     def nuts_task(self) -> Callable:
         """
         Returns the task that nornir should execute for the test module.
@@ -60,14 +104,14 @@ def nornir_filter(self):
         """
         return None
 
-    def transform_result(self, general_result: AggregatedResult) -> Any:
+    def transform_result(self, general_result: AggregatedResult) -> Dict[str, Any]:
         """
         Transforms the raw nornir result and wraps it into a `NutsResult`.
 
         :param general_result: The raw answer as provided by nornir's executed task
-        :return: Usually a dict where keys are the hosts, values are a `NutsResult`
+        :return: A dict where keys are the hosts, values are a `NutsResult`
         """
-        return general_result
+        raise NotImplementedError
 
     def general_result(self) -> AggregatedResult:
         """
@@ -102,13 +146,3 @@ def teardown(self):
         Defines code which is executed after the nornir task.
         """
         pass
-
-    def transformed_result(self) -> Any:
-        """
-        The result from nornir's task, transformed to be passed on later to a test's fixture
-        called `single_result`.
-        :return: The transformed result
-        """
-        if not self._transformed_result:
-            self._transformed_result = self.transform_result(self.general_result())
-        return self._transformed_result

nuts/plugin.py

@@ -7,73 +7,35 @@
 from _pytest.python import Metafunc
 from _pytest.fixtures import FixtureRequest
 from _pytest.config import Config
-from nornir import InitNornir
-from nornir.core import Nornir
 from py._path.local import LocalPath
-from nuts.helpers.errors import NutsSetupError
 
-from nuts.context import NutsContext, NornirNutsContext
+from nuts.context import NutsContext
 from nuts.helpers.result import NutsResult
 from nuts.yamlloader import NutsYamlFile, get_parametrize_data
 
 
-@pytest.fixture(scope="session")
-def nornir_config_file() -> str:
-    """
-    Returns the filename to a nornir configuration file.
-    https://nornir.readthedocs.io/en/stable/configuration/index.html
-
-    :return: The filename of the configuration
-    """
-    return "nr-config.yaml"
-
-
-@pytest.fixture(scope="session")
-def initialized_nornir(nornir_config_file: str) -> Nornir:
-    """
-    Initalizes nornir with a provided configuration file.
-
-    :param nornir_config_file: The filename of a nornir configuration file
-    :return: An initialized nornir instance
-    """
-    return InitNornir(config_file=nornir_config_file, logging={"enabled": False})
-
-
 @pytest.fixture(scope="class")
 def nuts_ctx(request: FixtureRequest) -> NutsContext:
     params = request.node.params
     context_class = getattr(request.module, "CONTEXT", NutsContext)
-    return context_class(params)
-
-
-@pytest.fixture(scope="class")
-def nornir_nuts_ctx(nuts_ctx: NutsContext, initialized_nornir: Nornir) -> NornirNutsContext:
-    """
-    Injects an initialized nornir instance in the context of a test.
-
-    :param nuts_ctx: The context to which the nornir instance should be added
-    :param initialized_nornir: The nornir instance
-    :return: A NornirNutsContext with an initialized nornir instance
-    """
-    if not isinstance(nuts_ctx, NornirNutsContext):
-        raise NutsSetupError("The initialized context does not support the injection of nornir.")
-    nuts_ctx.nornir = initialized_nornir
-    return nuts_ctx
+    ctx = context_class(params)
+    ctx.initialize()
+    return ctx
 
 
 @pytest.fixture
-def single_result(nornir_nuts_ctx: NornirNutsContext, host: str) -> NutsResult:
+def single_result(nuts_ctx: NutsContext, host: str) -> NutsResult:
     """
     Returns the result which belongs to a specific host out of the overall set of results
     that has been returned by nornir's task.
 
     :param nornir_nuts_ctx: The context for a test with an initialized nornir instance
-    :param host: The host for which the corresponding result should be returned
+    :param host: The host from the test bundle (yaml-file) for which the corresponding result should be returned
+    :param destination: The corresponding destination to a host for tests that test a host-destination relationship
     :return: The `NutsResult` that belongs to a host
     """
-    result = nornir_nuts_ctx.transformed_result()
-    assert host in result, f"Host {host} not found in aggregated result."
-    return result[host]
+    assert host in nuts_ctx.transformed_result, f"Host {host} not found in aggregated result."
+    return nuts_ctx.transformed_result[host]
 
 
 @pytest.fixture

tests/base_tests/conftest.py

@@ -4,7 +4,6 @@
 from _pytest.fixtures import FixtureRequest
 from napalm.base.exceptions import ConnectionException
 from nornir.core.task import AggregatedResult
-from nuts.helpers.result import NutsResult
 
 from nuts.context import NornirNutsContext, NutsContext
 

tests/base_tests/test_napalm_bgp_neighbors.py

@@ -83,7 +83,7 @@ def test_contains_hosts_at_toplevel(self, transformed_result, host):
 
     @pytest.mark.parametrize("host, neighbors", [("R1", ["172.16.255.2", "172.16.255.3"]), ("R2", ["172.16.255.1"])])
     def test_contains_peers_at_second_level(self, transformed_result, host, neighbors):
-        assert list(transformed_result[host].result.keys()) == neighbors
+        assert list(transformed_result[host].result) == neighbors
 
     @pytest.mark.parametrize("host, neighbor, details", [("R1", "172.16.255.2", neighbor_details)])
     def test_contains_information_about_neighbor(self, transformed_result, host, neighbor, details):

tests/base_tests/test_napalm_interfaces.py

@@ -76,7 +76,7 @@ def test_contains_host_at_toplevel(self, transformed_result, host):
         [("R1", "GigabitEthernet1"), ("R1", "GigabitEthernet2"), ("R2", "Loopback0"), ("R2", "GigabitEthernet3")],
     )
     def test_contains_interface_names_at_second_level(self, transformed_result, host, interface_name):
-        assert interface_name in transformed_result[host].result.keys()
+        assert interface_name in transformed_result[host].result
 
     @pytest.mark.parametrize(
         "host, name, is_enabled, is_up, mac_address",