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 from topolog import logger
  25 from mininet.net import Mininet
  26
  27
  28 # L utility functions
  29 #
  30 # These functions are inteneted to provide support for CI testing within MiniNet
  31 # environments.
  32
  33 class lUtil:
  34     #to be made configurable in the future
  35     base_script_dir = '.'
  36     base_log_dir = '.'
  37     fout_name = 'output.log'
  38     fsum_name = 'summary.txt'
  39     l_level = 9
  40     CallOnFail = False
  41
  42     l_total = 0
  43     l_pass = 0
  44     l_fail = 0
  45     l_filename = ''
  46     l_last = None
  47     l_line = 0
  48
  49     fout = ''
  50     fsum = ''
  51     net  = ''
  52
  53     def log(self, str):
  54         if self.l_level > 0:
  55             if self.fout == '':
  56                 self.fout = open(self.fout_name, 'w', 0)
  57             self.fout.write(str+'\n')
  58         if self.l_level > 5:
  59             print(str)
  60
  61     def result(self, target, success, str):
  62         if success:
  63             p = 1
  64             f = 0
  65             self.l_pass += 1
  66         else:
  67             f = 1
  68             p = 0
  69             self.l_fail += 1
  70         res = "%-4d %-6s %-56s %-4d %d" % (self.l_total, target, str, p, f)
  71         self.log ('R:'+res)
  72         if self.fsum == '':
  73             self.fsum = open(self.fsum_name, 'w', 0)
  74             self.fsum.write('\
  75 ******************************************************************************\n')
  76             self.fsum.write('\
  77 Test Target Summary                                                  Pass Fail\n')
  78             self.fsum.write('\
  79 ******************************************************************************\n')
  80         self.fsum.write(res+'\n')
  81         if f == 1 and self.CallOnFail != False:
  82             self.CallOnFail()
  83
  84     def closeFiles(self):
  85         ret = '\
  86 ******************************************************************************\n\
  87 Total %-4d                                                           %-4d %d\n\
  88 ******************************************************************************'\
  89 % (self.l_total, self.l_pass, self.l_fail)
  90         if self.fsum != '':
  91             self.fsum.write(ret + '\n')
  92             self.fsum.close()
  93             self.fsum = ''
  94         if self.fout != '':
  95             if os.path.isfile(self.fsum_name):
  96                 r = open(self.fsum_name, 'r')
  97                 self.fout.write(r.read())
  98                 r.close()
  99             self.fout.close()
 100             self.fout = ''
 101         return ret
 102
 103     def setFilename(self, name):
 104         self.log('FILE: ' + name)
 105         self.l_filename = name
 106         self.line = 0
 107
 108     def getCallOnFail(self):
 109         return self.CallOnFail
 110
 111     def setCallOnFail(self, CallOnFail):
 112         self.CallOnFail = CallOnFail
 113
 114     def strToArray(self, string):
 115         a = []
 116         c = 0
 117         end = ''
 118         words = string.split()
 119         if len(words) < 1 or words[0].startswith('#'):
 120             return a
 121         words = string.split()
 122         for word in words:
 123             if len(end) == 0:
 124                 a.append(word)
 125             else:
 126                 a[c] += str(' '+word)
 127             if end == '\\':
 128                 end = ''
 129             if not word.endswith('\\'):
 130                 if end != '"':
 131                     if word.startswith('"'):
 132                         end = '"'
 133                     else:
 134                         c += 1
 135                 else:
 136                     if word.endswith('"'):
 137                         end = ''
 138                         c += 1
 139                     else:
 140                         c += 1
 141             else:
 142                 end = '\\'
 143     #        if len(end) == 0:
 144     #            print('%d:%s:' % (c, a[c-1]))
 145
 146         return a
 147
 148     def execTestFile(self, tstFile):
 149         if os.path.isfile(tstFile):
 150             f = open(tstFile)
 151             for line in f:
 152                 if len(line) > 1:
 153                     a = self.strToArray(line)
 154                     if len(a) >= 6:
 155                         luCommand(a[1], a[2], a[3], a[4], a[5])
 156                     else:
 157                         self.l_line += 1
 158                         self.log('%s:%s %s' % (self.l_filename, self.l_line , line))
 159                         if len(a) >= 2:
 160                             if a[0] == 'sleep':
 161                                 time.sleep(int(a[1]))
 162                             elif a[0] == 'include':
 163                                 self.execTestFile(a[1])
 164             f.close()
 165         else:
 166             self.log('unable to read: ' + tstFile)
 167             sys.exit(1)
 168
 169     def command(self, target, command, regexp, op, result):
 170         global net
 171         if op != 'wait':
 172             self.l_line  += 1
 173             self.l_total += 1
 174         self.log('%s:%s COMMAND:%s:%s:%s:%s:%s:' % \
 175                  (self.l_filename, self.l_line, target, command, regexp, op, result))
 176         if self.net == '':
 177             return False
 178         #self.log("Running %s %s" % (target, command))
 179         out = self.net[target].cmd(command).rstrip()
 180         if len(out) == 0:
 181             report = "<no output>"
 182         else:
 183             report = out
 184         self.log('COMMAND OUTPUT:%s:' % report)
 185         out = " ".join(out.splitlines())
 186         search = re.search(regexp, out)
 187         self.l_last = search
 188         if search == None:
 189             if op == 'fail':
 190                 success = True
 191             else:
 192                 success = False
 193             ret = success
 194         else:
 195             ret = search.group()
 196             self.log('found:%s:' % ret)
 197             if op != 'fail':
 198                 success = True
 199             else:
 200                 success = False
 201         if op == 'pass' or op == 'fail':
 202             self.result(target, success, result)
 203         return ret
 204
 205     def wait(self, target, command, regexp, op, result, wait):
 206         self.log('%s:%s WAIT:%s:%s:%s:%s:%s:%s:' % \
 207                  (self.l_filename, self.l_line, target, command, regexp, op, result,wait))
 208         llevel = LUtil.l_level
 209         found = False
 210         n = 0
 211         startt = time.time()
 212         delta = time.time() - startt
 213         while delta < wait and found is False:
 214             found = self.command(target, command, regexp, op, result)
 215             n+=1
 216             LUtil.l_level = 0
 217             delta = time.time() - startt
 218             if delta < wait and found is False:
 219                 time.sleep (0.5)
 220         LUtil.l_level = llevel
 221         self.log('Done after %d loops, time=%s, Found=%s' % (n, delta, found))
 222         found = self.command(target, command, regexp, 'pass', '%s +%4.2f secs' % (result, delta))
 223         return found
 224
 225 #initialized by luStart
 226 LUtil=None
 227
 228 #entry calls
 229 def luStart(baseScriptDir='.', baseLogDir='.', net='',
 230             fout='output.log', fsum='summary.txt', level=9):
 231     global LUtil
 232     #init class
 233     LUtil=lUtil()
 234     LUtil.base_script_dir = baseScriptDir
 235     LUtil.base_log_dir = baseLogDir
 236     LUtil.net = net
 237     if fout != '':
 238         LUtil.fout_name = baseLogDir + '/' + fout
 239     if fsum != None:
 240         LUtil.fsum_name = baseLogDir + '/' + fsum
 241     LUtil.l_level = level
 242
 243 def luCommand(target, command, regexp='.', op='none', result='', time=10):
 244     if op != 'wait':
 245         return LUtil.command(target, command, regexp, op, result)
 246     else:
 247         return LUtil.wait(target, command, regexp, op, result, time)
 248
 249 def luLast():
 250     if LUtil.l_last != None:
 251         LUtil.log('luLast:%s:' %  LUtil.l_last.group())
 252     return LUtil.l_last
 253
 254 def luInclude(filename, CallOnFail=None):
 255     tstFile = LUtil.base_script_dir + '/' + filename
 256     LUtil.setFilename(filename)
 257     if CallOnFail != None:
 258         oldCallOnFail = LUtil.getCallOnFail()
 259         LUtil.setCallOnFail(CallOnFail)
 260     if filename.endswith('.py'):
 261         LUtil.log("luInclude: execfile "+tstFile)
 262         execfile(tstFile)
 263     else:
 264         LUtil.log("luInclude: execTestFile "+tstFile)
 265         LUtil.execTestFile(tstFile)
 266     if CallOnFail != None:
 267         LUtil.setCallOnFail(oldCallOnFail)
 268
 269 def luFinish():
 270     global LUtil
 271     ret = LUtil.closeFiles()
 272     #done
 273     LUtil = None
 274     return ret;
 275
 276 def luNumFail():
 277     return LUtil.l_fail
 278
 279 def luNumPass():
 280     return LUtil.l_pass
 281
 282 def luShowFail():
 283     printed = 0
 284     sf = open(LUtil.fsum_name, 'r')
 285     for line in sf:
 286         if line[-2] != "0":
 287             printed+=1
 288             logger.error(line.rstrip())
 289     sf.close()
 290     if printed > 0:
 291          logger.error("See %s for details of errors" % LUtil.fout_name)
 292
 293 #for testing
 294 if __name__ == '__main__':
 295     print(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + '/lib')
 296     luStart()
 297     for arg in sys.argv[1:]:
 298         luInclude(arg)
 299     luFinish()
 300     sys.exit(0)
 301