from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
from scapy.layers.inet6 import ICMPv6EchoReply
from vpp_running import use_running
+from test_result_code import TestResultCode
logger = logging.getLogger(__name__)
null_logger = logging.getLogger("VppTestCase")
null_logger.addHandler(logging.NullHandler())
-PASS = 0
-FAIL = 1
-ERROR = 2
-SKIP = 3
-TEST_RUN = 4
-SKIP_CPU_SHORTAGE = 5
-
if config.debug_framework:
import debug_internal
self.stream = stream
self.descriptions = descriptions
self.verbosity = verbosity
+ self.result_code = TestResultCode.TEST_RUN
self.result_string = None
self.runner = runner
self.printed = []
: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)
- )
+ self.log_result("addSuccess", test)
unittest.TestResult.addSuccess(self, test)
self.result_string = colorize("OK", GREEN)
-
- self.send_result_through_pipe(test, PASS)
+ self.result_code = TestResultCode.PASS
+ self.send_result_through_pipe(test, self.result_code)
+
+ def addExpectedFailure(self, test, err):
+ self.log_result("addExpectedFailure", test, err)
+ super().addExpectedFailure(test, err)
+ self.result_string = colorize("FAIL", GREEN)
+ self.result_code = TestResultCode.EXPECTED_FAIL
+ self.send_result_through_pipe(test, self.result_code)
+
+ def addUnexpectedSuccess(self, test):
+ self.log_result("addUnexpectedSuccess", test)
+ super().addUnexpectedSuccess(test)
+ self.result_string = colorize("OK", RED)
+ self.result_code = TestResultCode.UNEXPECTED_PASS
+ self.send_result_through_pipe(test, self.result_code)
def addSkip(self, test, reason):
"""
: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,
- )
- )
+ self.log_result("addSkip", test, reason=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)
+ self.result_code = TestResultCode.SKIP_CPU_SHORTAGE
else:
- self.send_result_through_pipe(test, SKIP)
+ self.result_code = TestResultCode.SKIP
+ self.send_result_through_pipe(test, self.result_code)
def symlink_failed(self):
if self.current_test_case_info:
if pipe:
pipe.send((test.id(), result))
- def log_error(self, test, err, fn_name):
+ def log_result(self, fn, test, err=None, reason=None):
if self.current_test_case_info:
if isinstance(test, unittest.suite._ErrorHolder):
test_name = test.description
test._testMethodName,
test._testMethodDoc,
)
+ extra_msg = ""
+ if err:
+ extra_msg += f", error is {err}"
+ if reason:
+ extra_msg += f", reason is {reason}"
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))
+ f"--- {fn}() {test_name} called{extra_msg}"
)
+ if 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")
+ def add_error(self, test, err, unittest_fn, result_code):
+ self.result_code = result_code
+ if result_code == TestResultCode.FAIL:
+ self.log_result("addFailure", test, err=err)
error_type_str = colorize("FAIL", RED)
- elif error_type == ERROR:
- self.log_error(test, err, "addError")
+ elif result_code == TestResultCode.ERROR:
+ self.log_result("addError", test, err=err)
error_type_str = colorize("ERROR", RED)
else:
- raise Exception(
- "Error type %s cannot be used to record an "
- "error or a failure" % error_type
- )
+ raise Exception(f"Unexpected result code {result_code}")
unittest_fn(self, test, err)
if self.current_test_case_info:
else:
self.result_string = "%s [no temp dir]" % error_type_str
- self.send_result_through_pipe(test, error_type)
+ self.send_result_through_pipe(test, result_code)
def addFailure(self, test, err):
"""
:param err: error message
"""
- self.add_error(test, err, unittest.TestResult.addFailure, FAIL)
+ self.add_error(test, err, unittest.TestResult.addFailure, TestResultCode.FAIL)
def addError(self, test, err):
"""
:param err: error message
"""
- self.add_error(test, err, unittest.TestResult.addError, ERROR)
+ self.add_error(test, err, unittest.TestResult.addError, TestResultCode.ERROR)
def getDescription(self, test):
"""
"""
unittest.TestResult.stopTest(self, test)
+ result_code_to_suffix = {
+ TestResultCode.PASS: "",
+ TestResultCode.FAIL: "",
+ TestResultCode.ERROR: "",
+ TestResultCode.SKIP: "",
+ TestResultCode.TEST_RUN: "",
+ TestResultCode.SKIP_CPU_SHORTAGE: "",
+ TestResultCode.EXPECTED_FAIL: " [EXPECTED FAIL]",
+ TestResultCode.UNEXPECTED_PASS: " [UNEXPECTED PASS]",
+ }
+
if self.verbosity > 0:
self.stream.writeln(single_line_delim)
self.stream.writeln(
- "%-73s%s" % (self.getDescription(test), self.result_string)
+ "%-72s%s%s"
+ % (
+ self.getDescription(test),
+ self.result_string,
+ result_code_to_suffix[self.result_code],
+ )
)
self.stream.writeln(single_line_delim)
else:
self.stream.writeln(
- "%-68s %4.2f %s"
+ "%-67s %4.2f %s%s"
% (
self.getDescription(test),
time.time() - self.start_test,
self.result_string,
+ result_code_to_suffix[self.result_code],
)
)
- self.send_result_through_pipe(test, TEST_RUN)
+ self.send_result_through_pipe(test, TestResultCode.TEST_RUN)
def printErrors(self):
"""
Code Review / vpp.git / blobdiff