-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
- self.printed = []
-
- 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)
-
- if reason == "not enough cpus":
- self.send_result_through_pipe(test, SKIP_CPU_SHORTAGE)
- else:
- self.send_result_through_pipe(test, SKIP)
-
- def symlink_failed(self):
- if self.current_test_case_info:
- try:
- failed_dir = config.failed_dir
- link_path = os.path.join(
- failed_dir,
- "%s-FAILED" % os.path.basename(self.current_test_case_info.tempdir),
- )
-
- 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):
- 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:
- 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 test.__class__ in self.printed:
- return
-
- test_doc = getdoc(test)
- if not test_doc:
- raise Exception("No doc string for test '%s'" % test.id())
-
- test_title = test_doc.splitlines()[0].rstrip()
- test_title = colorize(test_title, GREEN)
- if test.is_tagged_run_solo():
- test_title = colorize(f"SOLO RUN: {test_title}", YELLOW)
-
- # This block may overwrite the colorized title above,
- # but we want this to stand out and be fixed
- if test.has_tag(TestCaseTag.FIXME_VPP_WORKERS):
- test_title = colorize(f"FIXME with VPP workers: {test_title}", RED)
-
- if test.has_tag(TestCaseTag.FIXME_ASAN):
- test_title = colorize(f"FIXME with ASAN: {test_title}", RED)
- test.skip_fixme_asan()
-
- if is_distro_ubuntu2204 == True and test.has_tag(
- TestCaseTag.FIXME_UBUNTU2204
- ):
- test_title = colorize(f"FIXME on Ubuntu-22.04: {test_title}", RED)
- test.skip_fixme_ubuntu2204()
-
- if hasattr(test, "vpp_worker_count"):
- if test.vpp_worker_count == 0:
- test_title += " [main thread only]"
- elif test.vpp_worker_count == 1:
- test_title += " [1 worker thread]"
- else:
- test_title += f" [{test.vpp_worker_count} worker threads]"
-
- if test.__class__.skipped_due_to_cpu_lack:
- test_title = colorize(
- f"{test_title} [skipped - not enough cpus, "
- f"required={test.__class__.get_cpus_required()}, "
- f"available={max_vpp_cpus}]",
- YELLOW,
- )
-
- print(double_line_delim)
- print(test_title)
- print(double_line_delim)
- self.printed.append(test.__class__)
-
- print_header(test)
- self.start_test = time.time()
- 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(
- "%-68s %4.2f %s"
- % (
- self.getDescription(test),
- time.time() - self.start_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
-
-
-class Worker(Thread):
- def __init__(self, executable_args, logger, env=None, *args, **kwargs):
- super(Worker, self).__init__(*args, **kwargs)
- self.logger = logger
- self.args = executable_args
- if hasattr(self, "testcase") and self.testcase.debug_all:
- if self.testcase.debug_gdbserver:
- self.args = [
- "/usr/bin/gdbserver",
- "localhost:{port}".format(port=self.testcase.gdbserver_port),
- ] + args
- elif self.testcase.debug_gdb and hasattr(self, "wait_for_gdb"):
- self.args.append(self.wait_for_gdb)
- self.app_bin = executable_args[0]
- self.app_name = os.path.basename(self.app_bin)
- if hasattr(self, "role"):
- self.app_name += " {role}".format(role=self.role)
- self.process = None
- self.result = None
- env = {} if env is None else env
- self.env = copy.deepcopy(env)
-
- def wait_for_enter(self):
- if not hasattr(self, "testcase"):
- return
- if self.testcase.debug_all and self.testcase.debug_gdbserver:
- print()
- print(double_line_delim)
- print(
- "Spawned GDB Server for '{app}' with PID: {pid}".format(
- app=self.app_name, pid=self.process.pid
- )
- )
- elif self.testcase.debug_all and self.testcase.debug_gdb:
- print()
- print(double_line_delim)
- print(
- "Spawned '{app}' with PID: {pid}".format(
- app=self.app_name, pid=self.process.pid
- )
- )
- else:
- return
- print(single_line_delim)
- print("You can debug '{app}' using:".format(app=self.app_name))
- if self.testcase.debug_gdbserver:
- print(
- "sudo gdb "
- + self.app_bin
- + " -ex 'target remote localhost:{port}'".format(
- port=self.testcase.gdbserver_port
- )
- )
- print(
- "Now is the time to attach gdb by running the above "
- "command, set up breakpoints etc., then resume from "
- "within gdb by issuing the 'continue' command"
- )
- self.testcase.gdbserver_port += 1
- elif self.testcase.debug_gdb:
- print(
- "sudo gdb "
- + self.app_bin
- + " -ex 'attach {pid}'".format(pid=self.process.pid)
- )
- print(
- "Now is the time to attach gdb by running the above "
- "command and set up breakpoints etc., then resume from"
- " within gdb by issuing the 'continue' command"
- )
- print(single_line_delim)
- input("Press ENTER to continue running the testcase...")
-
- def run(self):
- executable = self.args[0]
- if not os.path.exists(executable) or not os.access(
- executable, os.F_OK | os.X_OK
- ):
- # Exit code that means some system file did not exist,
- # could not be opened, or had some other kind of error.
- self.result = os.EX_OSFILE
- raise EnvironmentError(
- "executable '%s' is not found or executable." % executable
- )
- self.logger.debug(
- "Running executable '{app}': '{cmd}'".format(
- app=self.app_name, cmd=" ".join(self.args)
- )
- )
- env = os.environ.copy()
- env.update(self.env)
- env["CK_LOG_FILE_NAME"] = "-"
- self.process = subprocess.Popen(
- ["stdbuf", "-o0", "-e0"] + self.args,
- shell=False,
- env=env,
- preexec_fn=os.setpgrp,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- )
- self.wait_for_enter()
- out, err = self.process.communicate()
- self.logger.debug("Finished running `{app}'".format(app=self.app_name))
- self.logger.info("Return code is `%s'" % self.process.returncode)
- self.logger.info(single_line_delim)
- self.logger.info(
- "Executable `{app}' wrote to stdout:".format(app=self.app_name)
- )
- self.logger.info(single_line_delim)
- self.logger.info(out.decode("utf-8"))
- self.logger.info(single_line_delim)
- self.logger.info(
- "Executable `{app}' wrote to stderr:".format(app=self.app_name)
- )
- self.logger.info(single_line_delim)
- self.logger.info(err.decode("utf-8"))
- self.logger.info(single_line_delim)
- self.result = self.process.returncode
-
-