from __future__ import print_function
import gc
+import logging
import sys
import os
import select
import signal
+import subprocess
import unittest
import tempfile
import time
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
- import subprocess32 as subprocess
-else:
- import subprocess
-
-# Python2/3 compatible
-try:
- input = raw_input
-except NameError:
- pass
+logger = logging.getLogger(__name__)
+
+# Set up an empty logger for the testcase that can be overridden as necessary
+null_logger = logging.getLogger('VppTestCase')
+null_logger.addHandler(logging.NullHandler())
PASS = 0
FAIL = 1
extra_vpp_punt_config = []
extra_vpp_plugin_config = []
+ logger = null_logger
vapi_response_timeout = 5
@property
else:
return 0
+ @classmethod
+ def force_solo(cls):
+ """ if the test case class is timing-sensitive - return true """
+ return False
+
@classmethod
def instance(cls):
"""Return the instance of this testcase"""
"plugins",
"{", "plugin", "dpdk_plugin.so", "{", "disable",
"}", "plugin", "rdma_plugin.so", "{", "disable",
+ "}", "plugin", "lisp_unittest_plugin.so", "{",
+ "enable",
"}", "plugin", "unittest_plugin.so", "{", "enable",
"}"] + cls.extra_vpp_plugin_config + ["}", ]
time.sleep(0)
return
- if hasattr(cls, 'logger'):
- cls.logger.debug("Starting sleep for %es (%s)", timeout, remark)
+ cls.logger.debug("Starting sleep for %es (%s)", timeout, remark)
before = time.time()
time.sleep(timeout)
after = time.time()
- if hasattr(cls, 'logger') and after - before > 2 * timeout:
+ if after - before > 2 * timeout:
cls.logger.error("unexpected self.sleep() result - "
"slept for %es instead of ~%es!",
after - before, timeout)
- if hasattr(cls, 'logger'):
- cls.logger.debug(
+
+ cls.logger.debug(
"Finished sleep (%s) - slept %es (wanted %es)",
remark, after - before, timeout)
failed_dir,
'%s-FAILED' %
os.path.basename(self.current_test_case_info.tempdir))
- if self.current_test_case_info.logger:
- self.current_test_case_info.logger.debug(
+
+ self.current_test_case_info.logger.debug(
"creating a link to the failed test")
- self.current_test_case_info.logger.debug(
+ self.current_test_case_info.logger.debug(
"os.symlink(%s, %s)" %
(self.current_test_case_info.tempdir, link_path))
if os.path.exists(link_path):
- if self.current_test_case_info.logger:
- self.current_test_case_info.logger.debug(
+ self.current_test_case_info.logger.debug(
'symlink already exists')
else:
os.symlink(self.current_test_case_info.tempdir, link_path)
except Exception as e:
- if self.current_test_case_info.logger:
- self.current_test_case_info.logger.error(e)
+ self.current_test_case_info.logger.error(e)
def send_result_through_pipe(self, test, result):
if hasattr(self, 'test_framework_result_pipe'):
"""
def print_header(test):
+ test_doc = getdoc(test)
+ if not test_doc:
+ raise Exception("No doc string for test '%s'" % test.id())
+ test_title = test_doc.splitlines()[0]
+ test_title_colored = colorize(test_title, GREEN)
+ if test.force_solo():
+ # long live PEP-8 and 80 char width limitation...
+ c = YELLOW
+ test_title_colored = colorize("SOLO RUN: " + test_title, c)
+
if not hasattr(test.__class__, '_header_printed'):
print(double_line_delim)
- print(colorize(getdoc(test).splitlines()[0], GREEN))
+ print(test_title_colored)
print(double_line_delim)
test.__class__._header_printed = True