tests/topotests/lib/lutil.py

   1 #!/usr/bin/env python
   2
   3 # Copyright 2017, LabN Consulting, L.L.C.
   4 #
   5 # This program is free software; you can redistribute it and/or
   6 # modify it under the terms of the GNU General Public License
   7 # as published by the Free Software Foundation; either version 2
   8 # of the License, or (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along
  16 # with this program; see the file COPYING; if not, write to the Free Software
  17 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
  18
  19 import os
  20 import re
  21 import sys
  22 import time
  23 import datetime
  24 import json
  25 import math
  26 import time
  27 from lib.topolog import logger
  28 from mininet.net import Mininet
  29
  30
  31 # L utility functions
  32 #
  33 # These functions are inteneted to provide support for CI testing within MiniNet
  34 # environments.
  35
  36
  37 class lUtil:
  38     # to be made configurable in the future
  39     base_script_dir = "."
  40     base_log_dir = "."
  41     fout_name = "output.log"
  42     fsum_name = "summary.txt"
  43     l_level = 6
  44     CallOnFail = False
  45
  46     l_total = 0
  47     l_pass = 0
  48     l_fail = 0
  49     l_filename = ""
  50     l_last = None
  51     l_line = 0
  52     l_dotall_experiment = False
  53     l_last_nl = None
  54
  55     fout = ""
  56     fsum = ""
  57     net = ""
  58
  59     def log(self, str, level=6):
  60         if self.l_level > 0:
  61             if self.fout == "":
  62                 self.fout = open(self.fout_name, "w")
  63             self.fout.write(str + "\n")
  64         if level <= self.l_level:
  65             print(str)
  66
  67     def summary(self, str):
  68         if self.fsum == "":
  69             self.fsum = open(self.fsum_name, "w")
  70             self.fsum.write(
  71                 "\
  72 ******************************************************************************\n"
  73             )
  74             self.fsum.write(
  75                 "\
  76 Test Target Summary                                                  Pass Fail\n"
  77             )
  78             self.fsum.write(
  79                 "\
  80 ******************************************************************************\n"
  81             )
  82         self.fsum.write(str + "\n")
  83
  84     def result(self, target, success, str, logstr=None):
  85         if success:
  86             p = 1
  87             f = 0
  88             self.l_pass += 1
  89             sstr = "PASS"
  90         else:
  91             f = 1
  92             p = 0
  93             self.l_fail += 1
  94             sstr = "FAIL"
  95         self.l_total += 1
  96         if logstr != None:
  97             self.log("R:%d %s: %s" % (self.l_total, sstr, logstr))
  98         res = "%-4d %-6s %-56s %-4d %d" % (self.l_total, target, str, p, f)
  99         self.log("R:" + res)
 100         self.summary(res)
 101         if f == 1 and self.CallOnFail != False:
 102             self.CallOnFail()
 103
 104     def closeFiles(self):
 105         ret = (
 106             "\
 107 ******************************************************************************\n\
 108 Total %-4d                                                           %-4d %d\n\
 109 ******************************************************************************"
 110             % (self.l_total, self.l_pass, self.l_fail)
 111         )
 112         if self.fsum != "":
 113             self.fsum.write(ret + "\n")
 114             self.fsum.close()
 115             self.fsum = ""
 116         if self.fout != "":
 117             if os.path.isfile(self.fsum_name):
 118                 r = open(self.fsum_name, "r")
 119                 self.fout.write(r.read())
 120                 r.close()
 121             self.fout.close()
 122             self.fout = ""
 123         return ret
 124
 125     def setFilename(self, name):
 126         str = "FILE: " + name
 127         self.log(str)
 128         self.summary(str)
 129         self.l_filename = name
 130         self.line = 0
 131
 132     def getCallOnFail(self):
 133         return self.CallOnFail
 134
 135     def setCallOnFail(self, CallOnFail):
 136         self.CallOnFail = CallOnFail
 137
 138     def strToArray(self, string):
 139         a = []
 140         c = 0
 141         end = ""
 142         words = string.split()
 143         if len(words) < 1 or words[0].startswith("#"):
 144             return a
 145         words = string.split()
 146         for word in words:
 147             if len(end) == 0:
 148                 a.append(word)
 149             else:
 150                 a[c] += str(" " + word)
 151             if end == "\\":
 152                 end = ""
 153             if not word.endswith("\\"):
 154                 if end != '"':
 155                     if word.startswith('"'):
 156                         end = '"'
 157                     else:
 158                         c += 1
 159                 else:
 160                     if word.endswith('"'):
 161                         end = ""
 162                         c += 1
 163                     else:
 164                         c += 1
 165             else:
 166                 end = "\\"
 167         #        if len(end) == 0:
 168         #            print('%d:%s:' % (c, a[c-1]))
 169
 170         return a
 171
 172     def execTestFile(self, tstFile):
 173         if os.path.isfile(tstFile):
 174             f = open(tstFile)
 175             for line in f:
 176                 if len(line) > 1:
 177                     a = self.strToArray(line)
 178                     if len(a) >= 6:
 179                         luCommand(a[1], a[2], a[3], a[4], a[5])
 180                     else:
 181                         self.l_line += 1
 182                         self.log("%s:%s %s" % (self.l_filename, self.l_line, line))
 183                         if len(a) >= 2:
 184                             if a[0] == "sleep":
 185                                 time.sleep(int(a[1]))
 186                             elif a[0] == "include":
 187                                 self.execTestFile(a[1])
 188             f.close()
 189         else:
 190             self.log("unable to read: " + tstFile)
 191             sys.exit(1)
 192
 193     def command(self, target, command, regexp, op, result, returnJson):
 194         global net
 195         if op != "wait":
 196             self.l_line += 1
 197         self.log(
 198             "%s (#%d) %s:%s COMMAND:%s:%s:%s:%s:%s:"
 199             % (
 200                 time.asctime(),
 201                 self.l_total + 1,
 202                 self.l_filename,
 203                 self.l_line,
 204                 target,
 205                 command,
 206                 regexp,
 207                 op,
 208                 result,
 209             )
 210         )
 211         if self.net == "":
 212             return False
 213         # self.log("Running %s %s" % (target, command))
 214         js = None
 215         out = self.net[target].cmd(command).rstrip()
 216         if len(out) == 0:
 217             report = "<no output>"
 218         else:
 219             report = out
 220             if returnJson == True:
 221                 try:
 222                     js = json.loads(out)
 223                 except:
 224                     js = None
 225                     self.log(
 226                         "WARNING: JSON load failed -- confirm command output is in JSON format."
 227                     )
 228         self.log("COMMAND OUTPUT:%s:" % report)
 229
 230         # Experiment: can we achieve the same match behavior via DOTALL
 231         # without converting newlines to spaces?
 232         out_nl = out
 233         search_nl = re.search(regexp, out_nl, re.DOTALL)
 234         self.l_last_nl = search_nl
 235         # Set up for comparison
 236         if search_nl != None:
 237             group_nl = search_nl.group()
 238             group_nl_converted = " ".join(group_nl.splitlines())
 239         else:
 240             group_nl_converted = None
 241
 242         out = " ".join(out.splitlines())
 243         search = re.search(regexp, out)
 244         self.l_last = search
 245         if search == None:
 246             if op == "fail":
 247                 success = True
 248             else:
 249                 success = False
 250             ret = success
 251         else:
 252             ret = search.group()
 253             if op != "fail":
 254                 success = True
 255                 level = 7
 256             else:
 257                 success = False
 258                 level = 5
 259             self.log("found:%s:" % ret, level)
 260             # Experiment: compare matched strings obtained each way
 261             if self.l_dotall_experiment and (group_nl_converted != ret):
 262                 self.log(
 263                     "DOTALL experiment: strings differ dotall=[%s] orig=[%s]"
 264                     % (group_nl_converted, ret),
 265                     9,
 266                 )
 267         if op == "pass" or op == "fail":
 268             self.result(target, success, result)
 269         if js != None:
 270             return js
 271         return ret
 272
 273     def wait(
 274         self, target, command, regexp, op, result, wait, returnJson, wait_time=0.5
 275     ):
 276         self.log(
 277             "%s:%s WAIT:%s:%s:%s:%s:%s:%s:%s:"
 278             % (
 279                 self.l_filename,
 280                 self.l_line,
 281                 target,
 282                 command,
 283                 regexp,
 284                 op,
 285                 result,
 286                 wait,
 287                 wait_time,
 288             )
 289         )
 290         found = False
 291         n = 0
 292         startt = time.time()
 293
 294         # Calculate the amount of `sleep`s we are going to peform.
 295         wait_count = int(math.ceil(wait / wait_time)) + 1
 296
 297         while wait_count > 0:
 298             n += 1
 299             found = self.command(target, command, regexp, op, result, returnJson)
 300             if found is not False:
 301                 break
 302
 303             wait_count -= 1
 304             if wait_count > 0:
 305                 time.sleep(wait_time)
 306
 307         delta = time.time() - startt
 308         self.log("Done after %d loops, time=%s, Found=%s" % (n, delta, found))
 309         found = self.command(
 310             target,
 311             command,
 312             regexp,
 313             "pass",
 314             "%s +%4.2f secs" % (result, delta),
 315             returnJson,
 316         )
 317         return found
 318
 319
 320 # initialized by luStart
 321 LUtil = None
 322
 323 # entry calls
 324 def luStart(
 325     baseScriptDir=".",
 326     baseLogDir=".",
 327     net="",
 328     fout="output.log",
 329     fsum="summary.txt",
 330     level=None,
 331 ):
 332     global LUtil
 333     # init class
 334     LUtil = lUtil()
 335     LUtil.base_script_dir = baseScriptDir
 336     LUtil.base_log_dir = baseLogDir
 337     LUtil.net = net
 338     if fout != "":
 339         LUtil.fout_name = baseLogDir + "/" + fout
 340     if fsum != None:
 341         LUtil.fsum_name = baseLogDir + "/" + fsum
 342     if level != None:
 343         LUtil.l_level = level
 344     LUtil.l_dotall_experiment = False
 345     LUtil.l_dotall_experiment = True
 346
 347
 348 def luCommand(
 349     target,
 350     command,
 351     regexp=".",
 352     op="none",
 353     result="",
 354     time=10,
 355     returnJson=False,
 356     wait_time=0.5,
 357 ):
 358     if op != "wait":
 359         return LUtil.command(target, command, regexp, op, result, returnJson)
 360     else:
 361         return LUtil.wait(
 362             target, command, regexp, op, result, time, returnJson, wait_time
 363         )
 364
 365
 366 def luLast(usenl=False):
 367     if usenl:
 368         if LUtil.l_last_nl != None:
 369             LUtil.log("luLast:%s:" % LUtil.l_last_nl.group(), 7)
 370         return LUtil.l_last_nl
 371     else:
 372         if LUtil.l_last != None:
 373             LUtil.log("luLast:%s:" % LUtil.l_last.group(), 7)
 374         return LUtil.l_last
 375
 376
 377 def luInclude(filename, CallOnFail=None):
 378     tstFile = LUtil.base_script_dir + "/" + filename
 379     LUtil.setFilename(filename)
 380     if CallOnFail != None:
 381         oldCallOnFail = LUtil.getCallOnFail()
 382         LUtil.setCallOnFail(CallOnFail)
 383     if filename.endswith(".py"):
 384         LUtil.log("luInclude: execfile " + tstFile)
 385         with open(tstFile) as infile:
 386             exec(infile.read())
 387     else:
 388         LUtil.log("luInclude: execTestFile " + tstFile)
 389         LUtil.execTestFile(tstFile)
 390     if CallOnFail != None:
 391         LUtil.setCallOnFail(oldCallOnFail)
 392
 393
 394 def luFinish():
 395     global LUtil
 396     ret = LUtil.closeFiles()
 397     # done
 398     LUtil = None
 399     return ret
 400
 401
 402 def luNumFail():
 403     return LUtil.l_fail
 404
 405
 406 def luNumPass():
 407     return LUtil.l_pass
 408
 409
 410 def luResult(target, success, str, logstr=None):
 411     return LUtil.result(target, success, str, logstr)
 412
 413
 414 def luShowResults(prFunction):
 415     printed = 0
 416     sf = open(LUtil.fsum_name, "r")
 417     for line in sf:
 418         printed += 1
 419         prFunction(line.rstrip())
 420     sf.close()
 421
 422
 423 def luShowFail():
 424     printed = 0
 425     sf = open(LUtil.fsum_name, "r")
 426     for line in sf:
 427         if line[-2] != "0":
 428             printed += 1
 429             logger.error(line.rstrip())
 430     sf.close()
 431     if printed > 0:
 432         logger.error("See %s for details of errors" % LUtil.fout_name)
 433
 434
 435 # for testing
 436 if __name__ == "__main__":
 437     print(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/lib")
 438     luStart()
 439     for arg in sys.argv[1:]:
 440         luInclude(arg)
 441     luFinish()
 442     sys.exit(0)