import unittest
import tempfile
import time
-import resource
import faulthandler
+import random
+import copy
+import psutil
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 vpp_papi.vpp_stats import VPPStats
+from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
+ getLogger, colorize
from vpp_object import VppObjectRegistry
+from util import ppp
+from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
+from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
+from scapy.layers.inet6 import ICMPv6EchoReply
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 pump_output(testclass):
""" pump output from vpp stdout/stderr to proper queues """
- while not testclass.pump_thread_stop_flag.wait(0):
+ stdout_fragment = ""
+ stderr_fragment = ""
+ while not testclass.pump_thread_stop_flag.is_set():
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)
+ read = os.read(testclass.vpp.stdout.fileno(), 102400)
+ if len(read) > 0:
+ split = read.splitlines(True)
+ if len(stdout_fragment) > 0:
+ split[0] = "%s%s" % (stdout_fragment, split[0])
+ if len(split) > 0 and split[-1].endswith("\n"):
+ limit = None
+ else:
+ limit = -1
+ stdout_fragment = split[-1]
+ testclass.vpp_stdout_deque.extend(split[:limit])
+ if not testclass.cache_vpp_output:
+ for line in split[:limit]:
+ testclass.logger.debug(
+ "VPP STDOUT: %s" % line.rstrip("\n"))
if testclass.vpp.stderr.fileno() in readable:
- read = os.read(testclass.vpp.stderr.fileno(), 1024)
- testclass.vpp_stderr_deque.append(read)
+ read = os.read(testclass.vpp.stderr.fileno(), 102400)
+ if len(read) > 0:
+ split = read.splitlines(True)
+ if len(stderr_fragment) > 0:
+ split[0] = "%s%s" % (stderr_fragment, split[0])
+ if len(split) > 0 and split[-1].endswith("\n"):
+ limit = None
+ else:
+ limit = -1
+ stderr_fragment = split[-1]
+ testclass.vpp_stderr_deque.extend(split[:limit])
+ if not testclass.cache_vpp_output:
+ for line in split[:limit]:
+ testclass.logger.debug(
+ "VPP STDERR: %s" % line.rstrip("\n"))
# 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
+ 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):
else:
raise Exception("Unrecognized DEBUG option: '%s'" % d)
+ @classmethod
+ def get_least_used_cpu(self):
+ cpu_usage_list = [set(range(psutil.cpu_count()))]
+ vpp_processes = [p for p in psutil.process_iter(attrs=['pid', 'name'])
+ if 'vpp_main' == p.info['name']]
+ for vpp_process in vpp_processes:
+ for cpu_usage_set in cpu_usage_list:
+ try:
+ cpu_num = vpp_process.cpu_num()
+ if cpu_num in cpu_usage_set:
+ cpu_usage_set_index = cpu_usage_list.index(
+ cpu_usage_set)
+ if cpu_usage_set_index == len(cpu_usage_list) - 1:
+ cpu_usage_list.append({cpu_num})
+ else:
+ cpu_usage_list[cpu_usage_set_index + 1].add(
+ cpu_num)
+ cpu_usage_set.remove(cpu_num)
+ break
+ except psutil.NoSuchProcess:
+ pass
+
+ for cpu_usage_set in cpu_usage_list:
+ if len(cpu_usage_set) > 0:
+ min_usage_set = cpu_usage_set
+ break
+
+ return random.choice(tuple(min_usage_set))
+
@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
+ 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"
+
+ cpu_core_number = cls.get_least_used_cpu()
+
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), "}",
+ "stats", "{", "socket-name",
+ cls.tempdir + "/stats.sock", "}",
"plugins", "{", "plugin", "dpdk_plugin.so", "{",
- "disable", "}", "}"]
+ "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)
Remove shared memory files, start vpp and connect the vpp-api
"""
gc.collect() # run garbage collection first
- cls.logger = getLogger(cls.__name__)
+ random.seed()
+ if not hasattr(cls, 'logger'):
+ cls.logger = getLogger(cls.__name__)
+ else:
+ cls.logger.name = cls.__name__
cls.tempdir = tempfile.mkdtemp(
prefix='vpp-unittest-%s-' % cls.__name__)
cls.file_handler = FileHandler("%s/log.txt" % cls.tempdir)
datefmt="%H:%M:%S"))
cls.file_handler.setLevel(DEBUG)
cls.logger.addHandler(cls.file_handler)
- cls.shm_prefix = cls.tempdir.split("/")[-1]
+ cls.shm_prefix = os.path.basename(cls.tempdir)
os.chdir(cls.tempdir)
cls.logger.info("Temporary dir is %s, shm prefix is %s",
cls.tempdir, cls.shm_prefix)
hook = PollHook(cls)
cls.vapi.register_hook(hook)
cls.sleep(0.1, "after vpp startup, before initial poll")
+ cls.statistics = VPPStats(socketname=cls.tempdir+'/stats.sock')
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):
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()
+ # first signal that we want to stop the pump thread, then wake it up
+ if hasattr(cls, 'pump_thread_stop_flag'):
+ cls.pump_thread_stop_flag.set()
+ if hasattr(cls, 'pump_thread_wakeup_pipe'):
+ os.write(cls.pump_thread_wakeup_pipe[1], 'ding dong wake up')
if hasattr(cls, 'pump_thread'):
cls.logger.debug("Waiting for pump thread to stop")
cls.pump_thread.join()
""" 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 """
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)
# Save/Dump VPP api trace log
api_trace = "vpp_api_trace.%s.log" % self._testMethodName
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" %
+ self.logger.info(self.vapi.ppcli("api trace custom-dump %s" %
vpp_api_trace_log))
else:
self.registry.unregister_all(self.logger)
type(self).test_instance = self
@classmethod
- def pg_enable_capture(cls, interfaces):
+ def pg_enable_capture(cls, interfaces=None):
"""
Enable capture on packet-generator interfaces
- :param interfaces: iterable interface indexes
+ :param interfaces: iterable interface indexes (if None,
+ use self.pg_interfaces)
"""
+ if interfaces is None:
+ interfaces = cls.pg_interfaces
for i in interfaces:
i.enable_capture()
return result
@classmethod
- def create_loopback_interfaces(cls, interfaces):
+ def create_loopback_interfaces(cls, count):
"""
Create loopback interfaces.
- :param interfaces: iterable indexes of the interfaces.
+ :param count: number of interfaces created.
:returns: List of created interfaces.
"""
- result = []
- for i in interfaces:
- intf = VppLoInterface(cls, i)
+ result = [VppLoInterface(cls) for i in range(count)]
+ for intf in result:
setattr(cls, intf.name, intf)
- result.append(intf)
cls.lo_interfaces = result
return result
@staticmethod
- def extend_packet(packet, size):
+ def extend_packet(packet, size, padding=' '):
"""
- Extend packet to given size by padding with spaces
+ Extend packet to given size by padding with spaces or custom padding
NOTE: Currently works only when Raw layer is present.
:param packet: packet
:param size: target size
+ :param padding: padding used to extend the payload
"""
packet_len = len(packet) + 4
extend = size - packet_len
if extend > 0:
- packet[Raw].load += ' ' * extend
+ num = (extend / len(padding)) + 1
+ packet[Raw].load += (padding * num)[:extend]
@classmethod
def reset_packet_infos(cls):
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)
name, real_value, expected_min, expected_max)
self.assertTrue(expected_min <= real_value <= expected_max, msg)
+ def assert_packet_checksums_valid(self, packet,
+ ignore_zero_udp_checksums=True):
+ received = packet.__class__(str(packet))
+ self.logger.debug(
+ ppp("Verifying packet checksums for packet:", received))
+ udp_layers = ['UDP', 'UDPerror']
+ checksum_fields = ['cksum', 'chksum']
+ checksums = []
+ counter = 0
+ temp = received.__class__(str(received))
+ while True:
+ layer = temp.getlayer(counter)
+ if layer:
+ for cf in checksum_fields:
+ if hasattr(layer, cf):
+ if ignore_zero_udp_checksums and \
+ 0 == getattr(layer, cf) and \
+ layer.name in udp_layers:
+ continue
+ delattr(layer, cf)
+ checksums.append((counter, cf))
+ else:
+ break
+ counter = counter + 1
+ if 0 == len(checksums):
+ return
+ temp = temp.__class__(str(temp))
+ for layer, cf in checksums:
+ calc_sum = getattr(temp[layer], cf)
+ self.assert_equal(
+ getattr(received[layer], cf), calc_sum,
+ "packet checksum on layer #%d: %s" % (layer, temp[layer].name))
+ self.logger.debug(
+ "Checksum field `%s` on `%s` layer has correct value `%s`" %
+ (cf, temp[layer].name, calc_sum))
+
+ def assert_checksum_valid(self, received_packet, layer,
+ field_name='chksum',
+ ignore_zero_checksum=False):
+ """ Check checksum of received packet on given layer """
+ received_packet_checksum = getattr(received_packet[layer], field_name)
+ if ignore_zero_checksum and 0 == received_packet_checksum:
+ return
+ recalculated = received_packet.__class__(str(received_packet))
+ delattr(recalculated[layer], field_name)
+ recalculated = recalculated.__class__(str(recalculated))
+ self.assert_equal(received_packet_checksum,
+ getattr(recalculated[layer], field_name),
+ "packet checksum on layer: %s" % layer)
+
+ def assert_ip_checksum_valid(self, received_packet,
+ ignore_zero_checksum=False):
+ self.assert_checksum_valid(received_packet, 'IP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_tcp_checksum_valid(self, received_packet,
+ ignore_zero_checksum=False):
+ self.assert_checksum_valid(received_packet, 'TCP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_udp_checksum_valid(self, received_packet,
+ ignore_zero_checksum=True):
+ self.assert_checksum_valid(received_packet, 'UDP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_embedded_icmp_checksum_valid(self, received_packet):
+ if received_packet.haslayer(IPerror):
+ self.assert_checksum_valid(received_packet, 'IPerror')
+ if received_packet.haslayer(TCPerror):
+ self.assert_checksum_valid(received_packet, 'TCPerror')
+ if received_packet.haslayer(UDPerror):
+ self.assert_checksum_valid(received_packet, 'UDPerror',
+ ignore_zero_checksum=True)
+ if received_packet.haslayer(ICMPerror):
+ self.assert_checksum_valid(received_packet, 'ICMPerror')
+
+ def assert_icmp_checksum_valid(self, received_packet):
+ self.assert_checksum_valid(received_packet, 'ICMP')
+ self.assert_embedded_icmp_checksum_valid(received_packet)
+
+ def assert_icmpv6_checksum_valid(self, pkt):
+ if pkt.haslayer(ICMPv6DestUnreach):
+ self.assert_checksum_valid(pkt, 'ICMPv6DestUnreach', 'cksum')
+ self.assert_embedded_icmp_checksum_valid(pkt)
+ if pkt.haslayer(ICMPv6EchoRequest):
+ self.assert_checksum_valid(pkt, 'ICMPv6EchoRequest', 'cksum')
+ if pkt.haslayer(ICMPv6EchoReply):
+ self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
+
@classmethod
def sleep(cls, timeout, remark=None):
if hasattr(cls, 'logger'):
"Finished sleep (%s) - slept %ss (wanted %ss)" % (
remark, after - before, timeout))
+ def send_and_assert_no_replies(self, intf, pkts, remark="", timeout=None):
+ self.vapi.cli("clear trace")
+ intf.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ if not timeout:
+ timeout = 1
+ for i in self.pg_interfaces:
+ i.get_capture(0, timeout=timeout)
+ i.assert_nothing_captured(remark=remark)
+ timeout = 0.1
+
+ def send_and_expect(self, input, pkts, output):
+ self.vapi.cli("clear trace")
+ input.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ rx = output.get_capture(len(pkts))
+ return rx
+
+
+def get_testcase_doc_name(test):
+ return getdoc(test.__class__).splitlines()[0]
+
+
+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
+ else:
+ return str(test)
+
class TestCasePrinter(object):
_shared_state = {}
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(colorize(get_testcase_doc_name(case), GREEN))
print(double_line_delim)
self._test_case_set.add(case.__class__)
self.verbosity = verbosity
self.result_string = None
self.printer = TestCasePrinter()
+ self.passed = []
def addSuccess(self, test):
"""
% (test.__class__.__name__,
test._testMethodName,
test._testMethodDoc))
+ self.passed.append(test.id())
unittest.TestResult.addSuccess(self, test)
self.result_string = colorize("OK", GREEN)
if hasattr(test, 'tempdir'):
try:
failed_dir = os.getenv('VPP_TEST_FAILED_DIR')
- link_path = '%s/%s-FAILED' % (failed_dir,
- test.tempdir.split("/")[-1])
+ 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))
- os.symlink(test.tempdir, link_path)
+ if os.path.exists(link_path):
+ if logger:
+ logger.debug('symlink already exists')
+ else:
+ os.symlink(test.tempdir, link_path)
+
except Exception as e:
if logger:
logger.error(e)
- def send_failure_through_pipe(self, test):
- if hasattr(self, 'test_framework_failed_pipe'):
- pipe = self.test_framework_failed_pipe
+ def send_results_through_pipe(self):
+ if hasattr(self, 'test_framework_results_pipe'):
+ pipe = self.test_framework_results_pipe
if pipe:
- pipe.send(test.__class__)
+ pipe.send(self)
def addFailure(self, test, err):
"""
else:
self.result_string = colorize("FAIL", RED) + ' [no temp dir]'
- self.send_failure_through_pipe(test)
-
def addError(self, test, err):
"""
Record a test error result
else:
self.result_string = colorize("ERROR", RED) + ' [no temp dir]'
- self.send_failure_through_pipe(test)
-
def getDescription(self, test):
"""
Get test description
:returns: test description
"""
- # TODO: if none print warning not raise exception
- short_description = test.shortDescription()
- if self.descriptions and short_description:
- return short_description
- else:
- return str(test)
+ return get_test_description(self.descriptions, test)
def startTest(self, test):
"""
def stopTest(self, test):
"""
- Stop a test
+ Called when the given test has been run
:param test:
else:
self.stream.writeln("%-73s%s" % (self.getDescription(test),
self.result_string))
+ self.send_results_through_pipe()
def printErrors(self):
"""
self.stream.writeln("%s" % err)
-class Filter_by_test_option:
- def __init__(self, filter_file_name, filter_class_name, filter_func_name):
- self.filter_file_name = filter_file_name
- self.filter_class_name = filter_class_name
- self.filter_func_name = filter_func_name
-
- def __call__(self, file_name, class_name, func_name):
- if self.filter_file_name and file_name != self.filter_file_name:
- return False
- if self.filter_class_name and class_name != self.filter_class_name:
- return False
- if self.filter_func_name and func_name != self.filter_func_name:
- return False
- return True
-
-
class VppTestRunner(unittest.TextTestRunner):
"""
A basic test runner implementation which prints results to standard error.
"""Class maintaining the results of the tests"""
return VppTestResult
- def __init__(self, keep_alive_pipe=None, failed_pipe=None,
- stream=sys.stderr, descriptions=True,
- verbosity=1, failfast=False, buffer=False, resultclass=None):
+ def __init__(self, keep_alive_pipe=None, descriptions=True, verbosity=1,
+ results_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,
resultclass)
reporter = KeepAliveReporter()
reporter.pipe = keep_alive_pipe
- # this is super-ugly, but very simple to implement and works as long
- # as we run only one test at the same time
- VppTestResult.test_framework_failed_pipe = failed_pipe
- test_option = "TEST"
-
- def parse_test_option(self):
- try:
- f = os.getenv(self.test_option)
- except:
- f = None
- filter_file_name = None
- filter_class_name = None
- filter_func_name = None
- if f:
- if '.' in f:
- parts = f.split('.')
- if len(parts) > 3:
- raise Exception("Unrecognized %s option: %s" %
- (self.test_option, f))
- if len(parts) > 2:
- if parts[2] not in ('*', ''):
- filter_func_name = parts[2]
- if parts[1] not in ('*', ''):
- filter_class_name = parts[1]
- if parts[0] not in ('*', ''):
- if parts[0].startswith('test_'):
- filter_file_name = parts[0]
- else:
- filter_file_name = 'test_%s' % parts[0]
- else:
- if f.startswith('test_'):
- filter_file_name = f
- else:
- filter_file_name = 'test_%s' % f
- return filter_file_name, filter_class_name, filter_func_name
-
- @staticmethod
- def filter_tests(tests, filter_cb):
- result = unittest.suite.TestSuite()
- for t in tests:
- if isinstance(t, unittest.suite.TestSuite):
- # this is a bunch of tests, recursively filter...
- x = filter_tests(t, filter_cb)
- if x.countTestCases() > 0:
- result.addTest(x)
- elif isinstance(t, unittest.TestCase):
- # this is a single test
- parts = t.id().split('.')
- # t.id() for common cases like this:
- # test_classifier.TestClassifier.test_acl_ip
- # apply filtering only if it is so
- if len(parts) == 3:
- if not filter_cb(parts[0], parts[1], parts[2]):
- continue
- result.addTest(t)
- else:
- # unexpected object, don't touch it
- result.addTest(t)
- return result
+ VppTestResult.test_framework_results_pipe = results_pipe
def run(self, test):
"""
"""
faulthandler.enable() # emit stack trace to stderr if killed by signal
- 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" % (
- filter_file, filter_class, filter_func))
- filter_cb = Filter_by_test_option(
- filter_file, filter_class, filter_func)
- filtered = self.filter_tests(test, filter_cb)
- print("%s out of %s tests match specified filters" % (
- filtered.countTestCases(), test.countTestCases()))
- if not running_extended_tests():
- print("Not running extended tests (some tests will be skipped)")
- return super(VppTestRunner, self).run(filtered)
+
+ result = super(VppTestRunner, self).run(test)
+ return result
+
+
+class Worker(Thread):
+ 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,
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ out, err = self.process.communicate()
+ self.logger.debug("Finished running `%s'" % executable)
+ self.logger.info("Return code is `%s'" % self.process.returncode)
+ self.logger.info(single_line_delim)
+ self.logger.info("Executable `%s' wrote to stdout:" % executable)
+ self.logger.info(single_line_delim)
+ self.logger.info(out)
+ self.logger.info(single_line_delim)
+ self.logger.info("Executable `%s' wrote to stderr:" % executable)
+ self.logger.info(single_line_delim)
+ self.logger.info(err)
+ self.logger.info(single_line_delim)
+ self.result = self.process.returncode
Code Review / vpp.git / blobdiff