- return rx
-
- def runTest(self):
- """ unittest calls runTest when TestCase is instantiated without a
- test case. Use case: Writing unittests against VppTestCase"""
- pass
-
-
-def get_testcase_doc_name(test):
- return getdoc(test.__class__).splitlines()[0]
-
-
-def get_test_description(descriptions, test):
- short_description = test.shortDescription()
- if descriptions and short_description:
- return short_description
- else:
- return str(test)
-
-
-class TestCaseInfo(object):
- def __init__(self, logger, tempdir, vpp_pid, vpp_bin_path):
- self.logger = logger
- self.tempdir = tempdir
- self.vpp_pid = vpp_pid
- self.vpp_bin_path = vpp_bin_path
- self.core_crash_test = None
-
-
-class VppTestResult(unittest.TestResult):
- """
- @property result_string
- String variable to store the test case result string.
- @property errors
- List variable containing 2-tuples of TestCase instances and strings
- holding formatted tracebacks. Each tuple represents a test which
- raised an unexpected exception.
- @property failures
- List variable containing 2-tuples of TestCase instances and strings
- holding formatted tracebacks. Each tuple represents a test where
- a failure was explicitly signalled using the TestCase.assert*()
- methods.
- """
-
- failed_test_cases_info = set()
- core_crash_test_cases_info = set()
- current_test_case_info = None
-
- def __init__(self, stream=None, descriptions=None, verbosity=None,
- runner=None):
- """
- :param stream File descriptor to store where to report test results.
- Set to the standard error stream by default.
- :param descriptions Boolean variable to store information if to use
- test case descriptions.
- :param verbosity Integer variable to store required verbosity level.
- """
- super(VppTestResult, self).__init__(stream, descriptions, verbosity)
- self.stream = stream
- self.descriptions = descriptions
- self.verbosity = verbosity
- self.result_string = None
- self.runner = runner
-
- def addSuccess(self, test):
- """
- Record a test succeeded result
-
- :param test:
-
- """
- if self.current_test_case_info:
- self.current_test_case_info.logger.debug(
- "--- addSuccess() %s.%s(%s) called" % (test.__class__.__name__,
- test._testMethodName,
- test._testMethodDoc))
- unittest.TestResult.addSuccess(self, test)
- self.result_string = colorize("OK", GREEN)
-
- self.send_result_through_pipe(test, PASS)
-
- def addSkip(self, test, reason):
- """
- Record a test skipped.
-
- :param test:
- :param reason:
-
- """
- 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)