- """
- if self.current_test_case_info:
- self.current_test_case_info.logger.debug(
- "--- addSkip() %s.%s(%s) called, reason is %s" %
- (test.__class__.__name__, test._testMethodName,
- test._testMethodDoc, reason))
- unittest.TestResult.addSkip(self, test, reason)
- self.result_string = colorize("SKIP", YELLOW)
-
- self.send_result_through_pipe(test, SKIP)
-
- def symlink_failed(self):
- if self.current_test_case_info:
- try:
- failed_dir = os.getenv('FAILED_DIR')
- link_path = os.path.join(
- failed_dir,
- '%s-FAILED' %
- os.path.basename(self.current_test_case_info.tempdir))
- if self.current_test_case_info.logger:
- self.current_test_case_info.logger.debug(
- "creating a link to the failed test")
- self.current_test_case_info.logger.debug(
- "os.symlink(%s, %s)" %
- (self.current_test_case_info.tempdir, link_path))
- if os.path.exists(link_path):
- if self.current_test_case_info.logger:
- self.current_test_case_info.logger.debug(
- 'symlink already exists')
- else:
- os.symlink(self.current_test_case_info.tempdir, link_path)
-
- except Exception as e:
- if self.current_test_case_info.logger:
- self.current_test_case_info.logger.error(e)
-
- def send_result_through_pipe(self, test, result):
- if hasattr(self, 'test_framework_result_pipe'):
- pipe = self.test_framework_result_pipe
- if pipe:
- pipe.send((test.id(), result))
-
- def log_error(self, test, err, fn_name):
- if self.current_test_case_info:
- if isinstance(test, unittest.suite._ErrorHolder):
- test_name = test.description
- else:
- test_name = '%s.%s(%s)' % (test.__class__.__name__,
- test._testMethodName,
- test._testMethodDoc)
- self.current_test_case_info.logger.debug(
- "--- %s() %s called, err is %s" %
- (fn_name, test_name, err))
- self.current_test_case_info.logger.debug(
- "formatted exception is:\n%s" %
- "".join(format_exception(*err)))
-
- def add_error(self, test, err, unittest_fn, error_type):
- if error_type == FAIL:
- self.log_error(test, err, 'addFailure')
- error_type_str = colorize("FAIL", RED)
- elif error_type == ERROR:
- self.log_error(test, err, 'addError')
- error_type_str = colorize("ERROR", RED)
- else:
- raise Exception('Error type %s cannot be used to record an '
- 'error or a failure' % error_type)
-
- unittest_fn(self, test, err)
- if self.current_test_case_info:
- self.result_string = "%s [ temp dir used by test case: %s ]" % \
- (error_type_str,
- self.current_test_case_info.tempdir)
- self.symlink_failed()
- self.failed_test_cases_info.add(self.current_test_case_info)
- if is_core_present(self.current_test_case_info.tempdir):
- if not self.current_test_case_info.core_crash_test:
- if isinstance(test, unittest.suite._ErrorHolder):
- test_name = str(test)
- else:
- test_name = "'{!s}' ({!s})".format(
- get_testcase_doc_name(test), test.id())
- self.current_test_case_info.core_crash_test = test_name
- self.core_crash_test_cases_info.add(
- self.current_test_case_info)
- else:
- self.result_string = '%s [no temp dir]' % error_type_str
-
- self.send_result_through_pipe(test, error_type)
-
- def addFailure(self, test, err):
- """
- Record a test failed result
-
- :param test:
- :param err: error message
-
- """
- self.add_error(test, err, unittest.TestResult.addFailure, FAIL)
-
- def addError(self, test, err):
- """
- Record a test error result
-
- :param test:
- :param err: error message
-
- """
- self.add_error(test, err, unittest.TestResult.addError, ERROR)
-
- def getDescription(self, test):
- """
- Get test description
-
- :param test:
- :returns: test description
-
- """
- return get_test_description(self.descriptions, test)
-
- def startTest(self, test):
- """
- Start a test
-
- :param test:
-
- """
-
- def print_header(test):
- if not hasattr(test.__class__, '_header_printed'):
- print(double_line_delim)
- print(colorize(getdoc(test).splitlines()[0], GREEN))
- print(double_line_delim)
- test.__class__._header_printed = True
-
- print_header(test)
-
- unittest.TestResult.startTest(self, test)
- if self.verbosity > 0:
- self.stream.writeln(
- "Starting " + self.getDescription(test) + " ...")
- self.stream.writeln(single_line_delim)
-
- def stopTest(self, test):
- """
- Called when the given test has been run
-
- :param test:
-
- """
- unittest.TestResult.stopTest(self, test)
- if self.verbosity > 0:
- self.stream.writeln(single_line_delim)
- self.stream.writeln("%-73s%s" % (self.getDescription(test),
- self.result_string))
- self.stream.writeln(single_line_delim)
- else:
- self.stream.writeln("%-73s%s" % (self.getDescription(test),
- self.result_string))
-
- self.send_result_through_pipe(test, TEST_RUN)
-
- def printErrors(self):
- """
- Print errors from running the test case
- """
- if len(self.errors) > 0 or len(self.failures) > 0:
- self.stream.writeln()
- self.printErrorList('ERROR', self.errors)
- self.printErrorList('FAIL', self.failures)
-
- # ^^ that is the last output from unittest before summary
- if not self.runner.print_summary:
- devnull = unittest.runner._WritelnDecorator(open(os.devnull, 'w'))
- self.stream = devnull
- self.runner.stream = devnull
-
- def printErrorList(self, flavour, errors):
- """
- Print error list to the output stream together with error type
- and test case description.
-
- :param flavour: error type
- :param errors: iterable errors
-
- """
- for test, err in errors:
- self.stream.writeln(double_line_delim)
- self.stream.writeln("%s: %s" %
- (flavour, self.getDescription(test)))
- self.stream.writeln(single_line_delim)
- self.stream.writeln("%s" % err)
-
-
-class VppTestRunner(unittest.TextTestRunner):
- """
- A basic test runner implementation which prints results to standard error.
- """
-
- @property
- def resultclass(self):
- """Class maintaining the results of the tests"""
- return VppTestResult
-
- def __init__(self, keep_alive_pipe=None, descriptions=True, verbosity=1,
- result_pipe=None, failfast=False, buffer=False,
- resultclass=None, print_summary=True, **kwargs):
- # ignore stream setting here, use hard-coded stdout to be in sync
- # with prints from VppTestCase methods ...
- super(VppTestRunner, self).__init__(sys.stdout, descriptions,
- verbosity, failfast, buffer,
- resultclass, **kwargs)
- KeepAliveReporter.pipe = keep_alive_pipe
-
- self.orig_stream = self.stream
- self.resultclass.test_framework_result_pipe = result_pipe
-
- self.print_summary = print_summary
-
- def _makeResult(self):
- return self.resultclass(self.stream,
- self.descriptions,
- self.verbosity,
- self)
-
- def run(self, test):
- """
- Run the tests
-
- :param test:
-
- """
- faulthandler.enable() # emit stack trace to stderr if killed by signal
-
- result = super(VppTestRunner, self).run(test)
- if not self.print_summary:
- self.stream = self.orig_stream
- result.stream = self.orig_stream
- return result