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 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.
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)
try:
cls.vapi.connect()
except:
+ try:
+ cls.vapi.disconnect()
+ except:
+ pass
if cls.debug_gdbserver:
print(colorize("You're running VPP inside gdbserver but "
"VPP-API connection failed, did you forget "
cls.quit()
except:
pass
- raise t, v, tb
+ raise (t, v, tb)
@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 """
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