Merge pull request #444 from nornir-automation/ogenstad/netconf_plugins

Add netconf plugins

nornir/plugins/tasks/networking/__init__.py

@@ -2,6 +2,9 @@
 from .napalm_configure import napalm_configure
 from .napalm_get import napalm_get
 from .napalm_validate import napalm_validate
+from .netconf_edit_config import netconf_edit_config
+from .netconf_get import netconf_get
+from .netconf_get_config import netconf_get_config
 from .netmiko_commit import netmiko_commit
 from .netmiko_file_transfer import netmiko_file_transfer
 from .netmiko_send_command import netmiko_send_command
@@ -14,6 +17,9 @@
     "napalm_configure",
     "napalm_get",
     "napalm_validate",
+    "netconf_edit_config",
+    "netconf_get",
+    "netconf_get_config",
     "netmiko_commit",
     "netmiko_file_transfer",
     "netmiko_send_command",

nornir/plugins/tasks/networking/netconf_edit_config.py

@@ -0,0 +1,22 @@
+from nornir.core.task import Result, Task
+
+
+def netconf_edit_config(task: Task, config: str, target: str = "running") -> Result:
+    """
+    Edit configuration of device using Netconf
+
+    Arguments:
+        config: Configuration snippet to apply
+        target: Target configuration store
+
+    Examples:
+
+        Simple example::
+
+            > nr.run(task=netconf_edit_config, config=desired_config)
+
+    """
+    manager = task.host.get_connection("netconf", task.nornir.config)
+    manager.edit_config(config, target=target)
+
+    return Result(host=task.host)

nornir/plugins/tasks/networking/netconf_get.py

@@ -0,0 +1,42 @@
+from nornir.core.task import Result, Task
+
+
+def netconf_get(task: Task, path: str = "", filter_type: str = "xpath") -> Result:
+    """
+    Get information over Netconf from device
+
+    Arguments:
+        path: Subtree or xpath to filter
+        filter_type: Type of filtering to use, 'xpath' or 'subtree'
+
+    Examples:
+
+        Simple example::
+
+            > nr.run(task=netconf_get)
+
+        Passing options using ``xpath``::
+
+            > query = "/devices/device"
+            > nr.run(task=netconf_get,
+            >        path=query)
+
+       Passing options using ``subtree``::
+
+            > query = "<interfaces></interfaces>"
+            > nr.run(task=netconf_get,
+            >        filter_type="subtree",
+            >        path=query)
+
+
+    Returns:
+        Result object with the following attributes set:
+          * result (``str``): The collected data as an XML string
+    """
+    manager = task.host.get_connection("netconf", task.nornir.config)
+    params = {}
+    if path:
+        params["filter"] = (filter_type, path)
+    result = manager.get(**params)
+
+    return Result(host=task.host, result=result.data_xml)

nornir/plugins/tasks/networking/netconf_get_config.py

@@ -0,0 +1,54 @@
+from typing import Any, Dict
+
+from nornir.core.task import Result, Task
+
+
+def netconf_get_config(
+    task: Task, source: str = "running", path: str = "", filter_type: str = "xpath"
+) -> Result:
+    """
+    Get configuration over Netconf from device
+
+    Arguments:
+        source: Configuration store to collect from
+        path: Subtree or xpath to filter
+        filter_type: Type of filtering to use, 'xpath' or 'subtree'
+
+    Examples:
+
+        Simple example::
+
+            > nr.run(task=netconf_get_config)
+
+        Collect startup config::
+
+            > nr.run(task=netconf_get_config, source="startup")
+
+
+        Passing options using ``xpath``::
+
+            > xpath = /devices/device"
+            > nr.run(task=netconf_get_config,
+            >        path=xpath)
+
+        Passing options using ``subtree``::
+
+            > xpath = /devices/device"
+            > nr.run(task=netconf_get_config,
+            >        filter_type="subtree",
+            >        path=subtree)
+
+
+    Returns:
+        Result object with the following attributes set:
+          * result (``str``): The collected data as an XML string
+    """
+    manager = task.host.get_connection("netconf", task.nornir.config)
+    parameters: Dict[str, Any] = {"source": source}
+
+    if path:
+        parameters["filter"] = (filter_type, path)
+
+    result = manager.get_config(**parameters)
+
+    return Result(host=task.host, result=result.data_xml)

tests/plugins/tasks/networking/test_netconf_edit_config.py

@@ -0,0 +1,42 @@
+from nornir.plugins.tasks import networking
+
+CONFIG = """
+<nc:config
+    xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0">
+    <turing-machine
+        xmlns="http://example.net/turing-machine">
+        <transition-function>
+            <delta nc:operation="{operation}">
+                <label>this-is-nornir</label>
+                <input>
+                    <symbol>4</symbol>
+                    <state>1</state>
+                </input>
+            </delta>
+        </transition-function>
+    </turing-machine>
+</nc:config>
+"""
+
+
+def test_netconf_edit_config(netconf):
+    result = netconf.run(networking.netconf_get_config)
+
+    for _, v in result.items():
+        assert "nornir" not in v.result
+
+    netconf.run(networking.netconf_edit_config, config=CONFIG.format(operation="merge"))
+
+    result = netconf.run(networking.netconf_get_config)
+
+    for _, v in result.items():
+        assert "nornir" in v.result
+
+    netconf.run(
+        networking.netconf_edit_config, config=CONFIG.format(operation="delete")
+    )
+
+    result = netconf.run(networking.netconf_get_config)
+
+    for _, v in result.items():
+        assert "nornir" not in v.result

tests/plugins/tasks/networking/test_netconf_get.py

@@ -0,0 +1,19 @@
+from nornir.plugins.tasks import networking
+
+
+def test_netconf_get(netconf):
+    result = netconf.run(networking.netconf_get)
+
+    for _, v in result.items():
+        assert "<turing-machine" in v.result
+
+
+def test_netconf_get_subtree(netconf):
+    result = netconf.run(
+        networking.netconf_get,
+        path="<turing-machine></turing-machine>",
+        filter_type="subtree",
+    )
+
+    for _, v in result.items():
+        assert "<turing-machine" in v.result

tests/plugins/tasks/networking/test_netconf_get_config.py

@@ -0,0 +1,20 @@
+from nornir.plugins.tasks import networking
+
+
+def test_netconf_get_config(netconf):
+    result = netconf.run(networking.netconf_get_config, source="startup")
+
+    for _, v in result.items():
+        assert "<turing-machine" in v.result
+
+
+def test_netconf_get_config_subtree(netconf):
+    result = netconf.run(
+        networking.netconf_get_config,
+        source="startup",
+        path="<interfaces></interfaces>",
+        filter_type="subtree",
+    )
+
+    for _, v in result.items():
+        assert "<interface" in v.result