test/infamy: Initial add of python based test library

test/infamy/.gitignore

@@ -0,0 +1 @@
+/__pycache__

test/infamy/__init__.py

@@ -0,0 +1,14 @@
+import os
+
+from .env import Env
+from .netns import IsolatedMacVlan
+from .tap import Test
+
+def std_topology(name):
+    return os.path.realpath(
+        os.path.join(
+            os.path.dirname(__file__),
+            "topologies",
+            name + ".dot"
+        )
+    )

test/infamy/env.py

@@ -0,0 +1,93 @@
+import argparse
+import networkx
+import os
+import pydot
+import shlex
+import sys
+
+from . import neigh, netconf, tap, topology
+
+class ArgumentParser(argparse.ArgumentParser):
+    def __init__(self, ltop):
+        super().__init__()
+
+        self.add_argument("-d", "--debug", default=False, action="store_true")
+        self.add_argument("-l", "--logical-topology", dest="ltop", default=ltop)
+        self.add_argument("-y", "--yangdir", default=None)
+        self.add_argument("ptop", nargs=1, metavar="topology")
+
+
+class Env(object):
+    def __init__(self, ltop=None, argv=sys.argv[1::], environ=os.environ):
+        if "INFAMY_ARGS" in environ:
+            argv = shlex.split(environ["INFAMY_ARGS"]) + argv
+
+        self.args = ArgumentParser(ltop).parse_args(argv)
+
+        pdot = pydot.graph_from_dot_file(self.args.ptop[0])[0]
+        self.ptop = networkx.nx_pydot.read_dot(self.args.ptop[0])
+
+        if self.args.ltop:
+            self.ltop = networkx.nx_pydot.read_dot(self.args.ltop)
+            ldot = pydot.graph_from_dot_file(self.args.ltop)[0]
+            mapping = topology.find_mapping(pdot, ldot)
+            if not mapping:
+                raise tap.TestSkip()
+
+            self.mapping = {}
+            for (log, phy) in mapping.items():
+                if ":" in log:
+                    lnode, lport = log.split(":")
+                    pnode, pport = phy.split(":")
+
+                    self._map_node(lnode, pnode)
+                    self._map_port((lnode, lport), (pnode, pport))
+                else:
+                    self._map_node(log, phy)
+
+    def _map_node(self, lnode, pnode):
+        if lnode in self.mapping:
+            assert(self.mapping[lnode][None] == pnode)
+        else:
+            self.mapping[lnode] = { None: pnode }
+
+    def _map_port(self, log, phy):
+        (lnode, lport) = log
+        (pnode, pport) = phy
+
+        if lport in self.mapping[lnode]:
+            assert(self.mapping[lnode][lport] == pport)
+        else:
+            self.mapping[lnode][lport] = pport
+
+    def xlate(self, lnode, lport=None):
+        if lnode not in self.mapping:
+            return None
+
+        nodemap = self.mapping[lnode]
+
+        if lport not in nodemap:
+            return None
+
+        if not lport:
+            return nodemap[None]
+
+        return (nodemap[None], nodemap[lport])
+
+
+    def attach(self, node, port):
+        if self.mapping:
+            mapping = self.mapping[node]
+            node, port = mapping[None], mapping[port]
+        else:
+            mapping = None
+
+        hostport = list(self.ptop.neighbors(f"{node}:{port}"))[0]
+        hnode, hport = hostport.split(":")
+
+        mgmtip = neigh.ll6ping(hport)
+        return netconf.Device(
+            location=netconf.Location(mgmtip),
+            mapping=mapping,
+            yangdir=self.args.yangdir
+        )

test/infamy/neigh.py

@@ -0,0 +1,18 @@
+import json
+import re
+import subprocess
+
+def ll6ping(ifname, flags=["-w60", "-c1", "-L", "-n"]):
+    argv = ["ping"] + flags + ["ff02::1%{}".format(ifname)]
+
+    try:
+        ping = subprocess.run(argv,
+                              stdout=subprocess.PIPE,
+                              stderr=subprocess.DEVNULL,
+                              text=True, check=True)
+    except subprocess.CalledProcessError:
+        return None
+
+    m = re.search("^\d+ bytes from ([:0-9a-f]+%\S+):", ping.stdout, re.MULTILINE)
+    return m.group(1) if m else None
+

test/infamy/netconf.py

