else:
import subprocess
+# Python2/3 compatible
+try:
+ input = raw_input
+except NameError:
+ pass
+
PASS = 0
FAIL = 1
ERROR = 2
# flag will take care of properly terminating the loop
-def is_skip_aarch64_set():
+def _is_skip_aarch64_set():
return os.getenv('SKIP_AARCH64', 'n').lower() in ('yes', 'y', '1')
+is_skip_aarch64_set = _is_skip_aarch64_set()
+
-def is_platform_aarch64():
+def _is_platform_aarch64():
return platform.machine() == 'aarch64'
+is_platform_aarch64 = _is_platform_aarch64()
+
-def running_extended_tests():
+def _running_extended_tests():
s = os.getenv("EXTENDED_TESTS", "n")
return True if s.lower() in ("y", "yes", "1") else False
+running_extended_tests = _running_extended_tests()
-def running_on_centos():
+
+def _running_on_centos():
os_id = os.getenv("OS_ID", "")
return True if "centos" in os_id.lower() else False
+running_on_centos = _running_on_centos
+
class KeepAliveReporter(object):
"""
return random.choice(tuple(min_usage_set))
- @classmethod
- def print_header(cls):
- if not hasattr(cls, '_header_printed'):
- print(double_line_delim)
- print(colorize(getdoc(cls).splitlines()[0], GREEN))
- print(double_line_delim)
- cls._header_printed = True
-
@classmethod
def setUpConstants(cls):
""" Set-up the test case class based on environment variables """
print("Now is the time to attach a gdb by running the above "
"command and set up breakpoints etc.")
print(single_line_delim)
- raw_input("Press ENTER to continue running the testcase...")
+ input("Press ENTER to continue running the testcase...")
@classmethod
def run_vpp(cls):
"""
gc.collect() # run garbage collection first
random.seed()
- cls.print_header()
cls.logger = get_logger(cls.__name__)
if hasattr(cls, 'parallel_handler'):
cls.logger.addHandler(cls.parallel_handler)
print(double_line_delim)
print("VPP or GDB server is still running")
print(single_line_delim)
- raw_input("When done debugging, press ENTER to kill the "
- "process and finish running the testcase...")
+ input("When done debugging, press ENTER to kill the "
+ "process and finish running the testcase...")
# first signal that we want to stop the pump thread, then wake it up
if hasattr(cls, 'pump_thread_stop_flag'):
(self.__class__.__name__, self._testMethodName,
self._testMethodDoc))
if not self.vpp_dead:
- self.logger.debug(self.vapi.cli("show trace"))
+ self.logger.debug(self.vapi.cli("show trace max 1000"))
self.logger.info(self.vapi.ppcli("show interface"))
self.logger.info(self.vapi.ppcli("show hardware"))
self.logger.info(self.statistics.set_errors_str())
cls.logger.debug("Removing zombie capture %s" % cap_name)
cls.vapi.cli('packet-generator delete %s' % cap_name)
- cls.vapi.cli("trace add pg-input 50") # 50 is maximum
+ cls.vapi.cli("trace add pg-input 1000")
cls.vapi.cli('packet-generator enable')
cls._zombie_captures = cls._captures
cls._captures = []
input.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- if isinstance(object, (list,)):
- rx = []
- for o in output:
- rx.append(output.get_capture(len(pkts)))
- else:
- rx = output.get_capture(len(pkts))
+ rx = output.get_capture(len(pkts))
return rx
def send_and_expect_only(self, input, pkts, output, timeout=None):
input.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- if isinstance(object, (list,)):
- outputs = output
- rx = []
- for o in outputs:
- rx.append(output.get_capture(len(pkts)))
- else:
- rx = output.get_capture(len(pkts))
- outputs = [output]
+ rx = output.get_capture(len(pkts))
+ outputs = [output]
if not timeout:
timeout = 1
for i in self.pg_interfaces:
return rx
+ def runTest(self):
+ """ unittest calls runTest when TestCase is instantiated without a
+ test case. Use case: Writing unittests against VppTestCase"""
+ pass
+
def get_testcase_doc_name(test):
return getdoc(test.__class__).splitlines()[0]
test case descriptions.
:param verbosity Integer variable to store required verbosity level.
"""
- unittest.TestResult.__init__(self, stream, descriptions, verbosity)
+ super(VppTestResult, self).__init__(stream, descriptions, verbosity)
self.stream = stream
self.descriptions = descriptions
self.verbosity = verbosity
:param test:
"""
- test.print_header()
+
+ def print_header(test):
+ if not hasattr(test.__class__, '_header_printed'):
+ print(double_line_delim)
+ print(colorize(getdoc(test).splitlines()[0], GREEN))
+ print(double_line_delim)
+ test.__class__._header_printed = True
+
+ print_header(test)
unittest.TestResult.startTest(self, test)
if self.verbosity > 0:
def __init__(self, keep_alive_pipe=None, descriptions=True, verbosity=1,
result_pipe=None, failfast=False, buffer=False,
- resultclass=None, print_summary=True):
+ resultclass=None, print_summary=True, **kwargs):
# ignore stream setting here, use hard-coded stdout to be in sync
# with prints from VppTestCase methods ...
super(VppTestRunner, self).__init__(sys.stdout, descriptions,
verbosity, failfast, buffer,
- resultclass)
+ resultclass, **kwargs)
KeepAliveReporter.pipe = keep_alive_pipe
self.orig_stream = self.stream
self.logger.info(err)
self.logger.info(single_line_delim)
self.result = self.process.returncode
+
+if __name__ == '__main__':
+ pass
Code Review / vpp.git / blobdiff