Plugin support to pick test cases using marvin.

Using this plugin Nose will drive tests and this also allows using other
nose plugins. Cmd Line plugin support to be added
This commit is contained in:
Prasanna Santhanam 2012-07-19 23:27:18 +05:30
parent 135a999db8
commit 63cb30e196
4 changed files with 96 additions and 68 deletions

View File

@ -1,4 +1,3 @@
from functools import partial
import unittest
import nose
import nose.core
@ -7,59 +6,16 @@ import sys
import logging
import xmlrunner
from cloudstackTestCase import cloudstackTestCase
from marvinPlugin import MarvinPlugin
from nose.plugins.xunit import Xunit
def testCaseLogger(message, logger=None):
if logger is not None:
logger.debug(message)
class NoseCloudStackTestLoader(nose.loader.TestLoader):
"""
Custom test loader for the cloudstackTestCase to be loaded into nose
"""
def loadTestsFromTestCase(self, testCaseClass):
if issubclass(testCaseClass, cloudstackTestCase):
testCaseNames = self.getTestCaseNames(testCaseClass)
tests = []
for testCaseName in testCaseNames:
testCase = testCaseClass(testCaseName)
self._injectClients(testCase)
tests.append(testCase)
return self.suiteClass(tests)
else:
return super(NoseCloudStackTestLoader, self).loadTestsFromTestCase(testCaseClass)
def loadTestsFromName(self, name, module=None, discovered=False):
return nose.loader.TestLoader.loadTestsFromName(self, name, module=module, discovered=discovered)
def loadTestsFromNames(self, names, module=None):
return nose.loader.TestLoader.loadTestsFromNames(self, names, module=module)
def setClient(self, client):
self.testclient = client
def setClientLog(self, clientlog):
self.log = clientlog
def _injectClients(self, test):
testcaselogger = logging.getLogger("testclient.testcase.%s"%test.__class__.__name__)
fh = logging.FileHandler(self.log)
fh.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
testcaselogger.addHandler(fh)
testcaselogger.setLevel(logging.DEBUG)
setattr(test, "testClient", self.testclient)
setattr(test, "debug", partial(testCaseLogger, logger=testcaselogger))
setattr(test.__class__, "clstestclient", self.testclient)
if hasattr(test, "UserName"):
self.testclient.createNewApiClient(test.UserName, test.DomainName, test.AcctType)
class NoseTestExecuteEngine(object):
"""
Runs the CloudStack tests using nose as the execution engine
"""
def __init__(self, testclient=None, workingdir=None, filename=None, clientLog=None, resultLog=None, format="text"):
def __init__(self, testclient=None, workingdir=None, filename=None, clientLog=None, resultLog=None, format="text", xmlDir='xml-reports'):
self.testclient = testclient
self.logformat = logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s")
self.suite = []
@ -82,24 +38,43 @@ class NoseTestExecuteEngine(object):
self.testResultLogFile = sys.stderr
if workingdir is not None:
self.loader = NoseCloudStackTestLoader()
self.loader.setClient(self.testclient)
self.loader.setClientLog(self.logfile)
self.suite = self.loader.loadTestsFromDir(workingdir)
self.workingdir = workingdir
self.test_picker = MarvinPlugin()
self.test_picker.setClient(self.testclient)
self.test_picker.setClientLog(self.logfile)
elif filename is not None:
self.loader = NoseCloudStackTestLoader()
self.loader.setClient(self.testclient)
self.loader.setClientLog(self.logfile)
self.suite = self.loader.loadTestsFromFile(filename)
self.test_picker = MarvinPlugin()
self.test_picker.setClient(self.testclient)
self.test_picker.setClientLog(self.logfile)
self.filename = filename
else:
raise EnvironmentError("Need to give either a test directory or a test file")
plug_mgr = nose.plugins.manager.PluginManager()
plug_mgr.addPlugin(self.test_picker)
self.cfg = nose.config.Config()
self.cfg.plugins = plug_mgr
if format == "text":
self.runner = nose.core.TextTestRunner(stream=self.testResultLogFile, descriptions=1, verbosity=2, config=None)
self.runner = \
nose.core.TextTestRunner(stream=self.testResultLogFile,
descriptions=1, verbosity=2, config=self.cfg)
else:
self.runner = xmlrunner.XMLTestRunner(output='xml-reports', verbose=True)
self.runner = xmlrunner.XMLTestRunner(output=xmlDir, verbose=True)
def runTests(self):
#nose.core.TestProgram(argv=["--process-timeout=3600"], testRunner=self.runner, testLoader=self.loader)
nose.core.TestProgram(argv=["--process-timeout=3600"], \
testRunner=self.runner, suite=self.suite)
#options = ["--process-timeout=3600", "--with-xunit", "-a tags=advanced"] #TODO: Add support for giving nose args
#DEBUG
# options = ["--process-timeout=3600", "--with-xunit", "--collect-only"]
#DEBUG
options = ["--process-timeout=3600"]
options.append("-w%s" %self.workingdir)
if self.workingdir is not None:
nose.core.TestProgram(argv=options, testRunner=self.runner,
config=self.cfg)
elif self.filename is not None:
tests = self.loader.loadTestsFromFile(self.filename)
nose.core.TestProgram(argv=options, testRunner=self.runner,
config=self.cfg)

