Adding upstream version 9.1.upstream/9.1

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 499 insertions, 0 deletions

diff --git a/tests/topotests/lib/lutil.py b/tests/topotests/lib/lutil.py
new file mode 100644
index 0000000..1eb88f2
--- /dev/null
+++ b/tests/topotests/lib/lutil.py
@@ -0,0 +1,499 @@
+#!/usr/bin/env python
+# SPDX-License-Identifier: GPL-2.0-or-later
+
+# Copyright 2017, LabN Consulting, L.L.C.
+
+import os
+import re
+import sys
+import time
+import json
+import math
+import time
+from lib.topolog import logger
+from lib.topotest import json_cmp
+
+
+# L utility functions
+#
+# These functions are inteneted to provide support for CI testing within MiniNet
+# environments.
+
+
+class lUtil:
+    # to be made configurable in the future
+    base_script_dir = "."
+    base_log_dir = "."
+    fout_name = "output.log"
+    fsum_name = "summary.txt"
+    l_level = 6
+    CallOnFail = False
+
+    l_total = 0
+    l_pass = 0
+    l_fail = 0
+    l_filename = ""
+    l_last = None
+    l_line = 0
+    l_dotall_experiment = False
+    l_last_nl = None
+    l_wait_strict = 1
+
+    fout = ""
+    fsum = ""
+    net = ""
+
+    def log(self, str, level=6):
+        if self.l_level > 0:
+            if self.fout == "":
+                self.fout = open(self.fout_name, "w")
+            self.fout.write(str + "\n")
+        if level <= self.l_level:
+            print(str)
+
+    def summary(self, str):
+        if self.fsum == "":
+            self.fsum = open(self.fsum_name, "w")
+            self.fsum.write(
+                "\
+******************************************************************************\n"
+            )
+            self.fsum.write(
+                "\
+Test Target Summary                                                  Pass Fail\n"
+            )
+            self.fsum.write(
+                "\
+******************************************************************************\n"
+            )
+        self.fsum.write(str + "\n")
+
+    def result(self, target, success, str, logstr=None):
+        if success:
+            p = 1
+            f = 0
+            self.l_pass += 1
+            sstr = "PASS"
+        else:
+            f = 1
+            p = 0
+            self.l_fail += 1
+            sstr = "FAIL"
+        self.l_total += 1
+        if logstr != None:
+            self.log("R:%d %s: %s" % (self.l_total, sstr, logstr))
+        res = "%-4d %-6s %-56s %-4d %d" % (self.l_total, target, str, p, f)
+        self.log("R:" + res)
+        self.summary(res)
+        if f == 1 and self.CallOnFail != False:
+            self.CallOnFail()
+
+    def closeFiles(self):
+        ret = (
+            "\
+******************************************************************************\n\
+Total %-4d                                                           %-4d %d\n\
+******************************************************************************"
+            % (self.l_total, self.l_pass, self.l_fail)
+        )
+        if self.fsum != "":
+            self.fsum.write(ret + "\n")
+            self.fsum.close()
+            self.fsum = ""
+        if self.fout != "":
+            if os.path.isfile(self.fsum_name):
+                r = open(self.fsum_name, "r")
+                self.fout.write(r.read())
+                r.close()
+            self.fout.close()
+            self.fout = ""
+        return ret
+
+    def setFilename(self, name):
+        str = "FILE: " + name
+        self.log(str)
+        self.summary(str)
+        self.l_filename = name
+        self.line = 0
+
+    def getCallOnFail(self):
+        return self.CallOnFail
+
+    def setCallOnFail(self, CallOnFail):
+        self.CallOnFail = CallOnFail
+
+    def strToArray(self, string):
+        a = []
+        c = 0
+        end = ""
+        words = string.split()
+        if len(words) < 1 or words[0].startswith("#"):
+            return a
+        words = string.split()
+        for word in words:
+            if len(end) == 0:
+                a.append(word)
+            else:
+                a[c] += str(" " + word)
+            if end == "\\":
+                end = ""
+            if not word.endswith("\\"):
+                if end != '"':
+                    if word.startswith('"'):
+                        end = '"'
+                    else:
+                        c += 1
+                else:
+                    if word.endswith('"'):
+                        end = ""
+                        c += 1
+                    else:
+                        c += 1
+            else:
+                end = "\\"
+        #        if len(end) == 0:
+        #            print('%d:%s:' % (c, a[c-1]))
+
+        return a
+
+    def execTestFile(self, tstFile):
+        if os.path.isfile(tstFile):
+            f = open(tstFile)
+            for line in f:
+                if len(line) > 1:
+                    a = self.strToArray(line)
+                    if len(a) >= 6:
+                        luCommand(a[1], a[2], a[3], a[4], a[5])
+                    else:
+                        self.l_line += 1
+                        self.log("%s:%s %s" % (self.l_filename, self.l_line, line))
+                        if len(a) >= 2:
+                            if a[0] == "sleep":
+                                time.sleep(int(a[1]))
+                            elif a[0] == "include":
+                                self.execTestFile(a[1])
+            f.close()
+        else:
+            self.log("unable to read: " + tstFile)
+            sys.exit(1)
+
+    def command(
+        self,
+        target,
+        command,
+        regexp,
+        op,
+        result,
+        returnJson,
+        startt=None,
+        force_result=False,
+    ):
+        global net
+        if op == "jsoncmp_pass" or op == "jsoncmp_fail":
+            returnJson = True
+
+        self.log(
+            "%s (#%d) %s:%s COMMAND:%s:%s:%s:%s:%s:"
+            % (
+                time.asctime(),
+                self.l_total + 1,
+                self.l_filename,
+                self.l_line,
+                target,
+                command,
+                regexp,
+                op,
+                result,
+            )
+        )
+        if self.net == "":
+            return False
+        # self.log("Running %s %s" % (target, command))
+        js = None
+        out = self.net[target].cmd(command).rstrip()
+        if len(out) == 0:
+            report = "<no output>"
+        else:
+            report = out
+            if returnJson == True:
+                try:
+                    js = json.loads(out)
+                except:
+                    js = None
+                    self.log(
+                        "WARNING: JSON load failed -- confirm command output is in JSON format."
+                    )
+        self.log("COMMAND OUTPUT:%s:" % report)
+
+        # JSON comparison
+        if op == "jsoncmp_pass" or op == "jsoncmp_fail":
+            try:
+                expect = json.loads(regexp)
+            except:
+                expect = None
+                self.log(
+                    "WARNING: JSON load failed -- confirm regex input is in JSON format."
+                )
+            json_diff = json_cmp(js, expect)
+            if json_diff != None:
+                if op == "jsoncmp_fail":
+                    success = True
+                else:
+                    success = False
+                    self.log("JSON DIFF:%s:" % json_diff)
+                ret = success
+            else:
+                if op == "jsoncmp_fail":
+                    success = False
+                else:
+                    success = True
+            self.result(target, success, result)
+            if js != None:
+                return js
+            return ret
+
+        # Experiment: can we achieve the same match behavior via DOTALL
+        # without converting newlines to spaces?
+        out_nl = out
+        search_nl = re.search(regexp, out_nl, re.DOTALL)
+        self.l_last_nl = search_nl
+        # Set up for comparison
+        if search_nl != None:
+            group_nl = search_nl.group()
+            group_nl_converted = " ".join(group_nl.splitlines())
+        else:
+            group_nl_converted = None
+
+        out = " ".join(out.splitlines())
+        search = re.search(regexp, out)
+        self.l_last = search
+        if search == None:
+            if op == "fail":
+                success = True
+            else:
+                success = False
+            ret = success
+        else:
+            ret = search.group()
+            if op != "fail":
+                success = True
+                level = 7
+            else:
+                success = False
+                level = 5
+            self.log("found:%s:" % ret, level)
+            # Experiment: compare matched strings obtained each way
+            if self.l_dotall_experiment and (group_nl_converted != ret):
+                self.log(
+                    "DOTALL experiment: strings differ dotall=[%s] orig=[%s]"
+                    % (group_nl_converted, ret),
+                    9,
+                )
+        if startt != None:
+            if js != None or ret is not False or force_result is not False:
+                delta = time.time() - startt
+                self.result(target, success, "%s +%4.2f secs" % (result, delta))
+        elif op == "pass" or op == "fail":
+            self.result(target, success, result)
+        if js != None:
+            return js
+        return ret
+
+    def wait(
+        self, target, command, regexp, op, result, wait, returnJson, wait_time=0.5
+    ):
+        self.log(
+            "%s:%s WAIT:%s:%s:%s:%s:%s:%s:%s:"
+            % (
+                self.l_filename,
+                self.l_line,
+                target,
+                command,
+                regexp,
+                op,
+                result,
+                wait,
+                wait_time,
+            )
+        )
+        found = False
+        n = 0
+        startt = time.time()
+
+        if (op == "wait-strict") or ((op == "wait") and self.l_wait_strict):
+            strict = True
+        else:
+            strict = False
+
+        # Calculate the amount of `sleep`s we are going to peform.
+        wait_count = int(math.ceil(wait / wait_time)) + 1
+
+        force_result = False
+        while wait_count > 0:
+            n += 1
+
+            # log a failure on last iteration if we don't get desired regexp
+            if strict and (wait_count == 1):
+                force_result = True
+
+            found = self.command(
+                target, command, regexp, op, result, returnJson, startt, force_result
+            )
+            if found is not False:
+                break
+
+            wait_count -= 1
+            if wait_count > 0:
+                time.sleep(wait_time)
+
+        delta = time.time() - startt
+        self.log("Done after %d loops, time=%s, Found=%s" % (n, delta, found))
+        return found
+
+
+# initialized by luStart
+LUtil = None
+
+
+# entry calls
+def luStart(
+    baseScriptDir=".",
+    baseLogDir=".",
+    net="",
+    fout="output.log",
+    fsum="summary.txt",
+    level=None,
+):
+    global LUtil
+    # init class
+    LUtil = lUtil()
+    LUtil.base_script_dir = baseScriptDir
+    LUtil.base_log_dir = baseLogDir
+    LUtil.net = net
+    if fout != "":
+        LUtil.fout_name = baseLogDir + "/" + fout
+    if fsum != None:
+        LUtil.fsum_name = baseLogDir + "/" + fsum
+    if level != None:
+        LUtil.l_level = level
+    LUtil.l_dotall_experiment = False
+    LUtil.l_dotall_experiment = True
+
+
+def luCommand(
+    target,
+    command,
+    regexp=".",
+    op="none",
+    result="",
+    time=10,
+    returnJson=False,
+    wait_time=0.5,
+):
+    waitops = ["wait", "wait-strict", "wait-nostrict"]
+
+    if op in waitops:
+        return LUtil.wait(
+            target, command, regexp, op, result, time, returnJson, wait_time
+        )
+    else:
+        return LUtil.command(target, command, regexp, op, result, returnJson)
+
+
+def luLast(usenl=False):
+    if usenl:
+        if LUtil.l_last_nl != None:
+            LUtil.log("luLast:%s:" % LUtil.l_last_nl.group(), 7)
+        return LUtil.l_last_nl
+    else:
+        if LUtil.l_last != None:
+            LUtil.log("luLast:%s:" % LUtil.l_last.group(), 7)
+        return LUtil.l_last
+
+
+def luInclude(filename, CallOnFail=None):
+    tstFile = LUtil.base_script_dir + "/" + filename
+    LUtil.setFilename(filename)
+    if CallOnFail != None:
+        oldCallOnFail = LUtil.getCallOnFail()
+        LUtil.setCallOnFail(CallOnFail)
+    if filename.endswith(".py"):
+        LUtil.log("luInclude: execfile " + tstFile)
+        with open(tstFile) as infile:
+            exec(infile.read())
+    else:
+        LUtil.log("luInclude: execTestFile " + tstFile)
+        LUtil.execTestFile(tstFile)
+    if CallOnFail != None:
+        LUtil.setCallOnFail(oldCallOnFail)
+
+
+def luFinish():
+    global LUtil
+    ret = LUtil.closeFiles()
+    # done
+    LUtil = None
+    return ret
+
+
+def luNumFail():
+    return LUtil.l_fail
+
+
+def luNumPass():
+    return LUtil.l_pass
+
+
+def luResult(target, success, str, logstr=None):
+    return LUtil.result(target, success, str, logstr)
+
+
+def luShowResults(prFunction):
+    printed = 0
+    sf = open(LUtil.fsum_name, "r")
+    for line in sf:
+        printed += 1
+        prFunction(line.rstrip())
+    sf.close()
+
+
+def luShowFail():
+    printed = 0
+    sf = open(LUtil.fsum_name, "r")
+    for line in sf:
+        if line[-2] != "0":
+            printed += 1
+            logger.error(line.rstrip())
+    sf.close()
+    if printed > 0:
+        logger.error("See %s for details of errors" % LUtil.fout_name)
+
+
+#
+# Sets default wait type for luCommand(op="wait) (may be overridden by
+# specifying luCommand(op="wait-strict") or luCommand(op="wait-nostrict")).
+#
+# "nostrict" is the historical default behavior, which is to ignore
+# failures to match the specified regexp in the specified time.
+#
+# "strict" means that failure to match the specified regexp in the
+# specified time yields an explicit, logged failure result
+#
+def luSetWaitType(waittype):
+    if waittype == "strict":
+        LUtil.l_wait_strict = 1
+    else:
+        if waittype == "nostrict":
+            LUtil.l_wait_strict = 0
+        else:
+            raise ValueError('waittype must be one of "strict" or "nostrict"')
+
+
+# for testing
+if __name__ == "__main__":
+    print(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/lib")
+    luStart()
+    for arg in sys.argv[1:]:
+        luInclude(arg)
+    luFinish()
+    sys.exit(0)