resiliencytest: add possibility to run standalone tests.

parent f2724b81
...@@ -64,6 +64,7 @@ setup(name=name, ...@@ -64,6 +64,7 @@ setup(name=name,
'htpasswd = slapos.htpasswd:main', 'htpasswd = slapos.htpasswd:main',
'killpidfromfile = slapos.systool:killpidfromfile', 'killpidfromfile = slapos.systool:killpidfromfile',
'runResiliencyTest = slapos.resiliencytest:runResiliencyTest', 'runResiliencyTest = slapos.resiliencytest:runResiliencyTest',
'runStandaloneResiliencyTest = slapos.resiliencytest:runStandaloneResiliencyTest',
'lampconfigure = slapos.lamp:run [lampconfigure]', 'lampconfigure = slapos.lamp:run [lampconfigure]',
'onetimedownload = slapos.onetimedownload:main', 'onetimedownload = slapos.onetimedownload:main',
'onetimeupload = slapos.onetimeupload:main', 'onetimeupload = slapos.onetimeupload:main',
......
...@@ -81,27 +81,52 @@ def parseArguments(): ...@@ -81,27 +81,52 @@ def parseArguments():
return parser.parse_args() return parser.parse_args()
def setupLogging(log_path, name=__name__): def setupLogging(log_path=None, name=__name__):
logger_format = '%(asctime)s %(name)-13s: %(levelname)-8s %(message)s' logger_format = '%(asctime)s %(name)-13s: %(levelname)-8s %(message)s'
formatter = logging.Formatter(logger_format) formatter = logging.Formatter(logger_format)
logging.basicConfig(level=logging.INFO, logging.basicConfig(level=logging.INFO,
format=logger_format) format=logger_format)
logger = logging.getLogger(name) logger = logging.getLogger(name)
logger.addHandler(logging.NullHandler()) logger.addHandler(logging.NullHandler())
file_handler = logging.handlers.RotatingFileHandler( if log_path:
filename=log_path, file_handler = logging.handlers.RotatingFileHandler(
maxBytes=20000000, backupCount=4) filename=log_path,
file_handler.setFormatter(formatter) maxBytes=20000000, backupCount=4)
logger.addHandler(file_handler) file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
return logger return logger
def runTestSuite(test_suite_title, test_suite_arguments, log):
"""
Run a specified test suite, by dynamically loading the module and calling
its "runTestSuite" method.
"""
try:
# Generate the additional arguments that were given using the syntax
# additionalargument1=value1 additionalargument2=value2
parsed_arguments = dict(key.split('=') for key in test_suite_arguments)
test_suite_module = importFrom(test_suite_title)
success = test_suite_module.runTestSuite(**parsed_arguments)
except:
log('Impossible to run resiliency test:')
log(traceback.print_exc())
success = False
return success
class ScalabilityTest(object): class ScalabilityTest(object):
"""
Simple structure carrying test data.
"""
def __init__(self, data, test_result): def __init__(self, data, test_result):
self.__dict__ = {} self.__dict__ = {}
self.__dict__.update(data) self.__dict__.update(data)
self.test_result = test_result self.test_result = test_result
class ScalabilityLauncher(object): class ScalabilityLauncher(object):
"""
Core part of the code, responsible of speaking with the ERP5 testnode Master
and running tests.
"""
def __init__(self): def __init__(self):
self._argumentNamespace = parseArguments() self._argumentNamespace = parseArguments()
log_path = os.path.join(self._argumentNamespace.log_path, log_path = os.path.join(self._argumentNamespace.log_path,
...@@ -123,7 +148,7 @@ class ScalabilityLauncher(object): ...@@ -123,7 +148,7 @@ class ScalabilityLauncher(object):
or None if no test_case ready or None if no test_case ready
""" """
data = self.test_result.getNextTestCase() data = self.test_result.getNextTestCase()
if data == None : if data == None:
return None return None
decoded_data = Utils.deunicodeData(json.loads( decoded_data = Utils.deunicodeData(json.loads(
data data
...@@ -131,19 +156,6 @@ class ScalabilityLauncher(object): ...@@ -131,19 +156,6 @@ class ScalabilityLauncher(object):
next_test = ScalabilityTest(decoded_data, self.test_result) next_test = ScalabilityTest(decoded_data, self.test_result)
return next_test return next_test
def runCurrentTestSuite(self):
try:
# Generate the additional arguments that were given using the syntax
# additionalargument1=value1 additionalargument2=value2
additional_arguments = dict(key.split('=') for key in self._argumentNamespace.additional_arguments)
test_suite_module = importFrom(self._argumentNamespace.test_suite)
success = test_suite_module.runTestSuite(**additional_arguments)
except:
self.log('Impossible to run resiliency test:')
self.log(traceback.print_exc())
success = False
return success
def run(self): def run(self):
self.log('Resiliency Launcher started, with:') self.log('Resiliency Launcher started, with:')
self.log('Test suite master url: %s' % self._argumentNamespace.test_suite_master_url) self.log('Test suite master url: %s' % self._argumentNamespace.test_suite_master_url)
...@@ -172,7 +184,11 @@ class ScalabilityLauncher(object): ...@@ -172,7 +184,11 @@ class ScalabilityLauncher(object):
current_test.title current_test.title
) )
success = self.runCurrentTestSuite() success = runTestSuite(
self._argumentNamespace.test_suite,
self._argumentNamespace.additional_arguments,
self.log,
)
if success: if success:
error_count = 0 error_count = 0
...@@ -189,8 +205,28 @@ class ScalabilityLauncher(object): ...@@ -189,8 +205,28 @@ class ScalabilityLauncher(object):
return error_message_set, exit_status return error_message_set, exit_status
def runResiliencyTest(): def runResiliencyTest():
"""
Used for automated test suite run from "Scalability" Test Node infrastructure.
It means the instance running this code should have been deployed by a
"Scalability" Testnode.
"""
error_message_set, exit_status = ScalabilityLauncher().run() error_message_set, exit_status = ScalabilityLauncher().run()
for error_message in error_message_set: for error_message in error_message_set:
print >>sys.stderr, 'ERROR: %s' % error_message print >>sys.stderr, 'ERROR: %s' % error_message
sys.exit(exit_status) sys.exit(exit_status)
def runStandaloneResiliencyTest():
"""
Used to bypass the Test Node infrastructure and manually run a test.
Useful for running the test without any infrastructure.
"""
parser = argparse.ArgumentParser()
parser.add_argument('--test-suite-title')
parser.add_argument('additional_arguments', nargs=argparse.REMAINDER)
arguments = parser.parse_args()
runTestSuite(
arguments.test_suite_title,
arguments.additional_arguments,
setupLogging().info
)
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment