test: Add backend to run tests with restconf

Very simple tests work (hostname.py), more complex
tests (static_routing.py) failes due to the lack of
model prefix, for example. infix-ip:ipv4 instead of just
ipv4.

We need to fix this before you send your configuration, run
it through libyang for example and get the full model-name.

test/.env

@@ -2,7 +2,7 @@
 # shellcheck disable=SC2034,SC2154
 
 # Current container image
-INFIX_TEST=ghcr.io/kernelkit/infix-test:1.5
+INFIX_TEST=ghcr.io/kernelkit/infix-test:1.6
 
 ixdir=$(readlink -f "$testdir/..")
 logdir=$(readlink -f "$testdir/.log")

test/docker/Dockerfile

@@ -22,6 +22,8 @@ RUN apk add --no-cache \
     sshpass \
     tcpdump \
     tshark \
+    openssl \
+    curl \
     make
 
 ARG MTOOL_VERSION="3.0"

test/docker/pip-requirements.txt

@@ -5,3 +5,4 @@ networkx==3.1
 pydot==1.4.2
 pyyaml==6.0.1
 passlib==1.7.4
+requests==2.25.1

test/infamy/env.py

@@ -5,7 +5,7 @@ import pydot
 import shlex
 import sys
 
-from . import neigh, netconf, ssh, tap, topology
+from . import neigh, netconf, restconf, ssh, tap, topology
 
 class NullEnv:
     def attr(self, _, default=None):
@@ -86,5 +86,9 @@ class Env(object):
             )
         elif protocol == "ssh":
             return ssh.Device(ssh.Location(mgmtip, password))
+        elif protocol == "restconf":
+            return restconf.Device(location=restconf.Location(cport, mgmtip, password),
+                                   mapping=mapping,
+                                   factory_default=factory_default)
 
         raise Exception(f"Unsupported management procotol \"{protocol}\"")

test/infamy/restconf.py

@@ -0,0 +1,121 @@
+import requests
+import json
+import warnings
+
+from requests.auth import HTTPBasicAuth
+from urllib.parse import quote
+from urllib3.exceptions import InsecureRequestWarning
+from dataclasses import dataclass
+
+# We know we have a self-signed certificate, silence warning about it
+warnings.simplefilter('ignore', InsecureRequestWarning)
+
+@dataclass
+class Location:
+    interface: str
+    host: str
+    password: str
+    username: str = "admin"
+    port: int = 443
+
+class Device(object):
+    def __init__(self,
+                 location: Location,
+                 mapping: dict,
+                 factory_default = True):
+        self.location = location
+        self.url = f"https://[{location.host}]:{location.port}/restconf"
+        self.headers = {
+            'Content-Type': 'application/yang-data+json',
+            'Accept': 'application/yang-data+json'
+        }
+        self.auth = HTTPBasicAuth(location.username, location.password)
+        if factory_default:
+            self.factory_default()
+    def get_mgmt_ip(self):
+        return self.location.host
+
+    def get_mgmt_iface(self):
+        return self.location.interface
+
+    def reachable(self):
+        neigh = ll6ping(self.location.interface, flags=["-w1", "-c1", "-L", "-n"])
+        return bool(neigh)
+
+    def get_datastore(self, datastore, xpath="", as_xml=False):
+        path = f"/ds/ietf-datastores:{datastore}/{xpath}"
+        path = quote(path, safe="/:")
+        url = f"{self.url}{path}"
+        try:
+            response = requests.get(url, headers=self.headers, auth=self.auth, verify=False)
+            response.raise_for_status()  # Raise an exception for HTTP errors
+            data = response.json()
+            return data
+        except requests.exceptions.HTTPError as errh:
+            print("HTTP Error:", errh)
+        except requests.exceptions.ConnectionError as errc:
+            print("Error Connecting:", errc)
+        except requests.exceptions.Timeout as errt:
+            print("Timeout Error:", errt)
+        except Exception as err:
+            print("Unknown error", err)
+
+    def put_datastore(self, datastore, data):
+        try:
+            response = requests.put(
+                f"{self.url}/ds/ietf-datastores:{datastore}/",
+                json=data,  # Directly pass the dictionary without using json.dumps
+                headers=self.headers,
+                auth=self.auth,
+                verify=False
+            )
+            response.raise_for_status()  # Raise an exception for HTTP errors
+        except requests.exceptions.HTTPError as errh:
+            print("HTTP Error:", errh)
+        except requests.exceptions.ConnectionError as errc:
+            print("Error Connecting:", errc)
+        except requests.exceptions.Timeout as errt:
+            print("Timeout Error:", errt)
+        except requests.exceptions.RequestException as err:
+            print("Unknown error", err)
+
+    def get_config_dict(self, modname):
+        ds = self.get_datastore("running", modname)
+        model, container = modname.split(":")
+        for k, v in ds.items():
+            return {container: v}
+
+    def merge_dicts(self, dict1, dict2):
+        merged = dict1.copy()  # Start with the first dictionary
+
+        for key, value in dict2.items():
+            if key in merged:
+                if isinstance(merged[key], dict) and isinstance(value, dict):
+                    # If both values are dictionaries, merge them recursively
+                    merged[key] = self.merge_dicts(merged[key], value)
+                else:
+                    # If they are not both dictionaries, overwrite the value from d2
+                    merged[key] = value
+
+        return merged
+
+    def put_config_dict(self, modname, edit):
+        new = {}
+        for k, v in edit.items():
+            new[f"{modname}:{k}"] = v
+
+        running = self.get_datastore("running")
+        running=self.merge_dicts(running, new)
+
+        return self.put_datastore("running", running)
+
+    def get_data(self, xpath=None, as_xml=False):
+        data=self.get_datastore("operational", xpath, as_xml)
+        for k,v in data.items():
+            model, container = k.split(":")
+
+        return {container: v}
+
+    def factory_default(self):
+        factory=self.get_datastore("factory-default")
+        self.put_datastore("running", factory)