Add connection test for Netconf (#427)

* prototype

* Add connection test for Netconf

* bump

Makefile

@@ -10,7 +10,7 @@ NORNIR_DIRS=nornir tests docs
 start_dev_env:
 	${DOCKER_COMPOSE} \
 		up -d \
-		dev1.group_1 dev2.group_1 dev3.group_2 dev4.group_2 dev5.no_group httpbin
+		dev1.group_1 dev2.group_1 dev3.group_2 dev4.group_2 dev5.no_group httpbin netconf1.no_group
 
 .PHONY: stop_dev_env
 stop_dev_env:

docker-compose.yaml

@@ -56,6 +56,13 @@ services:
             net1:
                 ipv4_address: 10.21.33.105
 
+    netconf1.no_group:
+        hostname: netconf1.no_group
+        image: sysrepo/sysrepo-netopeer2:v0.7.7
+        networks:
+            net1:
+                ipv4_address: 10.21.33.106
+
     httpbin:
         hostname: httpbin
         image: bungoume/httpbin-container

nornir/plugins/connections/netconf.py

@@ -103,7 +103,7 @@ def open(
             "host": hostname,
             "username": username,
             "password": password,
-            "port": port,
+            "port": port or 830,
         }
 
         if "ssh_config" not in extras:

test.py

@@ -0,0 +1,55 @@
+from nornir import InitNornir
+from nornir.core.task import Result, Task
+
+nr = InitNornir(
+    inventory={
+        "options": {
+            "hosts": {
+                "rtr00": {
+                    "hostname": "localhost",
+                    "username": "admin",
+                    "password": "admin",
+                    "port": 65030,
+                    "platform": "whatever",
+                    "connection_options": {
+                        "netconf": {"extras": {"hostkey_verify": False}}
+                    },
+                }
+            }
+        }
+    }
+)
+
+
+def netconf_code(task: Task) -> Result:
+    manager = task.host.get_connection("netconf", task.nornir.config)
+
+    # get running config and system state
+    print(manager.get())
+
+    # get only hostname
+    print(manager.get(filter=("xpath", "/sys:system/sys:hostname")))
+
+    # get candidate config
+    print(manager.get_config("candidate"))
+
+    # lock
+    print(manager.lock("candidate"))
+
+    # edit configuration
+    res = manager.edit_config(
+        "candidate",
+        "<sys:system><sys:hostname>asd</sys:hostname></sys:system>",
+        default_operation="merge",
+    )
+    print(res)
+
+    print(manager.commit())
+
+    # unlock
+    print(manager.unlock("candidate"))
+
+    return Result(result="ok", host=task.host)
+
+
+nr.run(task=netconf_code)

tests/conftest.py

@@ -32,3 +32,20 @@ def nornir(request):
 def reset_data():
     global_data.dry_run = True
     global_data.reset_failed_hosts()
+
+
+@pytest.fixture(scope="session", autouse=True)
+def netconf(request):
+    """Initializes nornir"""
+    dir_path = os.path.dirname(os.path.realpath(__file__))
+
+    nornir = InitNornir(
+        inventory={
+            "options": {
+                "host_file": "{}/inventory_data/netconf_hosts.yaml".format(dir_path)
+            }
+        },
+        dry_run=True,
+    )
+    nornir.data = global_data
+    return nornir

tests/inventory_data/netconf_hosts.yaml

@@ -0,0 +1,11 @@
+---
+netconf1.no_group:
+    hostname: netconf1.no_group
+    username: netconf
+    password: netconf
+    connection_options:
+        netconf:
+            extras:
+                allow_agent: False
+                hostkey_verify: False
+                look_for_keys: False

tests/plugins/tasks/networking/test_netconf_connection.py

@@ -0,0 +1,14 @@
+from nornir.core.task import Result, Task
+
+
+def netconf_capabilities(task: Task) -> Result:
+    manager = task.host.get_connection("netconf", task.nornir.config)
+    capabilities = [capability for capability in manager.server_capabilities]
+    return Result(host=task.host, result=capabilities)
+
+
+def test_netconf(netconf):
+    result = netconf.run(netconf_capabilities)
+
+    for _, v in result.items():
+        assert "urn:ietf:params:netconf:capability:writable-running:1.0" in v.result