from struct import pack, unpack
import scapy.compat
-from scapy.packet import Raw
+from scapy.packet import Raw, Packet
import hook as hookmodule
from vpp_pg_interface import VppPGInterface
from vpp_sub_interface import VppSubInterface
RUN_SOLO = 1
# marks the suites broken on VPP multi-worker
FIXME_VPP_WORKERS = 2
+ # marks the suites broken when ASan is enabled
+ FIXME_ASAN = 3
def create_tag_decorator(e):
tag_run_solo = create_tag_decorator(TestCaseTag.RUN_SOLO)
tag_fixme_vpp_workers = create_tag_decorator(TestCaseTag.FIXME_VPP_WORKERS)
+tag_fixme_asan = create_tag_decorator(TestCaseTag.FIXME_ASAN)
class DummyVpp:
""" if the test case class is timing-sensitive - return true """
return cls.has_tag(TestCaseTag.RUN_SOLO)
+ @classmethod
+ def skip_fixme_asan(cls):
+ """ if @tag_fixme_asan & ASan is enabled - mark for skip """
+ if cls.has_tag(TestCaseTag.FIXME_ASAN):
+ vpp_extra_cmake_args = os.environ.get('VPP_EXTRA_CMAKE_ARGS', '')
+ if 'DVPP_ENABLE_SANITIZE_ADDR=ON' in vpp_extra_cmake_args:
+ cls = unittest.skip("Skipping @tag_fixme_asan tests")(cls)
+
@classmethod
def instance(cls):
"""Return the instance of this testcase"""
"Finished sleep (%s) - slept %es (wanted %es)",
remark, after - before, timeout)
+ def virtual_sleep(self, timeout, remark=None):
+ self.logger.debug("Moving VPP time by %s (%s)", timeout, remark)
+ self.vapi.cli("set clock adjust %s" % timeout)
+
def pg_send(self, intf, pkts, worker=None, trace=True):
intf.add_stream(pkts, worker=worker)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start(trace=trace)
- def send_and_assert_no_replies(self, intf, pkts, remark="", timeout=None):
+ def send_and_assert_no_replies(self, intf, pkts, remark="", timeout=None,
+ trace=True):
self.pg_send(intf, pkts)
if not timeout:
timeout = 1
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"))
def send_and_expect(self, intf, pkts, output, n_rx=None, worker=None,
trace=True):
if not n_rx:
- n_rx = len(pkts)
+ 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:
test_title = colorize(
f"FIXME with VPP workers: {test_title}", RED)
+ if test.has_tag(TestCaseTag.FIXME_ASAN):
+ test_title = colorize(
+ f"FIXME with ASAN: {test_title}", RED)
+ test.skip_fixme_asan()
+
if hasattr(test, 'vpp_worker_count'):
if test.vpp_worker_count == 0:
test_title += " [main thread only]"