blob: 930ce2db2a42acb106ae1c25f2cc8767e1a906bf [file] [log] [blame]
#!/usr/bin/env python
'''
Created on 23-Oct-2012
Modified 2015 by ON.Lab
Please refer questions to either the onos test mailing list at <onos-test@onosproject.org>,
the System Testing Plans and Results wiki page at <https://wiki.onosproject.org/x/voMg>,
or the System Testing Guide page at <https://wiki.onosproject.org/x/WYQg>
TestON 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.
TestON 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 TestON. If not, see <http://www.gnu.org/licenses/>.
Utilities will take care about the basic functions like :
* Extended assertion,
* parse_args for key-value pair handling
* Parsing the params or topology file.
'''
import re
from configobj import ConfigObj
from core import ast as ast
import smtplib
import email
import os
import email.mime.application
import time
import random
class Utilities:
'''
Utilities will take care about the basic functions like :
* Extended assertion,
* parse_args for key-value pair handling
* Parsing the params or topology file.
'''
def __init__(self):
self.wrapped = sys.modules[__name__]
def __getattr__(self, name):
'''
This will invoke, if the attribute wasn't found the usual ways.
Here it will look for assert_attribute and will execute when AttributeError occurs.
It will return the result of the assert_attribute.
'''
try:
return getattr(self.wrapped, name)
except AttributeError:
def assertHandling(**kwargs):
nameVar = re.match("^assert",name,flags=0)
matchVar = re.match("assert(_not_|_)(equals|matches|greater|lesser)",name,flags=0)
notVar = 0
operators = ""
try :
if matchVar.group(1) == "_not_" and matchVar.group(2) :
notVar = 1
operators = matchVar.group(2)
elif matchVar.group(1) == "_" and matchVar.group(2):
operators = matchVar.group(2)
except AttributeError:
if matchVar==None and nameVar:
operators ='equals'
result = self._assert(NOT=notVar,operator=operators,**kwargs)
if result == main.TRUE:
main.log.info("Assertion Passed")
main.STEPRESULT = main.TRUE
elif result == main.FALSE:
main.log.warn("Assertion Failed")
main.STEPRESULT = main.FALSE
else:
main.log.error("There is an Error in Assertion")
main.STEPRESULT = main.ERROR
return result
return assertHandling
def _assert (self,**assertParam):
'''
It will take the arguments :
expect:'Expected output'
actual:'Actual output'
onpass:'Action or string to be triggered or displayed respectively when the assert passed'
onfail:'Action or string to be triggered or displayed respectively when the assert failed'
not:'optional argument to specify the negation of the each assertion type'
operator:'assertion type will be defined by using operator. Like equal , greater, lesser, matches.'
It will return the assertion result.
'''
arguments = self.parse_args(["EXPECT","ACTUAL","ONPASS","ONFAIL","NOT","OPERATOR"],**assertParam)
result = 0
valuetype = ''
operation = "not "+ str(arguments["OPERATOR"]) if arguments['NOT'] and arguments['NOT'] == 1 else arguments["OPERATOR"]
operators = {'equals':{'STR':'==','NUM':'=='}, 'matches' : '=~', 'greater':'>' ,'lesser':'<'}
expectMatch = re.match('^\s*[+-]?0(e0)?\s*$', str(arguments["EXPECT"]), re.I+re.M)
if not ((not expectMatch) and (arguments["EXPECT"]==0)):
valuetype = 'NUM'
else :
if arguments["OPERATOR"] == 'greater' or arguments["OPERATOR"] == 'lesser':
main.log.error("Numeric comparison on strings is not possibele")
return main.ERROR
valuetype = 'STR'
arguments["ACTUAL"] = str(arguments["ACTUAL"])
if arguments["OPERATOR"] != 'matches':
arguments["EXPECT"] = str(arguments["EXPECT"])
try :
opcode = operators[str(arguments["OPERATOR"])][valuetype] if arguments["OPERATOR"] == 'equals' else operators[str(arguments["OPERATOR"])]
except KeyError as e:
print "Key Error in assertion"
print e
return main.FALSE
if opcode == '=~':
try:
assert re.search(str(arguments["EXPECT"]),str(arguments["ACTUAL"]))
result = main.TRUE
except AssertionError:
try :
assert re.match(str(arguments["EXPECT"]),str(arguments["ACTUAL"]))
result = main.TRUE
except AssertionError:
main.log.error("Assertion Failed")
result = main.FALSE
else :
try:
if str(opcode)=="==":
main.log.info("Verifying the Expected is equal to the actual or not using assert_equal")
if (arguments["EXPECT"] == arguments["ACTUAL"]):
result = main.TRUE
else :
result = main.FALSE
elif str(opcode) == ">":
main.log.info("Verifying the Expected is Greater than the actual or not using assert_greater")
if (ast.literal_eval(arguments["EXPECT"]) > ast.literal_eval(arguments["ACTUAL"])) :
result = main.TRUE
else :
result = main.FALSE
elif str(opcode) == "<":
main.log.info("Verifying the Expected is Lesser than the actual or not using assert_lesser")
if (ast.literal_eval(arguments["EXPECT"]) < ast.literal_eval(arguments["ACTUAL"])):
result = main.TRUE
else :
result = main.FALSE
except AssertionError:
main.log.error("Assertion Failed")
result = main.FALSE
result = result if result else 0
result = not result if arguments["NOT"] and arguments["NOT"] == 1 else result
resultString = ""
if result :
resultString = str(resultString) + "PASS"
main.log.info(arguments["ONPASS"])
else :
resultString = str(resultString) + "FAIL"
if not isinstance(arguments["ONFAIL"],str):
eval(str(arguments["ONFAIL"]))
else :
main.log.error(arguments["ONFAIL"])
main.log.report(arguments["ONFAIL"])
main.onFailMsg = arguments[ 'ONFAIL' ]
msg = arguments["ON" + str(resultString)]
if not isinstance(msg,str):
try:
eval(str(msg))
except SyntaxError as e:
print "function definition is not right"
print e
main.last_result = result
if main.stepResults[2]:
main.stepResults[2][-1] = result
try:
main.stepResults[3][-1] = arguments[ 'ONFAIL' ]
except AttributeError:
pass
else:
main.log.warn( "Assertion called before a test step" )
return result
def parse_args(self,args, **kwargs):
'''
It will accept the (key,value) pair and will return the (key,value) pairs with keys in uppercase.
'''
newArgs = {}
for key,value in kwargs.iteritems():
if isinstance(args,list) and str.upper(key) in args:
for each in args:
if each==str.upper(key):
newArgs [str(each)] = value
elif each != str.upper(key) and (newArgs.has_key(str(each)) == False ):
newArgs[str(each)] = None
return newArgs
def send_mail(self):
# Create a text/plain message
msg = email.mime.Multipart.MIMEMultipart()
try :
if main.test_target:
sub = "Result summary of \"" + main.TEST + "\" run on component \"" +\
main.test_target + "\" Version \"" +\
vars( main )[main.test_target].get_version() + "\": " +\
str( main.TOTAL_TC_SUCCESS ) + "% Passed"
else :
sub = "Result summary of \"" + main.TEST + "\": " +\
str( main.TOTAL_TC_SUCCESS ) + "% Passed"
except ( KeyError, AttributeError ):
sub = "Result summary of \"" + main.TEST + "\": " +\
str( main.TOTAL_TC_SUCCESS ) + "% Passed"
msg['Subject'] = sub
msg['From'] = main.sender
msg['To'] = main.mail
# The main body is just another attachment
body = email.mime.Text.MIMEText( main.logHeader + "\n" +
main.testResult)
msg.attach( body )
# Attachments
for filename in os.listdir( main.logdir ):
filepath = main.logdir + "/" + filename
fp = open( filepath, 'rb' )
att = email.mime.application.MIMEApplication( fp.read(),
_subtype="" )
fp.close()
att.add_header( 'Content-Disposition',
'attachment',
filename=filename )
msg.attach( att )
try:
smtp = smtplib.SMTP( main.smtp )
smtp.starttls()
smtp.login( main.sender, main.senderPwd )
smtp.sendmail( msg['From'], [msg['To']], msg.as_string() )
smtp.quit()
except Exception:
main.log.exception( "Error sending email" )
return main.TRUE
def send_warning_email( self, subject=None ):
try:
if not subject:
subject = main.TEST + " PAUSED!"
# Create a text/plain message
msg = email.mime.Multipart.MIMEMultipart()
msg['Subject'] = subject
msg['From'] = main.sender
msg['To'] = main.mail
smtp = smtplib.SMTP( main.smtp )
smtp.starttls()
smtp.login( main.sender, main.senderPwd )
smtp.sendmail( msg['From'], [msg['To']], msg.as_string() )
smtp.quit()
except Exception:
main.log.exception( "" )
return main.FALSE
return main.TRUE
def parse(self,fileName):
'''
This will parse the params or topo or cfg file and return content in the file as Dictionary
'''
self.fileName = fileName
matchFileName = re.match(r'(.*)\.(cfg|params|topo)',self.fileName,re.M|re.I)
if matchFileName:
try :
parsedInfo = ConfigObj(self.fileName)
return parsedInfo
except StandardError:
print "There is no such file to parse "+fileName
else:
return 0
def retry( self, f, retValue, args=(), kwargs={},
sleep=1, attempts=2, randomTime=False ):
"""
Given a function and bad return values, retry will retry a function
until successful or give up after a certain number of attempts.
Arguments:
f - a callable object
retValue - Return value(s) of f to retry on. This can be a list or an
object.
args - A tuple containing the arguments of f.
kwargs - A dictionary containing the keyword arguments of f.
sleep - Time in seconds to sleep between retries. If random is True,
this is the max time to wait. Defaults to 1 second.
attempts - Max number of attempts before returning. If set to 1,
f will only be called once. Defaults to 2 trys.
random - Boolean indicating if the wait time is random between 0
and sleep or exactly sleep seconds. Defaults to False.
"""
# TODO: be able to pass in a conditional statement(s). For example:
# retCondition = "< 7"
# Then we do something like 'if eval( "ret " + retCondition ):break'
try:
assert attempts > 0, "attempts must be more than 1"
assert sleep >= 0, "sleep must be >= 0"
if not isinstance( retValue, list ):
retValue = [ retValue ]
for i in range( 0, attempts ):
ret = f( *args, **kwargs )
if ret not in retValue:
# NOTE that False in [ 0 ] == True
break
if randomTime:
sleeptime = random.randint( 0, sleep )
else:
sleeptime = sleep
time.sleep( sleeptime )
main.log.debug( str( i ) + " Attempts needed to pass this test" )
return ret
except AssertionError:
main.log.exception( "Invalid arguements for retry: " )
main.cleanup()
main.exit()
except Exception:
main.log.exception( "Uncaught exception in retry: " )
main.cleanup()
main.exit()
if __name__ != "__main__":
import sys
sys.modules[__name__] = Utilities()