from vpp_pg_interface import VppPGInterface
from vpp_sub_interface import VppSubInterface
from vpp_lo_interface import VppLoInterface
+from vpp_bvi_interface import VppBviInterface
from vpp_papi_provider import VppPapiProvider
from vpp_papi.vpp_stats import VPPStats
from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
cls.set_debug_flags(d)
cls.vpp_bin = os.getenv('VPP_BIN', "vpp")
cls.plugin_path = os.getenv('VPP_PLUGIN_PATH')
+ cls.test_plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH')
cls.extern_plugin_path = os.getenv('EXTERN_PLUGINS')
plugin_path = None
if cls.plugin_path is not None:
coredump_size, "runtime-dir", cls.tempdir, "}",
"api-trace", "{", "on", "}", "api-segment", "{",
"prefix", cls.shm_prefix, "}", "cpu", "{",
- "main-core", str(cpu_core_number), "}", "statseg",
- "{", "socket-name", cls.stats_sock, "}", "plugins",
+ "main-core", str(cpu_core_number), "}",
+ "statseg", "{", "socket-name", cls.stats_sock, "}",
+ "socksvr", "{", "socket-name", cls.api_sock, "}",
+ "plugins",
"{", "plugin", "dpdk_plugin.so", "{", "disable",
+ "}", "plugin", "rdma_plugin.so", "{", "disable",
"}", "plugin", "unittest_plugin.so", "{", "enable",
"}"] + cls.extra_vpp_plugin_config + ["}", ]
if cls.extra_vpp_punt_config is not None:
cls.vpp_cmdline.extend(cls.extra_vpp_punt_config)
if plugin_path is not None:
cls.vpp_cmdline.extend(["plugin_path", plugin_path])
+ if cls.test_plugin_path is not None:
+ cls.vpp_cmdline.extend(["test_plugin_path", cls.test_plugin_path])
+
cls.logger.info("vpp_cmdline args: %s" % cls.vpp_cmdline)
cls.logger.info("vpp_cmdline: %s" % " ".join(cls.vpp_cmdline))
if not ok:
cls.logger.critical("Couldn't stat : {}".format(cls.stats_sock))
+ @classmethod
+ def wait_for_coredump(cls):
+ corefile = cls.tempdir + "/core"
+ if os.path.isfile(corefile):
+ cls.logger.error("Waiting for coredump to complete: %s", corefile)
+ curr_size = os.path.getsize(corefile)
+ deadline = time.time() + 60
+ ok = False
+ while time.time() < deadline:
+ cls.sleep(1)
+ size = curr_size
+ curr_size = os.path.getsize(corefile)
+ if size == curr_size:
+ ok = True
+ break
+ if not ok:
+ cls.logger.error("Timed out waiting for coredump to complete:"
+ " %s", corefile)
+ else:
+ cls.logger.error("Coredump complete: %s, size %d",
+ corefile, curr_size)
+
@classmethod
def setUpClass(cls):
"""
if hasattr(cls, 'parallel_handler'):
cls.logger.addHandler(cls.parallel_handler)
cls.logger.propagate = False
+
cls.tempdir = tempfile.mkdtemp(
prefix='vpp-unittest-%s-' % cls.__name__)
cls.stats_sock = "%s/stats.sock" % cls.tempdir
+ cls.api_sock = "%s/api.sock" % cls.tempdir
cls.file_handler = FileHandler("%s/log.txt" % cls.tempdir)
cls.file_handler.setFormatter(
Formatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
datefmt="%H:%M:%S"))
cls.file_handler.setLevel(DEBUG)
cls.logger.addHandler(cls.file_handler)
+ cls.logger.debug("--- setUpClass() for %s called ---" %
+ cls.__name__)
cls.shm_prefix = os.path.basename(cls.tempdir)
os.chdir(cls.tempdir)
cls.logger.info("Temporary dir is %s, shm prefix is %s",
del cls.vapi
cls.vpp.poll()
if cls.vpp.returncode is None:
+ cls.wait_for_coredump()
cls.logger.debug("Sending TERM to vpp")
- cls.vpp.kill()
+ cls.vpp.terminate()
cls.logger.debug("Waiting for vpp to die")
cls.vpp.communicate()
cls.logger.debug("Deleting class vpp attribute on %s",
@classmethod
def tearDownClass(cls):
""" Perform final cleanup after running all tests in this test-case """
+ cls.logger.debug("--- tearDownClass() for %s called ---" %
+ cls.__name__)
cls.reporter.send_keep_alive(cls, 'tearDownClass')
cls.quit()
cls.file_handler.close()
if debug_framework:
debug_internal.on_tear_down_class(cls)
+ def show_commands_at_teardown(self):
+ """ Allow subclass specific teardown logging additions."""
+ self.logger.info("--- No test specific show commands provided. ---")
+
def tearDown(self):
""" Show various debug prints after each test """
self.logger.debug("--- tearDown() for %s.%s(%s) called ---" %
(self.__class__.__name__, self._testMethodName,
self._testMethodDoc))
if not self.vpp_dead:
+ self.logger.info(
+ "--- Logging show commands common to all testcases. ---")
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())
self.logger.info(self.vapi.ppcli("show run"))
self.logger.info(self.vapi.ppcli("show log"))
+ self.logger.info("Logging testcase specific show commands.")
+ self.show_commands_at_teardown()
self.registry.remove_vpp_config(self.logger)
# Save/Dump VPP api trace log
api_trace = "vpp_api_trace.%s.log" % self._testMethodName
""" Clear trace before running each test"""
super(VppTestCase, self).setUp()
self.reporter.send_keep_alive(self)
- self.logger.debug("--- setUp() for %s.%s(%s) called ---" %
- (self.__class__.__name__, self._testMethodName,
- self._testMethodDoc))
if self.vpp_dead:
raise Exception("VPP is dead when setting up the test")
self.sleep(.1, "during setUp")
cls.lo_interfaces = result
return result
+ @classmethod
+ def create_bvi_interfaces(cls, count):
+ """
+ Create BVI interfaces.
+
+ :param count: number of interfaces created.
+ :returns: List of created interfaces.
+ """
+ result = [VppBviInterface(cls) for i in range(count)]
+ for intf in result:
+ setattr(cls, intf.name, intf)
+ cls.bvi_interfaces = result
+ return result
+
@staticmethod
def extend_packet(packet, size, padding=' '):
"""
packet_len = len(packet) + 4
extend = size - packet_len
if extend > 0:
- num = (extend / len(padding)) + 1
- packet[Raw].load += (padding * num)[:extend]
+ num = (extend // len(padding)) + 1
+ packet[Raw].load += (padding * num)[:extend].encode("ascii")
@classmethod
def reset_packet_infos(cls):
Convert packet payload to _PacketInfo object
:param payload: packet payload
- :type: <class 'scapy.packet.Raw'>
- :param: payload_field: packet fieldname of payload "load" for
+ :type payload: <class 'scapy.packet.Raw'>
+ :param payload_field: packet fieldname of payload "load" for
<class 'scapy.packet.Raw'>
+ :type payload_field: str
:returns: _PacketInfo object containing de-serialized data from payload
"""
if pkt.haslayer(ICMPv6EchoReply):
self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
+ def get_packet_counter(self, counter):
+ if counter.startswith("/"):
+ counter_value = self.statistics.get_counter(counter)
+ else:
+ counters = self.vapi.cli("sh errors").split('\n')
+ counter_value = -1
+ for i in range(1, len(counters) - 1):
+ results = counters[i].split()
+ if results[1] == counter:
+ counter_value = int(results[0])
+ break
+ return counter_value
+
def assert_packet_counter_equal(self, counter, expected_value):
if counter.startswith("/"):
counter_value = self.statistics.get_counter(counter)
i.assert_nothing_captured(remark=remark)
timeout = 0.1
- def send_and_expect(self, intf, pkts, output):
+ def send_and_expect(self, intf, pkts, output, n_rx=None):
+ if not n_rx:
+ n_rx = len(pkts)
self.pg_send(intf, pkts)
- rx = output.get_capture(len(pkts))
+ rx = output.get_capture(n_rx)
return rx
def send_and_expect_only(self, intf, pkts, output, timeout=None):
Code Review / vpp.git / blobdiff