mirror of
https://git.proxmox.com/git/mirror_frr
synced 2025-05-01 07:26:05 +00:00
c63906e8cf
Prior behavior of luCommand was to convert newlines to spaces in DUT output before pattern matching. New method operating in parallel uses re.DOTALL to mimic same behavior and allow preserving original DUT output. The original output is needed for some scripts that parse line-by-line. There is also some test code to compare match results using the old way and new way and log a message if they are different. After some short time we can develop confidence that using this new method will not break any existing tests. Signed-off-by: G. Paul Ziemba <paulz@labn.net>
344 lines
10 KiB
Python
Executable File
344 lines
10 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
# Copyright 2017, LabN Consulting, L.L.C.
|
|
#
|
|
# This program is free software; you can redistribute it and/or
|
|
# modify it under the terms of the GNU General Public License
|
|
# as published by the Free Software Foundation; either version 2
|
|
# of the License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License along
|
|
# with this program; see the file COPYING; if not, write to the Free Software
|
|
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import datetime
|
|
import json
|
|
from topolog import logger
|
|
from mininet.net import Mininet
|
|
|
|
|
|
# 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 = 9
|
|
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
|
|
|
|
fout = ''
|
|
fsum = ''
|
|
net = ''
|
|
|
|
def log(self, str):
|
|
if self.l_level > 0:
|
|
if self.fout == '':
|
|
self.fout = open(self.fout_name, 'w', 0)
|
|
self.fout.write(str+'\n')
|
|
if self.l_level > 5:
|
|
print(str)
|
|
|
|
def summary(self, str):
|
|
if self.fsum == '':
|
|
self.fsum = open(self.fsum_name, 'w', 0)
|
|
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):
|
|
if success:
|
|
p = 1
|
|
f = 0
|
|
self.l_pass += 1
|
|
else:
|
|
f = 1
|
|
p = 0
|
|
self.l_fail += 1
|
|
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):
|
|
global net
|
|
if op != 'wait':
|
|
self.l_line += 1
|
|
if op == 'pass' or op == 'fail':
|
|
self.l_total += 1
|
|
self.log('%s:%s COMMAND:%s:%s:%s:%s:%s:' % \
|
|
(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)
|
|
|
|
# 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())
|
|
|
|
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()
|
|
self.log('found:%s:' % ret)
|
|
if op != 'fail':
|
|
success = True
|
|
else:
|
|
success = False
|
|
# 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))
|
|
if 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):
|
|
self.log('%s:%s WAIT:%s:%s:%s:%s:%s:%s:' % \
|
|
(self.l_filename, self.l_line, target, command, regexp, op, result,wait))
|
|
llevel = LUtil.l_level
|
|
found = False
|
|
n = 0
|
|
startt = time.time()
|
|
delta = time.time() - startt
|
|
while delta < wait and found is False:
|
|
found = self.command(target, command, regexp, op, result, returnJson)
|
|
n+=1
|
|
LUtil.l_level = 0
|
|
delta = time.time() - startt
|
|
if delta < wait and found is False:
|
|
time.sleep (0.5)
|
|
LUtil.l_level = llevel
|
|
self.log('Done after %d loops, time=%s, Found=%s' % (n, delta, found))
|
|
found = self.command(target, command, regexp, 'pass', '%s +%4.2f secs' % (result, delta), returnJson)
|
|
return found
|
|
|
|
#initialized by luStart
|
|
LUtil=None
|
|
|
|
#entry calls
|
|
def luStart(baseScriptDir='.', baseLogDir='.', net='',
|
|
fout='output.log', fsum='summary.txt', level=9):
|
|
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
|
|
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):
|
|
if op != 'wait':
|
|
return LUtil.command(target, command, regexp, op, result, returnJson)
|
|
else:
|
|
return LUtil.wait(target, command, regexp, op, result, time, returnJson)
|
|
|
|
def luLast(usenl=False):
|
|
if usenl:
|
|
if LUtil.l_last_nl != None:
|
|
LUtil.log('luLast:%s:' % LUtil.l_last_nl.group())
|
|
return LUtil.l_last_nl
|
|
else:
|
|
if LUtil.l_last != None:
|
|
LUtil.log('luLast:%s:' % LUtil.l_last.group())
|
|
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)
|
|
execfile(tstFile)
|
|
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):
|
|
return LUtil.result(target, success, str)
|
|
|
|
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)
|
|
|
|
#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)
|
|
|