#!/usr/bin/env python3
from __future__ import print_function
-import gc
import logging
import sys
import os
extra_vpp_plugin_config = []
logger = null_logger
vapi_response_timeout = 5
+ remove_configured_vpp_objects_on_tear_down = True
@property
def packet_infos(self):
@classmethod
def get_tempdir(cls):
- tmpdir = f"{config.tmp_dir}/vpp-unittest-{cls.__name__}"
- if config.wipe_tmp_dir:
- shutil.rmtree(tmpdir, ignore_errors=True)
- os.mkdir(tmpdir)
+ if cls.debug_attach:
+ tmpdir = f"{config.tmp_dir}/unittest-attach-gdb"
+ else:
+ tmpdir = f"{config.tmp_dir}/vpp-unittest-{cls.__name__}"
+ if config.wipe_tmp_dir:
+ shutil.rmtree(tmpdir, ignore_errors=True)
+ os.mkdir(tmpdir)
return tmpdir
@classmethod
self.logger.info(self.vapi.ppcli("show bihash"))
self.logger.info("Logging testcase specific show commands.")
self.show_commands_at_teardown()
- self.registry.remove_vpp_config(self.logger)
+ if self.remove_configured_vpp_objects_on_tear_down:
+ self.registry.remove_vpp_config(self.logger)
# Save/Dump VPP api trace log
m = self._testMethodName
api_trace = "vpp_api_trace.%s.%d.log" % (m, self.vpp.pid)
if pkt.haslayer(ICMPv6EchoReply):
self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
- def get_packet_counter(self, counter):
+ def get_counter(self, counter):
if counter.startswith("/"):
counter_value = self.statistics.get_counter(counter)
else:
break
return counter_value
+ def assert_counter_equal(self, counter, expected_value,
+ thread=None, index=0):
+ c = self.get_counter(counter)
+ if thread is not None:
+ c = c[thread][index]
+ else:
+ c = sum(x[index] for x in c)
+ self.assert_equal(c, expected_value, "counter `%s'" % counter)
+
def assert_packet_counter_equal(self, counter, expected_value):
- counter_value = self.get_packet_counter(counter)
+ counter_value = self.get_counter(counter)
self.assert_equal(counter_value, expected_value,
"packet counter `%s'" % counter)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start(trace=trace)
+ def snapshot_stats(self, stats_diff):
+ """Return snapshot of interesting stats based on diff dictionary."""
+ stats_snapshot = {}
+ for sw_if_index in stats_diff:
+ for counter in stats_diff[sw_if_index]:
+ stats_snapshot[counter] = self.statistics[counter]
+ self.logger.debug(f"Took statistics stats_snapshot: {stats_snapshot}")
+ return stats_snapshot
+
+ def compare_stats_with_snapshot(self, stats_diff, stats_snapshot):
+ """Assert appropriate difference between current stats and snapshot."""
+ for sw_if_index in stats_diff:
+ for cntr, diff in stats_diff[sw_if_index].items():
+ if sw_if_index == "err":
+ self.assert_equal(
+ self.statistics[cntr].sum(),
+ stats_snapshot[cntr].sum() + diff,
+ f"'{cntr}' counter value (previous value: "
+ f"{stats_snapshot[cntr].sum()}, "
+ f"expected diff: {diff})")
+ else:
+ try:
+ self.assert_equal(
+ self.statistics[cntr][:, sw_if_index].sum(),
+ stats_snapshot[cntr][:, sw_if_index].sum() + diff,
+ f"'{cntr}' counter value (previous value: "
+ f"{stats_snapshot[cntr][:, sw_if_index].sum()}, "
+ f"expected diff: {diff})")
+ except IndexError:
+ # if diff is 0, then this most probably a case where
+ # test declares multiple interfaces but traffic hasn't
+ # passed through this one yet - which means the counter
+ # value is 0 and can be ignored
+ if 0 != diff:
+ raise
+
def send_and_assert_no_replies(self, intf, pkts, remark="", timeout=None,
- trace=True):
+ stats_diff=None, trace=True, msg=None):
+ if stats_diff:
+ stats_snapshot = self.snapshot_stats(stats_diff)
+
self.pg_send(intf, pkts)
- 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
- if trace:
- self.logger.debug(self.vapi.cli("show trace"))
+
+ try:
+ if not timeout:
+ timeout = 1
+ for i in self.pg_interfaces:
+ i.assert_nothing_captured(timeout=timeout, remark=remark)
+ timeout = 0.1
+ finally:
+ if trace:
+ if msg:
+ self.logger.debug(f"send_and_assert_no_replies: {msg}")
+ self.logger.debug(self.vapi.cli("show trace"))
+
+ if stats_diff:
+ self.compare_stats_with_snapshot(stats_diff, stats_snapshot)
def send_and_expect(self, intf, pkts, output, n_rx=None, worker=None,
- trace=True):
+ trace=True, msg=None, stats_diff=None):
+ if stats_diff:
+ stats_snapshot = self.snapshot_stats(stats_diff)
+
if not n_rx:
n_rx = 1 if isinstance(pkts, Packet) else len(pkts)
self.pg_send(intf, pkts, worker=worker, trace=trace)
rx = output.get_capture(n_rx)
+ if trace:
+ if msg:
+ self.logger.debug(f"send_and_expect: {msg}")
+ self.logger.debug(self.vapi.cli("show trace"))
+
+ if stats_diff:
+ self.compare_stats_with_snapshot(stats_diff, stats_snapshot)
+
+ return rx
+
+ def send_and_expect_load_balancing(self, input, pkts, outputs,
+ worker=None, trace=True):
+ self.pg_send(input, pkts, worker=worker, trace=trace)
+ rxs = []
+ for oo in outputs:
+ rx = oo._get_capture(1)
+ self.assertNotEqual(0, len(rx))
+ rxs.append(rx)
if trace:
self.logger.debug(self.vapi.cli("show trace"))
+ return rxs
+
+ def send_and_expect_some(self, intf, pkts, output,
+ worker=None,
+ trace=True):
+ self.pg_send(intf, pkts, worker=worker, trace=trace)
+ rx = output._get_capture(1)
+ if trace:
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.assertTrue(len(rx) > 0)
+ self.assertTrue(len(rx) < len(pkts))
return rx
- def send_and_expect_only(self, intf, pkts, output, timeout=None):
+ def send_and_expect_only(self, intf, pkts, output, timeout=None,
+ stats_diff=None):
+ if stats_diff:
+ stats_snapshot = self.snapshot_stats(stats_diff)
+
self.pg_send(intf, pkts)
rx = output.get_capture(len(pkts))
outputs = [output]
timeout = 1
for i in self.pg_interfaces:
if i not in outputs:
- i.get_capture(0, timeout=timeout)
- i.assert_nothing_captured()
+ i.assert_nothing_captured(timeout=timeout)
timeout = 0.1
+ if stats_diff:
+ self.compare_stats_with_snapshot(stats_diff, stats_snapshot)
+
return rx
Code Review / vpp.git / blobdiff