#!/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.
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
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
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, "}",
+ 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, "}"]
if cls.plugin_path is not None:
cls.vpp_cmdline.extend(["plugin_path", cls.plugin_path])
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]
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'):
def tearDown(self):
""" Show various debug prints after each test """
+ self.logger.debug("--- tearDown() for %s.%s(%s) called ---" %
+ (self.__class__.__name__, self._testMethodName,
+ 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 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))
def setUp(self):
""" Clear trace before running each test"""
+ self.logger.debug("--- setUp() for %s.%s(%s) called ---" %
+ (self.__class__.__name__, self._testMethodName,
+ 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)
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)
+ @classmethod
+ def sleep(cls, timeout, remark=None):
+ if hasattr(cls, 'logger'):
+ cls.logger.debug("Sleeping for %ss (%s)" % (timeout, remark))
+ time.sleep(timeout)
+
class VppTestResult(unittest.TestResult):
"""
:param test:
"""
+ if hasattr(test, 'logger'):
+ test.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)
: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))
unittest.TestResult.addSkip(self, test, reason)
self.result_string = colorize("SKIP", YELLOW)
: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) + \
: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) + \
: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