#!/usr/bin/env python
-import subprocess
+from __future__ import print_function
+import gc
+import sys
+import os
+import select
import unittest
import tempfile
import time
import resource
from collections import deque
-from threading import Thread
+from threading import Thread, Event
from inspect import getdoc
from traceback import format_exception
+from logging import FileHandler, DEBUG, Formatter
+from scapy.packet import Raw
from hook import StepHook, PollHook
from vpp_pg_interface import VppPGInterface
from vpp_sub_interface import VppSubInterface
from vpp_lo_interface import VppLoInterface
from vpp_papi_provider import VppPapiProvider
-from scapy.packet import Raw
-from logging import FileHandler, DEBUG
from log import *
from vpp_object import VppObjectRegistry
+if os.name == 'posix' and sys.version_info[0] < 3:
+ # using subprocess32 is recommended by python official documentation
+ # @ https://docs.python.org/2/library/subprocess.html
+ import subprocess32 as subprocess
+else:
+ import subprocess
"""
Test framework module.
#: Store the index of the destination packet generator interface
#: of the packet.
dst = -1
+ #: Store expected ip version
+ ip = -1
+ #: Store expected upper protocol
+ proto = -1
#: Store the copy of the former packet.
data = None
return index and src and dst and data
-def pump_output(out, deque):
- for line in iter(out.readline, b''):
- deque.append(line)
+def pump_output(testclass):
+ """ pump output from vpp stdout/stderr to proper queues """
+ while not testclass.pump_thread_stop_flag.wait(0):
+ readable = select.select([testclass.vpp.stdout.fileno(),
+ testclass.vpp.stderr.fileno(),
+ testclass.pump_thread_wakeup_pipe[0]],
+ [], [])[0]
+ if testclass.vpp.stdout.fileno() in readable:
+ read = os.read(testclass.vpp.stdout.fileno(), 1024)
+ testclass.vpp_stdout_deque.append(read)
+ if testclass.vpp.stderr.fileno() in readable:
+ read = os.read(testclass.vpp.stderr.fileno(), 1024)
+ testclass.vpp_stderr_deque.append(read)
+ # ignoring the dummy pipe here intentionally - the flag will take care
+ # of properly terminating the loop
+
+
+def running_extended_tests():
+ try:
+ s = os.getenv("EXTENDED_TESTS")
+ return True if s.lower() in ("y", "yes", "1") else False
+ except:
+ return False
+ return False
class VppTestCase(unittest.TestCase):
return
dl = d.lower()
if dl == "core":
- if resource.getrlimit(resource.RLIMIT_CORE)[0] <= 0:
- # give a heads up if this is actually useless
- print(colorize("WARNING: core size limit is set 0, core files "
- "will NOT be created", RED))
cls.debug_core = True
elif dl == "gdb":
cls.debug_gdb = True
cls.set_debug_flags(d)
cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp")
cls.plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH')
+ cls.extern_plugin_path = os.getenv('EXTERN_PLUGINS')
+ plugin_path = None
+ if cls.plugin_path is not None:
+ if cls.extern_plugin_path is not None:
+ plugin_path = "%s:%s" % (
+ cls.plugin_path, cls.extern_plugin_path)
+ else:
+ plugin_path = cls.plugin_path
+ elif cls.extern_plugin_path is not None:
+ plugin_path = cls.extern_plugin_path
debug_cli = ""
if cls.step or cls.debug_gdb or cls.debug_gdbserver:
debug_cli = "cli-listen localhost:5002"
- cls.vpp_cmdline = [cls.vpp_bin,
- "unix", "{", "nodaemon", debug_cli, "}",
- "api-segment", "{", "prefix", cls.shm_prefix, "}"]
- if cls.plugin_path is not None:
- cls.vpp_cmdline.extend(["plugin_path", cls.plugin_path])
+ coredump_size = None
+ try:
+ size = os.getenv("COREDUMP_SIZE")
+ if size is not None:
+ coredump_size = "coredump-size %s" % size
+ except:
+ pass
+ if coredump_size is None:
+ coredump_size = "coredump-size unlimited"
+ cls.vpp_cmdline = [cls.vpp_bin, "unix",
+ "{", "nodaemon", debug_cli, coredump_size, "}",
+ "api-trace", "{", "on", "}",
+ "api-segment", "{", "prefix", cls.shm_prefix, "}",
+ "plugins", "{", "plugin", "dpdk_plugin.so", "{",
+ "disable", "}", "}"]
+ if plugin_path is not None:
+ cls.vpp_cmdline.extend(["plugin_path", plugin_path])
cls.logger.info("vpp_cmdline: %s" % cls.vpp_cmdline)
@classmethod
Perform class setup before running the testcase
Remove shared memory files, start vpp and connect the vpp-api
"""
+ gc.collect() # run garbage collection first
cls.logger = getLogger(cls.__name__)
cls.tempdir = tempfile.mkdtemp(
prefix='vpp-unittest-' + cls.__name__ + '-')
file_handler = FileHandler("%s/log.txt" % cls.tempdir)
+ file_handler.setFormatter(
+ Formatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
+ datefmt="%H:%M:%S"))
file_handler.setLevel(DEBUG)
cls.logger.addHandler(file_handler)
cls.shm_prefix = cls.tempdir.split("/")[-1]
cls.verbose = 0
cls.vpp_dead = False
cls.registry = VppObjectRegistry()
- print(double_line_delim)
- print(colorize(getdoc(cls).splitlines()[0], YELLOW))
- print(double_line_delim)
# need to catch exceptions here because if we raise, then the cleanup
# doesn't get called and we might end with a zombie vpp
try:
cls.run_vpp()
cls.vpp_stdout_deque = deque()
- cls.vpp_stdout_reader_thread = Thread(target=pump_output, args=(
- cls.vpp.stdout, cls.vpp_stdout_deque))
- cls.vpp_stdout_reader_thread.start()
cls.vpp_stderr_deque = deque()
- cls.vpp_stderr_reader_thread = Thread(target=pump_output, args=(
- cls.vpp.stderr, cls.vpp_stderr_deque))
- cls.vpp_stderr_reader_thread.start()
+ cls.pump_thread_stop_flag = Event()
+ cls.pump_thread_wakeup_pipe = os.pipe()
+ 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.step:
hook = StepHook(cls)
else:
hook = PollHook(cls)
cls.vapi.register_hook(hook)
- time.sleep(0.1)
+ cls.sleep(0.1, "after vpp startup, before initial poll")
hook.poll_vpp()
try:
cls.vapi.connect()
raw_input("When done debugging, press ENTER to kill the "
"process and finish running the testcase...")
+ os.write(cls.pump_thread_wakeup_pipe[1], 'ding dong wake up')
+ cls.pump_thread_stop_flag.set()
+ if hasattr(cls, 'pump_thread'):
+ cls.logger.debug("Waiting for pump thread to stop")
+ cls.pump_thread.join()
+ if hasattr(cls, 'vpp_stderr_reader_thread'):
+ cls.logger.debug("Waiting for stdderr pump to stop")
+ cls.vpp_stderr_reader_thread.join()
+
if hasattr(cls, 'vpp'):
if hasattr(cls, 'vapi'):
cls.vapi.disconnect()
+ del cls.vapi
cls.vpp.poll()
if cls.vpp.returncode is None:
+ cls.logger.debug("Sending TERM to vpp")
cls.vpp.terminate()
+ cls.logger.debug("Waiting for vpp to die")
+ cls.vpp.communicate()
del cls.vpp
if hasattr(cls, 'vpp_stdout_deque'):
self._testMethodDoc))
if not self.vpp_dead:
self.logger.debug(self.vapi.cli("show trace"))
- self.logger.info(self.vapi.ppcli("show int"))
+ self.logger.info(self.vapi.ppcli("show interfaces"))
self.logger.info(self.vapi.ppcli("show hardware"))
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show run"))
self.registry.remove_vpp_config(self.logger)
+ # Save/Dump VPP api trace log
+ api_trace = "vpp_api_trace.%s.log" % self._testMethodName
+ tmp_api_trace = "/tmp/%s" % api_trace
+ vpp_api_trace_log = "%s/%s" % (self.tempdir, api_trace)
+ self.logger.info(self.vapi.ppcli("api trace save %s" % api_trace))
+ self.logger.info("Moving %s to %s\n" % (tmp_api_trace,
+ vpp_api_trace_log))
+ os.rename(tmp_api_trace, vpp_api_trace_log)
+ self.logger.info(self.vapi.ppcli("api trace dump %s" %
+ vpp_api_trace_log))
+ else:
+ self.registry.unregister_all(self.logger)
def setUp(self):
""" Clear trace before running each test"""
self._testMethodDoc))
if self.vpp_dead:
raise Exception("VPP is dead when setting up the test")
- time.sleep(.1)
+ self.sleep(.1, "during setUp")
self.vpp_stdout_deque.append(
"--- test setUp() for %s.%s(%s) starts here ---\n" %
(self.__class__.__name__, self._testMethodName,
for stamp, cap_name in cls._zombie_captures:
wait = stamp + capture_ttl - now
if wait > 0:
- cls.logger.debug("Waiting for %ss before deleting capture %s",
- wait, cap_name)
- time.sleep(wait)
+ cls.sleep(wait, "before deleting capture %s" % cap_name)
now = time.time()
cls.logger.debug("Removing zombie capture %s" % cap_name)
cls.vapi.cli('packet-generator delete %s' % cap_name)
:returns: string containing serialized data from packet info
"""
- return "%d %d %d" % (info.index, info.src, info.dst)
+ return "%d %d %d %d %d" % (info.index, info.src, info.dst,
+ info.ip, info.proto)
@staticmethod
def payload_to_info(payload):
info.index = int(numbers[0])
info.src = int(numbers[1])
info.dst = int(numbers[2])
+ info.ip = int(numbers[3])
+ info.proto = int(numbers[4])
return info
def get_next_packet_info(self, info):
def assert_equal(self, real_value, expected_value, name_or_class=None):
if name_or_class is None:
- self.assertEqual(real_value, expected_value, msg)
+ self.assertEqual(real_value, expected_value)
return
try:
msg = "Invalid %s: %d('%s') does not match expected value %d('%s')"
name, real_value, expected_min, expected_max)
self.assertTrue(expected_min <= real_value <= expected_max, msg)
- def sleep(self, timeout):
- self.logger.debug("Sleeping for %ss" % timeout)
+ @classmethod
+ def sleep(cls, timeout, remark=None):
+ if hasattr(cls, 'logger'):
+ cls.logger.debug("Sleeping for %ss (%s)" % (timeout, remark))
time.sleep(timeout)
+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(getdoc(case.__class__).splitlines()[0], YELLOW))
+ print(double_line_delim)
+ self._test_case_set.add(case.__class__)
+
+
class VppTestResult(unittest.TestResult):
"""
@property result_string
self.descriptions = descriptions
self.verbosity = verbosity
self.result_string = None
+ self.printer = TestCasePrinter()
def addSuccess(self, test):
"""
:param test:
"""
+ self.printer.print_test_case_heading_if_first_time(test)
unittest.TestResult.startTest(self, test)
if self.verbosity > 0:
self.stream.writeln(
:param test:
"""
+ gc.disable() # disable garbage collection, we'll do that manually
print("Running tests using custom test runner") # debug message
filter_file, filter_class, filter_func = self.parse_test_option()
print("Active filters: file=%s, class=%s, function=%s" % (
Code Review / vpp.git / blobdiff