import unittest
import tempfile
import time
-import resource
import faulthandler
import random
+import copy
from collections import deque
from threading import Thread, Event
from inspect import getdoc, isclass
from traceback import format_exception
from logging import FileHandler, DEBUG, Formatter
from scapy.packet import Raw
-from hook import StepHook, PollHook
+from hook import StepHook, PollHook, VppDiedError
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 log import *
+from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
+ getLogger, colorize
from vpp_object import VppObjectRegistry
-from vpp_punt_socket import vpp_uds_socket_name
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
else:
import subprocess
+debug_framework = False
+if os.getenv('TEST_DEBUG', "0") == "1":
+ debug_framework = True
+ import debug_internal
+
+
"""
Test framework module.
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
+ s = os.getenv("EXTENDED_TESTS", "n")
+ return True if s.lower() in ("y", "yes", "1") else False
def running_on_centos():
- try:
- os_id = os.getenv("OS_ID")
- return True if "centos" in os_id.lower() else False
- except:
- return False
- return False
+ os_id = os.getenv("OS_ID", "")
+ return True if "centos" in os_id.lower() else False
class KeepAliveReporter(object):
@classmethod
def setUpConstants(cls):
""" Set-up the test case class based on environment variables """
- try:
- s = os.getenv("STEP")
- cls.step = True if s.lower() in ("y", "yes", "1") else False
- except:
- cls.step = False
- try:
- d = os.getenv("DEBUG")
- except:
- d = None
- try:
- c = os.getenv("CACHE_OUTPUT", "1")
- cls.cache_vpp_output = \
- False if c.lower() in ("n", "no", "0") else True
- except:
- cls.cache_vpp_output = True
+ s = os.getenv("STEP", "n")
+ cls.step = True if s.lower() in ("y", "yes", "1") else False
+ d = os.getenv("DEBUG", None)
+ c = os.getenv("CACHE_OUTPUT", "1")
+ cls.cache_vpp_output = False if c.lower() in ("n", "no", "0") else True
cls.set_debug_flags(d)
cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp")
cls.plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH')
if cls.step or cls.debug_gdb or cls.debug_gdbserver:
debug_cli = "cli-listen localhost:5002"
coredump_size = None
- try:
- size = os.getenv("COREDUMP_SIZE")
- if size is not None:
- coredump_size = "coredump-size %s" % size
- except:
- pass
+ size = os.getenv("COREDUMP_SIZE")
+ if size is not None:
+ coredump_size = "coredump-size %s" % size
if coredump_size is None:
coredump_size = "coredump-size unlimited"
cls.vpp_cmdline = [cls.vpp_bin, "unix",
coredump_size, "}", "api-trace", "{", "on", "}",
"api-segment", "{", "prefix", cls.shm_prefix, "}",
"plugins", "{", "plugin", "dpdk_plugin.so", "{",
- "disable", "}", "}",
- "punt", "{", "socket", cls.punt_socket_path, "}"]
+ "disable", "}", "}", ]
if plugin_path is not None:
cls.vpp_cmdline.extend(["plugin_path", plugin_path])
cls.logger.info("vpp_cmdline: %s" % cls.vpp_cmdline)
cls.file_handler.setLevel(DEBUG)
cls.logger.addHandler(cls.file_handler)
cls.shm_prefix = cls.tempdir.split("/")[-1]
- cls.punt_socket_path = '%s/%s' % (cls.tempdir, vpp_uds_socket_name)
os.chdir(cls.tempdir)
cls.logger.info("Temporary dir is %s, shm prefix is %s",
cls.tempdir, cls.shm_prefix)
cls.sleep(0.1, "after vpp startup, before initial poll")
try:
hook.poll_vpp()
- except:
+ except VppDiedError:
cls.vpp_startup_failed = True
cls.logger.critical(
"VPP died shortly after startup, check the"
raise
try:
cls.vapi.connect()
- except:
+ except Exception:
+ try:
+ cls.vapi.disconnect()
+ except Exception:
+ pass
if cls.debug_gdbserver:
print(colorize("You're running VPP inside gdbserver but "
"VPP-API connection failed, did you forget "
"to 'continue' VPP from within gdb?", RED))
raise
- except:
- t, v, tb = sys.exc_info()
+ except Exception:
try:
cls.quit()
- except:
+ except Exception:
pass
- raise t, v, tb
+ raise
@classmethod
def quit(cls):
""" Perform final cleanup after running all tests in this test-case """
cls.quit()
cls.file_handler.close()
+ cls.reset_packet_infos()
+ if debug_framework:
+ debug_internal.on_tear_down_class(cls)
def tearDown(self):
""" Show various debug prints after each test """
msg = msg % (getdoc(name_or_class).strip(),
real_value, str(name_or_class(real_value)),
expected_value, str(name_or_class(expected_value)))
- except:
+ except Exception:
msg = "Invalid %s: %s does not match expected value %s" % (
name_or_class, real_value, expected_value)
test_option = "TEST"
def parse_test_option(self):
- try:
- f = os.getenv(self.test_option)
- except:
- f = None
+ f = os.getenv(self.test_option, None)
filter_file_name = None
filter_class_name = None
filter_func_name = None
for t in tests:
if isinstance(t, unittest.suite.TestSuite):
# this is a bunch of tests, recursively filter...
- x = filter_tests(t, filter_cb)
+ x = VppTestRunner.filter_tests(t, filter_cb)
if x.countTestCases() > 0:
result.addTest(x)
elif isinstance(t, unittest.TestCase):
class Worker(Thread):
- def __init__(self, args, logger):
+ def __init__(self, args, logger, env={}):
self.logger = logger
self.args = args
self.result = None
+ self.env = copy.deepcopy(env)
super(Worker, self).__init__()
def run(self):
executable = self.args[0]
self.logger.debug("Running executable w/args `%s'" % self.args)
env = os.environ.copy()
+ env.update(self.env)
env["CK_LOG_FILE_NAME"] = "-"
self.process = subprocess.Popen(
self.args, shell=False, env=env, preexec_fn=os.setpgrp,
self.logger.info(single_line_delim)
self.logger.info("Executable `%s' wrote to stderr:" % executable)
self.logger.info(single_line_delim)
- self.logger.error(err)
+ self.logger.info(err)
self.logger.info(single_line_delim)
self.result = self.process.returncode