@@ -0,0 +1,116 @@
+
+from collections import namedtuple
+from dataclasses import dataclass
+
+import socket
+
+import libyang
+import lxml
+import netconf_client.connect
+import netconf_client.ncclient
+
+modinfo_fields = ("identifier", "version", "format", "namespace")
+ModInfoTuple = namedtuple("ModInfoTuple", modinfo_fields)
+class ModInfo(ModInfoTuple):
+    def xmlns(self):
+        return f"xmlns:{self.identifier}=\"{self.namespace}\""
+
+NS = {
+    "ietf-netconf-monitoring": "urn:ietf:params:xml:ns:yang:ietf-netconf-monitoring",
+}
+
+@dataclass
+class Location:
+    host: str
+    port: int = 830
+    username: str = "admin"
+    password: str = "admin"
+
+class Device(object):
+    def __init__(self,
+                 location: Location,
+                 mapping: dict,
+                 yangdir: None | str = None):
+
+        self.mapping = mapping
+        self.ly = libyang.Context(yangdir)
+
+        self._ncc_init(location)
+        self._ly_init(yangdir)
+        # self.update_schema()
+
+    def _ncc_init(self, location):
+        ai = socket.getaddrinfo(location.host, location.port,
+                                0, 0, socket.SOL_TCP)
+        sock = socket.socket(ai[0][0], ai[0][1], 0)
+        sock.settimeout(60)
+        sock.connect(ai[0][4])
+        sock.settimeout(None)
+
+        session = netconf_client.connect.connect_ssh(sock=sock,
+                                                     username=location.username,
+                                                     password=location.password)
+        self.ncc = netconf_client.ncclient.Manager(session)
+
+    def _ly_init(self, yangdir):
+        self.ly = libyang.Context(yangdir)
+
+        lib = self.ly.load_module("ietf-yang-library")
+        ns = libyang.util.c2str(lib.cdata.ns)
+
+        xml = lxml.etree.tostring(self.ncc.get(filter=f"""
+        	<filter type="subtree">
+        		<modules-state xmlns="{ns}" />
+        	</filter>""").data_ele[0])
+
+        data = self.ly.parse_data("xml", libyang.IOType.MEMORY,
+                                  xml, parse_only=True).print_dict()
+
+        self.modules = { m["name"] : m for m in data["modules-state"]["module"] }
+
+        for ms in self.modules.values():
+            if ms["conformance-type"] != "implement":
+                continue
+
+            mod = self.ly.load_module(ms["name"])
+
+            # TODO: ms["feature"] contains the list of enabled
+            # features, so ideally we should only enable the supported
+            # ones. However, features can depend on each other, so the
+            # naïve looping approach doesn't work.
+            mod.feature_enable_all()
+
+    def _modules_in_xpath(self, xpath):
+        modnames = []
+
+        # Find all referenced models
+        for seg in xpath.split("/"):
+            if ":" in seg:
+                modname, node = seg.split(":")
+                modnames.append(modname)
+
+        return list(filter(lambda m: m["name"] in modnames,
+                           self.modules.values()))
+
+    def get_config(self, xpath):
+        # Figure out which modules we are referencing
+        mods = self._modules_in_xpath(xpath)
+
+        # Fetch the data
+        xmlns = " ".join([f"xmlns:{m['name']}=\"{m['namespace']}\"" for m in mods])
+        filt = f"<filter type=\"xpath\" select=\"{xpath}\" {xmlns} />"
+        cfg = lxml.etree.tostring(self.ncc.get_config(filter=filt).data_ele[0])
+
+        return self.ly.parse_data_mem(cfg, "xml", parse_only=True)
+
+    def get_config_dict(self, xpath):
+        return self.get_config(xpath).print_dict()
+
+    def put_config(self, edit):
+        xml = "<config xmlns=\"urn:ietf:params:xml:ns:netconf:base:1.0\">" + edit + "</config>"
+        self.ncc.edit_config(xml)
+
+    def put_config_dict(self, modname, edit):
+        mod = self.ly.get_module(modname)
+        lyd = mod.parse_data_dict(edit, no_state=True)
+        return self.put_config(lyd.print_mem("xml", with_siblings=True, pretty=False))

test/infamy/netns.py