View File

@ -54,8 +54,6 @@ class TestCaseExecuteEngine(object):
if self.format == "xml" and (xmlDir is not None):
self.xmlDir = xmlDir
def loadTestsFromDir(self, testDirectory):
""" Load the test suites from a package with multiple test files """
self.suite = self.loader.discover(testDirectory)

View File

@ -17,7 +17,8 @@ from optparse import OptionParser
import os
if __name__ == "__main__":
parser = OptionParser()
parser = OptionParser() #TODO: deprecate and use the argparse module
parser.add_option("-c", "--config", action="store", default="./datacenterCfg", dest="config", help="the path where the json config file generated, by default is ./datacenterCfg")
parser.add_option("-d", "--directory", dest="testCaseFolder", help="the test case directory")
@ -26,7 +27,7 @@ if __name__ == "__main__":
parser.add_option("-l", "--load", dest="load", action="store_true", help="only load config, do not deploy, it will only run testcase")
parser.add_option("-f", "--file", dest="module", help="run tests in the given file")
parser.add_option("-n", "--nose", dest="nose", action="store_true", help="run tests using nose")
parser.add_option("-x", "--xml", dest="xmlrunner", action="store", default="./xml-reports", help="use the xml runner to generate xml reports and path to store xml files")
parser.add_option("-x", "--xml", dest="xmlrunner", help="use the xml runner to generate xml reports and path to store xml files")
(options, args) = parser.parse_args()
testResultLogFile = None
@ -42,8 +43,8 @@ if __name__ == "__main__":
else:
deploy.deploy()
format = "text"
xmlDir = "xml-reports"
format = "text"
xmlDir = None
if options.xmlrunner is not None:
xmlDir = options.xmlrunner
format = "xml"
@ -54,7 +55,7 @@ if __name__ == "__main__":
exit(1)
else:
if options.nose:
engine = NoseTestExecuteEngine.NoseTestExecuteEngine(deploy.testClient, testCaseLogFile, testResultLogFile, format)
engine = NoseTestExecuteEngine.NoseTestExecuteEngine(deploy.testClient, testCaseLogFile, testResultLogFile, format, xmlDir)
engine.runTestsFromFile(options.module)
else:
engine = TestCaseExecuteEngine.TestCaseExecuteEngine(deploy.testClient, testCaseLogFile, testResultLogFile, format, xmlDir)
@ -62,7 +63,7 @@ if __name__ == "__main__":
engine.run()
else:
if options.nose:
engine = NoseTestExecuteEngine.NoseTestExecuteEngine(deploy.testClient, clientLog=testCaseLogFile, resultLog=testResultLogFile, workingdir=options.testCaseFolder, format=format)
engine = NoseTestExecuteEngine.NoseTestExecuteEngine(deploy.testClient, clientLog=testCaseLogFile, resultLog=testResultLogFile, workingdir=options.testCaseFolder, format=format, xmlDir=xmlDir)
engine.runTests()
else:
engine = TestCaseExecuteEngine.TestCaseExecuteEngine(deploy.testClient, testCaseLogFile, testResultLogFile, format, xmlDir)

View File

@ -0,0 +1,54 @@
from cloudstackTestCase import cloudstackTestCase
from nose.plugins.base import Plugin
from functools import partial
import logging
def testCaseLogger(message, logger=None):
if logger is not None:
logger.debug(message)
class MarvinPlugin(Plugin):
"""
Custom test loader for the cloudstackTestCase to be loaded into nose
"""
name = "marvin"
def configure(self, options, conf):
self.enabled = 1
self.enableOpt = "--with-marvin"
return Plugin.configure(self, options, conf)
def options(self, parser, env):
Plugin.options(self, parser, env)
def __init__(self):
Plugin.__init__(self)
def wantClass(self, cls):
if issubclass(cls, cloudstackTestCase):
return True
return None
def loadTestsFromTestCase(self, cls):
self._injectClients(cls)
def setClient(self, client):
if client:
self.testclient = client
def setClientLog(self, clientlog):
if clientlog:
self.log = clientlog
def _injectClients(self, test):
testcaselogger = logging.getLogger("testclient.testcase.%s" % test.__class__.__name__)
fh = logging.FileHandler(self.log)
fh.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
testcaselogger.addHandler(fh)
testcaselogger.setLevel(logging.DEBUG)
setattr(test, "testClient", self.testclient)
setattr(test, "debug", partial(testCaseLogger, logger=testcaselogger))
setattr(test, "clstestclient", self.testclient)
if hasattr(test, "UserName"):
self.testclient.createNewApiClient(test.UserName, test.DomainName, test.AcctType)