diff options
Diffstat (limited to 'python/helpers/pydev/pydev_runfiles_nose.py')
| -rw-r--r-- | python/helpers/pydev/pydev_runfiles_nose.py | 180 |
1 files changed, 180 insertions, 0 deletions
diff --git a/python/helpers/pydev/pydev_runfiles_nose.py b/python/helpers/pydev/pydev_runfiles_nose.py new file mode 100644 index 000000000000..422d2a62a83a --- /dev/null +++ b/python/helpers/pydev/pydev_runfiles_nose.py @@ -0,0 +1,180 @@ +from nose.plugins.multiprocess import MultiProcessTestRunner # @UnresolvedImport +from nose.plugins.base import Plugin # @UnresolvedImport +import sys +import pydev_runfiles_xml_rpc +import time +from pydev_runfiles_coverage import StartCoverageSupport + +#======================================================================================================================= +# PydevPlugin +#======================================================================================================================= +class PydevPlugin(Plugin): + + def __init__(self, configuration): + self.configuration = configuration + Plugin.__init__(self) + + + def begin(self): + # Called before any test is run (it's always called, with multiprocess or not) + self.start_time = time.time() + self.coverage_files, self.coverage = StartCoverageSupport(self.configuration) + + + def finalize(self, result): + # Called after all tests are run (it's always called, with multiprocess or not) + self.coverage.stop() + self.coverage.save() + + pydev_runfiles_xml_rpc.notifyTestRunFinished('Finished in: %.2f secs.' % (time.time() - self.start_time,)) + + + + #=================================================================================================================== + # Methods below are not called with multiprocess (so, we monkey-patch MultiProcessTestRunner.consolidate + # so that they're called, but unfortunately we loose some info -- i.e.: the time for each test in this + # process). + #=================================================================================================================== + + + def reportCond(self, cond, test, captured_output, error=''): + ''' + @param cond: fail, error, ok + ''' + + # test.address() is something as: + # ('D:\\workspaces\\temp\\test_workspace\\pytesting1\\src\\mod1\\hello.py', 'mod1.hello', 'TestCase.testMet1') + # + # and we must pass: location, test + # E.g.: ['D:\\src\\mod1\\hello.py', 'TestCase.testMet1'] + try: + if hasattr(test, 'address'): + address = test.address() + address = address[0], address[2] + else: + # multiprocess + try: + address = test[0], test[1] + except TypeError: + # It may be an error at setup, in which case it's not really a test, but a Context object. + f = test.context.__file__ + if f.endswith('.pyc'): + f = f[:-1] + address = f, '?' + except: + sys.stderr.write("PyDev: Internal pydev error getting test address. Please report at the pydev bug tracker\n") + import traceback;traceback.print_exc() + sys.stderr.write("\n\n\n") + address = '?', '?' + + error_contents = self.getIoFromError(error) + try: + time_str = '%.2f' % (time.time() - test._pydev_start_time) + except: + time_str = '?' + + pydev_runfiles_xml_rpc.notifyTest(cond, captured_output, error_contents, address[0], address[1], time_str) + + + def startTest(self, test): + test._pydev_start_time = time.time() + if hasattr(test, 'address'): + address = test.address() + file, test = address[0], address[2] + else: + # multiprocess + file, test = test + pydev_runfiles_xml_rpc.notifyStartTest(file, test) + + + def getIoFromError(self, err): + if type(err) == type(()): + if len(err) != 3: + if len(err) == 2: + return err[1] # multiprocess + try: + from StringIO import StringIO + except: + from io import StringIO + s = StringIO() + etype, value, tb = err + import traceback;traceback.print_exception(etype, value, tb, file=s) + return s.getvalue() + return err + + + def getCapturedOutput(self, test): + if hasattr(test, 'capturedOutput') and test.capturedOutput: + return test.capturedOutput + return '' + + + def addError(self, test, err): + self.reportCond( + 'error', + test, + self.getCapturedOutput(test), + err, + ) + + + def addFailure(self, test, err): + self.reportCond( + 'fail', + test, + self.getCapturedOutput(test), + err, + ) + + + def addSuccess(self, test): + self.reportCond( + 'ok', + test, + self.getCapturedOutput(test), + '', + ) + + +PYDEV_NOSE_PLUGIN_SINGLETON = None +def StartPydevNosePluginSingleton(configuration): + global PYDEV_NOSE_PLUGIN_SINGLETON + PYDEV_NOSE_PLUGIN_SINGLETON = PydevPlugin(configuration) + return PYDEV_NOSE_PLUGIN_SINGLETON + + + + +original = MultiProcessTestRunner.consolidate +#======================================================================================================================= +# NewConsolidate +#======================================================================================================================= +def NewConsolidate(self, result, batch_result): + ''' + Used so that it can work with the multiprocess plugin. + Monkeypatched because nose seems a bit unsupported at this time (ideally + the plugin would have this support by default). + ''' + ret = original(self, result, batch_result) + + parent_frame = sys._getframe().f_back + # addr is something as D:\pytesting1\src\mod1\hello.py:TestCase.testMet4 + # so, convert it to what reportCond expects + addr = parent_frame.f_locals['addr'] + i = addr.rindex(':') + addr = [addr[:i], addr[i + 1:]] + + output, testsRun, failures, errors, errorClasses = batch_result + if failures or errors: + for failure in failures: + PYDEV_NOSE_PLUGIN_SINGLETON.reportCond('fail', addr, output, failure) + + for error in errors: + PYDEV_NOSE_PLUGIN_SINGLETON.reportCond('error', addr, output, error) + else: + PYDEV_NOSE_PLUGIN_SINGLETON.reportCond('ok', addr, output) + + + return ret + +MultiProcessTestRunner.consolidate = NewConsolidate |