@@ -0,0 +1,68 @@
+import subprocess
+
+class IsolatedMacVlan:
+    def __init__(self, parent, ifname="iface", lo=True):
+        self.parent, self.ifname, self.lo = parent, ifname, lo
+
+    def __enter__(self):
+        self.sleeper = subprocess.Popen(["unshare", "-r", "-n", "sh", "-c",
+                                         "echo && exec sleep infinity"],
+                                        stdout=subprocess.PIPE)
+        self.sleeper.stdout.readline()
+
+        try:
+            subprocess.run(["ip", "link", "add",
+                            "dev", self.ifname,
+                            "link", self.parent,
+                            "netns", str(self.sleeper.pid),
+                            "type", "macvlan"], check=True)
+            self.runsh(f"""
+            while ! ip link show dev {self.ifname}; do
+                sleep 0.1
+            done
+            """)
+        except Exception as e:
+            self.__exit__(self, None, None, None)
+            raise e
+
+        if self.lo:
+            try:
+                self.run(["ip", "link", "set", "dev", "lo", "up"])
+            except Exception as e:
+                self.__exit__(self, None, None, None)
+                raise e
+
+        return self
+
+    def __exit__(self, val, typ, tb):
+        self.sleeper.kill()
+        self.sleeper.wait()
+
+    def _mangle_subprocess_args(self, args, kwargs):
+        if args:
+            args = list(args)
+            if type(args[0]) == str:
+                if "shell" in kwargs and kwargs["shell"]:
+                    args[0] = ["/bin/sh", "-c", args[0]]
+                    kwargs["shell"] = False
+                else:
+                    args[0] = [args[0]]
+
+            if type(args[0]) == list:
+                args[0] = ["nsenter", "-t", str(self.sleeper.pid),
+                           "-n", "-U", "--preserve-credentials"] + args[0]
+                return args, kwargs
+
+        raise ValueError("Unable mangle subprocess arguments")
+
+    def run(self, *args, **kwargs):
+        args, kwargs = self._mangle_subprocess_args(args, kwargs)
+        return subprocess.run(*args, **kwargs)
+
+    def popen(self, *args, **kwargs):
+        args, kwargs = self._mangle_subprocess_args(args, kwargs)
+        return subprocess.Popen(*args, **kwargs)
+
+    def runsh(self, script):
+        return self.run("/bin/sh", text=True, input=script,
+                        stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

test/infamy/tap.py

@@ -0,0 +1,111 @@
+import contextlib
+import datetime
+import sys
+import traceback
+
+class Test:
+    def __init__(self, output=sys.stdout):
+        self.out = output
+        self.commenter = CommentWriter(self.out)
+        if self.out == sys.stdout:
+            sys.stdout = self.commenter
+
+        self.steps = 0
+
+    def __enter__(self):
+        now = datetime.datetime.now().strftime("%F %T")
+        self.out.write(f"# Starting ({now})\n")
+        return self
+
+    def __exit__(self, _, e, __):
+        now = datetime.datetime.now().strftime("%F %T")
+        self.out.write(f"# Exiting ({now})\n")
+
+        if not e:
+            self._not_ok("Missing explicit test result\n")
+        else:
+            if type(e) in (TestPass, TestSkip):
+                self.out.write(f"{self.steps}..{self.steps}\n")
+                raise SystemExit(0)
+
+            traceback.print_exception(e, file=self.commenter)
+
+        raise SystemExit(1)
+
+    @contextlib.contextmanager
+    def step(self, msg):
+        try:
+            yield
+            self._ok(msg)
+
+        except Exception as e:
+            if type(e) == TestPass:
+                self._ok(msg)
+            elif type(e) == TestSkip:
+                self._ok(directive="skip")
+            elif type(e) == TestFail:
+                self._not_ok(msg)
+            else:
+                self._not_ok(msg)
+
+            raise e
+
+    def _report(self, tag, msg):
+        self.steps += 1
+        self.out.write(f"{tag} {self.steps}{msg}\n")
+
+    def _ok(self, msg="", directive=None):
+        if msg:
+            msg = " - " + msg
+
+        if directive:
+            msg = msg + " # " + directive
+
+        self._report("ok", msg)
+
+    def _not_ok(self, msg):
+        self._report("not ok", " - " + msg)
+
+    def succeed(self):
+        raise TestPass()
+
+    def skip(self):
+        raise TestSkip()
+
+    def fail(self):
+        raise TestFail()
+
+class TestResult(Exception):
+    pass
+
+class TestPass(TestResult):
+    pass
+class TestFail(TestResult):
+    pass
+class TestSkip(TestResult):
+    pass
+
+class CommentWriter:
+    def __init__(self, f):
+        self.f = f
+        self.at_nl = True
+
+    def write(self, data):
+        if self.at_nl:
+            data = "# " + data
+            self.at_nl = False
+            if not len(data):
+                return
+
+        if data.endswith("\n"):
+            self.at_nl = True
+            data = data[:-1]
+
+        data = data.replace("\n", "\n# ")
+        if self.at_nl:
+            data = data + "\n"
+
+        self.f.write(data)
+
+    def flush(self):
+        return self.f.flush()

test/infamy/topologies/1x1.dot

@@ -0,0 +1,23 @@
+graph "1x1" {
+	layout="neato";
+	overlap="false";
+	esep="+20";
+
+        node [shape=record, fontname="monospace"];
+	edge [color="cornflowerblue", penwidth="2"];
+
+	host [
+	    label="host | { <tgt> tgt }",
+	    pos="0,12!",
+	    kind="controller",
+	];
+
+        target [
+	    label="{ <mgmt> mgmt } | target",
+	    pos="10,12!",
+
+	    kind="infix",
+	];
+
+	host:tgt -- target:mgmt
+}

test/infamy/topologies/1x2.dot

@@ -0,0 +1,24 @@
+graph "1x2" {
+	layout="neato";
+	overlap="false";
+	esep="+20";
+
+        node [shape=record, fontname="monospace"];
+	edge [color="cornflowerblue", penwidth="2"];
+
+	host [
+	    label="host | { <tgt> tgt | <data> | data }",
+	    pos="0,12!",
+	    kind="controller",
+	];
+
+        target [
+	    label="{ <mgmt> mgmt | <data> data } | target",
+	    pos="10,12!",
+
+	    kind="infix",
+	];
+
+	host:tgt -- target:mgmt
+	host:data -- target:data
+}

test/infamy/topology.py

@@ -0,0 +1,68 @@
+import networkx as nx
+from networkx.algorithms import isomorphism
+
+def qstrip(text):
+    if text.startswith("\"") and text.endswith("\""):
+        return text[1:-1]
+    return text
+
+def match_kind(n1, n2):
+    return qstrip(n1["kind"]) == qstrip(n2["kind"])
+
+def find_mapping(phy, log, node_match=match_kind):
+    def annotate(nxg, dotg):
+        for e in list(nxg.edges):
+            s, d = e
+            nxg.nodes[s]["kind"] = "port"
+            nxg.nodes[d]["kind"] = "port"
+
+            sn, sp = s.split(":")
+            dn, dp = d.split(":")
+
+            try:
+                sk = dotg.get_node(sn)[0].get_attributes()["kind"]
+            except:
+                raise ValueError("\"{}\"'s kind is not known".format(sn))
+
+            try:
+                dk = dotg.get_node(dn)[0].get_attributes()["kind"]
+            except:
+                raise ValueError("\"{}\"'s kind is not known".format(dn))
+
+            nxg.add_node(sn, kind=sk)
+            nxg.add_edge(sn, s)
+            nxg.add_node(dn, kind=dk)
+            nxg.add_edge(dn, d)
+
+    phyedges = [(e.get_source(), e.get_destination()) for e in phy.get_edges()]
+    logedges = [(e.get_source(), e.get_destination()) for e in log.get_edges()]
+
+    phyedges.sort()
+    logedges.sort()
+    nxphy = nx.Graph(phyedges)
+    nxlog = nx.Graph(logedges)
+    annotate(nxphy, phy)
+    annotate(nxlog, log)
+
+    nxmap = isomorphism.GraphMatcher(nxphy, nxlog, node_match=node_match)
+    if nxmap.subgraph_is_isomorphic():
+        return { v: k for (k, v) in nxmap.mapping.items() }
+
+    return None
+
+# This let's us call this script like so...
+#
+#     python3 topology.py <physical> <logical>
+#
+# to inspect the graph matcher's results in isolation from the rest of
+# the system.
+if __name__ == "__main__":
+    import json
+    import pydot
+    import sys
+
+    pdot = pydot.graph_from_dot_file(sys.argv[1])[0]
+    ldot = pydot.graph_from_dot_file(sys.argv[2])[0]
+    mapping = find_mapping(pdot, ldot)
+
+    print(json.dumps(mapping))