from vpp_sub_interface import VppSubInterface
from vpp_lo_interface import VppLoInterface
from vpp_papi_provider import VppPapiProvider
+from vpp_papi.vpp_stats import VPPStats
from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
- getLogger, colorize
+ get_logger, colorize
from vpp_object import VppObjectRegistry
-from util import ppp
+from util import ppp, is_core_present
from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
from scapy.layers.inet6 import ICMPv6EchoReply
import subprocess
+PASS = 0
+FAIL = 1
+ERROR = 2
+SKIP = 3
+TEST_RUN = 4
+
+
debug_framework = False
if os.getenv('TEST_DEBUG', "0") == "1":
debug_framework = True
def __init__(self):
self.__dict__ = self._shared_state
+ self._pipe = None
@property
def pipe(self):
@pipe.setter
def pipe(self, pipe):
- if hasattr(self, '_pipe'):
+ if self._pipe is not None:
raise Exception("Internal error - pipe should only be set once.")
self._pipe = pipe
- def send_keep_alive(self, test):
+ def send_keep_alive(self, test, desc=None):
"""
Write current test tmpdir & desc to keep-alive pipe to signal liveness
"""
return
if isclass(test):
- desc = test.__name__
+ desc = '%s (%s)' % (desc, unittest.util.strclass(test))
else:
- desc = test.shortDescription()
- if not desc:
- desc = str(test)
+ desc = test.id()
self.pipe.send((desc, test.vpp_bin, test.tempdir, test.vpp.pid))
return random.choice(tuple(min_usage_set))
+ @staticmethod
+ def print_header(cls):
+ if not hasattr(cls, '_header_printed'):
+ print(double_line_delim)
+ print(colorize(getdoc(cls).splitlines()[0], GREEN))
+ print(double_line_delim)
+ cls._header_printed = True
+
@classmethod
def setUpConstants(cls):
""" Set-up the test case class based on environment variables """
cls.vpp_cmdline = [cls.vpp_bin, "unix",
"{", "nodaemon", debug_cli, "full-coredump",
- coredump_size, "}", "api-trace", "{", "on", "}",
- "api-segment", "{", "prefix", cls.shm_prefix, "}",
- "cpu", "{", "main-core", str(cpu_core_number), "}",
- "plugins", "{", "plugin", "dpdk_plugin.so", "{",
- "disable", "}", "plugin", "unittest_plugin.so",
- "{", "enable", "}", "}", ]
+ coredump_size, "runtime-dir", cls.tempdir, "}",
+ "api-trace", "{", "on", "}", "api-segment", "{",
+ "prefix", cls.shm_prefix, "}", "cpu", "{",
+ "main-core", str(cpu_core_number), "}", "statseg",
+ "{", "socket-name", cls.stats_sock, "}", "plugins",
+ "{", "plugin", "dpdk_plugin.so", "{", "disable",
+ "}", "plugin", "unittest_plugin.so", "{", "enable",
+ "}", "}", ]
if plugin_path is not None:
cls.vpp_cmdline.extend(["plugin_path", plugin_path])
- cls.logger.info("vpp_cmdline: %s" % cls.vpp_cmdline)
+ cls.logger.info("vpp_cmdline args: %s" % cls.vpp_cmdline)
+ cls.logger.info("vpp_cmdline: %s" % " ".join(cls.vpp_cmdline))
@classmethod
def wait_for_enter(cls):
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
bufsize=1)
- except Exception as e:
+ except subprocess.CalledProcessError as e:
cls.logger.critical("Couldn't start vpp: %s" % e)
raise
cls.wait_for_enter()
+ @classmethod
+ def wait_for_stats_socket(cls):
+ deadline = time.time() + 3
+ ok = False
+ while time.time() < deadline or \
+ cls.debug_gdb or cls.debug_gdbserver:
+ if os.path.exists(cls.stats_sock):
+ ok = True
+ break
+ time.sleep(0.8)
+ if not ok:
+ cls.logger.critical("Couldn't stat : {}".format(cls.stats_sock))
+
@classmethod
def setUpClass(cls):
"""
"""
gc.collect() # run garbage collection first
random.seed()
- if not hasattr(cls, 'logger'):
- cls.logger = getLogger(cls.__name__)
- else:
- cls.logger.name = cls.__name__
+ cls.print_header(cls)
+ cls.logger = get_logger(cls.__name__)
+ if hasattr(cls, 'parallel_handler'):
+ cls.logger.addHandler(cls.parallel_handler)
+ cls.logger.propagate = False
cls.tempdir = tempfile.mkdtemp(
prefix='vpp-unittest-%s-' % cls.__name__)
+ cls.stats_sock = "%s/stats.sock" % cls.tempdir
cls.file_handler = FileHandler("%s/log.txt" % cls.tempdir)
cls.file_handler.setFormatter(
Formatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
# doesn't get called and we might end with a zombie vpp
try:
cls.run_vpp()
- cls.reporter.send_keep_alive(cls)
+ cls.reporter.send_keep_alive(cls, 'setUpClass')
+ VppTestResult.current_test_case_info = TestCaseInfo(
+ cls.logger, cls.tempdir, cls.vpp.pid, cls.vpp_bin)
cls.vpp_stdout_deque = deque()
cls.vpp_stderr_deque = deque()
cls.pump_thread_stop_flag = Event()
cls.pump_thread = Thread(target=pump_output, args=(cls,))
cls.pump_thread.daemon = True
cls.pump_thread.start()
- cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix, cls)
+ if cls.debug_gdb or cls.debug_gdbserver:
+ read_timeout = 0
+ else:
+ read_timeout = 5
+ cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix, cls,
+ read_timeout)
if cls.step:
hook = StepHook(cls)
else:
hook = PollHook(cls)
cls.vapi.register_hook(hook)
- cls.sleep(0.1, "after vpp startup, before initial poll")
+ cls.wait_for_stats_socket()
+ cls.statistics = VPPStats(socketname=cls.stats_sock)
try:
hook.poll_vpp()
except VppDiedError:
cls.vpp.poll()
if cls.vpp.returncode is None:
cls.logger.debug("Sending TERM to vpp")
- cls.vpp.terminate()
+ cls.vpp.kill()
cls.logger.debug("Waiting for vpp to die")
cls.vpp.communicate()
del cls.vpp
@classmethod
def tearDownClass(cls):
""" Perform final cleanup after running all tests in this test-case """
+ cls.reporter.send_keep_alive(cls, 'tearDownClass')
cls.quit()
cls.file_handler.close()
cls.reset_packet_infos()
self.logger.debug(self.vapi.cli("show trace"))
self.logger.info(self.vapi.ppcli("show interface"))
self.logger.info(self.vapi.ppcli("show hardware"))
- self.logger.info(self.vapi.ppcli("show error"))
+ self.logger.info(self.statistics.set_errors_str())
self.logger.info(self.vapi.ppcli("show run"))
self.logger.info(self.vapi.ppcli("show log"))
self.registry.remove_vpp_config(self.logger)
if pkt.haslayer(ICMPv6EchoReply):
self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
+ def assert_packet_counter_equal(self, counter, expected_value):
+ counters = self.vapi.cli("sh errors").split('\n')
+ counter_value = -1
+ for i in range(1, len(counters)-1):
+ results = counters[i].split()
+ if results[1] == counter:
+ counter_value = int(results[0])
+ break
+ self.assert_equal(counter_value, expected_value,
+ "packet counter `%s'" % counter)
+
@classmethod
def sleep(cls, timeout, remark=None):
if hasattr(cls, 'logger'):
input.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- rx = output.get_capture(len(pkts))
+ if isinstance(object, (list,)):
+ rx = []
+ for o in output:
+ rx.append(output.get_capture(len(pkts)))
+ else:
+ rx = output.get_capture(len(pkts))
+ return rx
+
+ def send_and_expect_only(self, input, pkts, output, timeout=None):
+ self.vapi.cli("clear trace")
+ input.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ if isinstance(object, (list,)):
+ outputs = output
+ rx = []
+ for o in outputs:
+ rx.append(output.get_capture(len(pkts)))
+ else:
+ rx = output.get_capture(len(pkts))
+ outputs = [output]
+ if not timeout:
+ timeout = 1
+ for i in self.pg_interfaces:
+ if i not in outputs:
+ i.get_capture(0, timeout=timeout)
+ i.assert_nothing_captured()
+ timeout = 0.1
+
return rx
def get_test_description(descriptions, test):
- # TODO: if none print warning not raise exception
short_description = test.shortDescription()
if descriptions and short_description:
return short_description
return str(test)
-class TestCasePrinter(object):
- _shared_state = {}
-
- def __init__(self):
- self.__dict__ = self._shared_state
- if not hasattr(self, "_test_case_set"):
- self._test_case_set = set()
-
- def print_test_case_heading_if_first_time(self, case):
- if case.__class__ not in self._test_case_set:
- print(double_line_delim)
- print(colorize(get_testcase_doc_name(case), GREEN))
- print(double_line_delim)
- self._test_case_set.add(case.__class__)
+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):
methods.
"""
+ failed_test_cases_info = set()
+ core_crash_test_cases_info = set()
+ current_test_case_info = None
+
def __init__(self, stream, descriptions, verbosity):
"""
:param stream File descriptor to store where to report test results.
self.descriptions = descriptions
self.verbosity = verbosity
self.result_string = None
- self.printer = TestCasePrinter()
- self.passed = 0
def addSuccess(self, test):
"""
:param test:
"""
- if hasattr(test, 'logger'):
- test.logger.debug("--- addSuccess() %s.%s(%s) called"
- % (test.__class__.__name__,
- test._testMethodName,
- test._testMethodDoc))
- self.passed += 1
+ 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 reason:
"""
- if hasattr(test, 'logger'):
- test.logger.debug("--- addSkip() %s.%s(%s) called, reason is %s"
- % (test.__class__.__name__,
- test._testMethodName,
- test._testMethodDoc,
- 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)
- def symlink_failed(self, test):
- logger = None
- if hasattr(test, 'logger'):
- logger = test.logger
- if hasattr(test, 'tempdir'):
+ self.send_result_through_pipe(test, SKIP)
+
+ def symlink_failed(self):
+ if self.current_test_case_info:
try:
failed_dir = os.getenv('VPP_TEST_FAILED_DIR')
- link_path = os.path.join(failed_dir, '%s-FAILED' %
- os.path.basename(test.tempdir))
- if logger:
- logger.debug("creating a link to the failed test")
- logger.debug("os.symlink(%s, %s)" %
- (test.tempdir, link_path))
+ 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 logger:
- logger.debug('symlink already exists')
+ if self.current_test_case_info.logger:
+ self.current_test_case_info.logger.debug(
+ 'symlink already exists')
else:
- os.symlink(test.tempdir, link_path)
+ os.symlink(self.current_test_case_info.tempdir, link_path)
except Exception as e:
- if logger:
- logger.error(e)
+ if self.current_test_case_info.logger:
+ self.current_test_case_info.logger.error(e)
- def send_results_through_pipe(self):
- if hasattr(self, 'test_framework_results_pipe'):
- pipe = self.test_framework_results_pipe
+ 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(self)
+ 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 = "'{}' ({})".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):
"""
:param err: error message
"""
- if hasattr(test, 'logger'):
- test.logger.debug("--- addFailure() %s.%s(%s) called, err is %s"
- % (test.__class__.__name__,
- test._testMethodName,
- test._testMethodDoc, err))
- test.logger.debug("formatted exception is:\n%s" %
- "".join(format_exception(*err)))
- unittest.TestResult.addFailure(self, test, err)
- if hasattr(test, 'tempdir'):
- self.result_string = colorize("FAIL", RED) + \
- ' [ temp dir used by test case: ' + test.tempdir + ' ]'
- self.symlink_failed(test)
- else:
- self.result_string = colorize("FAIL", RED) + ' [no temp dir]'
+ self.add_error(test, err, unittest.TestResult.addFailure, FAIL)
def addError(self, test, err):
"""
:param err: error message
"""
- if hasattr(test, 'logger'):
- test.logger.debug("--- addError() %s.%s(%s) called, err is %s"
- % (test.__class__.__name__,
- test._testMethodName,
- test._testMethodDoc, err))
- test.logger.debug("formatted exception is:\n%s" %
- "".join(format_exception(*err)))
- unittest.TestResult.addError(self, test, err)
- if hasattr(test, 'tempdir'):
- self.result_string = colorize("ERROR", RED) + \
- ' [ temp dir used by test case: ' + test.tempdir + ' ]'
- self.symlink_failed(test)
- else:
- self.result_string = colorize("ERROR", RED) + ' [no temp dir]'
+ self.add_error(test, err, unittest.TestResult.addError, ERROR)
def getDescription(self, test):
"""
:param test:
"""
- self.printer.print_test_case_heading_if_first_time(test)
+ test.print_header(test.__class__)
+
unittest.TestResult.startTest(self, test)
if self.verbosity > 0:
self.stream.writeln(
else:
self.stream.writeln("%-73s%s" % (self.getDescription(test),
self.result_string))
- self.send_results_through_pipe()
+
+ self.send_result_through_pipe(test, TEST_RUN)
def printErrors(self):
"""
return VppTestResult
def __init__(self, keep_alive_pipe=None, descriptions=True, verbosity=1,
- results_pipe=None, failfast=False, buffer=False,
+ result_pipe=None, failfast=False, buffer=False,
resultclass=None):
+
# 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)
- reporter = KeepAliveReporter()
- reporter.pipe = keep_alive_pipe
+ KeepAliveReporter.pipe = keep_alive_pipe
- VppTestResult.test_framework_results_pipe = results_pipe
+ VppTestResult.test_framework_result_pipe = result_pipe
def run(self, test):
"""
Code Review / vpp.git / blobdiff