def pump_output(testclass):
"""pump output from vpp stdout/stderr to proper queues"""
+ if not hasattr(testclass, "vpp"):
+ return
stdout_fragment = ""
stderr_fragment = ""
while not testclass.pump_thread_stop_flag.is_set():
is_distro_ubuntu2204 = _is_distro_ubuntu2204()
+def _is_distro_debian11():
+ with open("/etc/os-release") as f:
+ for line in f.readlines():
+ if "bullseye" in line:
+ return True
+ return False
+
+
+is_distro_debian11 = _is_distro_debian11()
+
+
class KeepAliveReporter(object):
"""
Singleton object which reports test start to parent process
"""
Write current test tmpdir & desc to keep-alive pipe to signal liveness
"""
- if self.pipe is None:
+ if not hasattr(test, "vpp") or self.pipe is None:
# if not running forked..
return
FIXME_ASAN = 3
# marks suites broken on Ubuntu-22.04
FIXME_UBUNTU2204 = 4
+ # marks suites broken on Debian-11
+ FIXME_DEBIAN11 = 5
def create_tag_decorator(e):
tag_fixme_vpp_workers = create_tag_decorator(TestCaseTag.FIXME_VPP_WORKERS)
tag_fixme_asan = create_tag_decorator(TestCaseTag.FIXME_ASAN)
tag_fixme_ubuntu2204 = create_tag_decorator(TestCaseTag.FIXME_UBUNTU2204)
+tag_fixme_debian11 = create_tag_decorator(TestCaseTag.FIXME_DEBIAN11)
class DummyVpp:
if cls.has_tag(TestCaseTag.FIXME_UBUNTU2204):
cls = unittest.skip("Skipping @tag_fixme_ubuntu2204 tests")(cls)
+ @classmethod
+ def skip_fixme_debian11(cls):
+ """if distro is Debian-11 and @tag_fixme_debian11 mark for skip"""
+ if cls.has_tag(TestCaseTag.FIXME_DEBIAN11):
+ cls = unittest.skip("Skipping @tag_fixme_debian11 tests")(cls)
+
@classmethod
def instance(cls):
"""Return the instance of this testcase"""
@classmethod
def run_vpp(cls):
+ if (
+ is_distro_ubuntu2204 == True and cls.has_tag(TestCaseTag.FIXME_UBUNTU2204)
+ ) or (is_distro_debian11 == True and cls.has_tag(TestCaseTag.FIXME_DEBIAN11)):
+ return
cls.logger.debug(f"Assigned cpus: {cls.cpus}")
cmdline = cls.vpp_cmdline
cls.attach_vpp()
else:
cls.run_vpp()
+ if not hasattr(cls, "vpp"):
+ return
cls.reporter.send_keep_alive(cls, "setUpClass")
VppTestResult.current_test_case_info = TestCaseInfo(
cls.logger, cls.tempdir, cls.vpp.pid, config.vpp
def tearDownClass(cls):
"""Perform final cleanup after running all tests in this test-case"""
cls.logger.debug("--- tearDownClass() for %s called ---" % cls.__name__)
+ if not hasattr(cls, "vpp"):
+ return
cls.reporter.send_keep_alive(cls, "tearDownClass")
cls.quit()
cls.file_handler.close()
"--- tearDown() for %s.%s(%s) called ---"
% (self.__class__.__name__, self._testMethodName, self._testMethodDoc)
)
+ if not hasattr(self, "vpp"):
+ return
try:
if not self.vpp_dead:
def setUp(self):
"""Clear trace before running each test"""
super(VppTestCase, self).setUp()
+ if not hasattr(self, "vpp"):
+ return
self.reporter.send_keep_alive(self)
if self.vpp_dead:
raise VppDiedError(
@classmethod
def create_pg_ip4_interfaces(cls, interfaces, gso=0, gso_size=0):
+ if not hasattr(cls, "vpp"):
+ cls.pg_interfaces = []
+ return cls.pg_interfaces
pgmode = VppEnum.vl_api_pg_interface_mode_t
return cls.create_pg_interfaces_internal(
interfaces, gso, gso_size, pgmode.PG_API_MODE_IP4
@classmethod
def create_pg_ip6_interfaces(cls, interfaces, gso=0, gso_size=0):
+ if not hasattr(cls, "vpp"):
+ cls.pg_interfaces = []
+ return cls.pg_interfaces
pgmode = VppEnum.vl_api_pg_interface_mode_t
return cls.create_pg_interfaces_internal(
interfaces, gso, gso_size, pgmode.PG_API_MODE_IP6
@classmethod
def create_pg_interfaces(cls, interfaces, gso=0, gso_size=0):
+ if not hasattr(cls, "vpp"):
+ cls.pg_interfaces = []
+ return cls.pg_interfaces
pgmode = VppEnum.vl_api_pg_interface_mode_t
return cls.create_pg_interfaces_internal(
interfaces, gso, gso_size, pgmode.PG_API_MODE_ETHERNET
@classmethod
def create_pg_ethernet_interfaces(cls, interfaces, gso=0, gso_size=0):
+ if not hasattr(cls, "vpp"):
+ cls.pg_interfaces = []
+ return cls.pg_interfaces
pgmode = VppEnum.vl_api_pg_interface_mode_t
return cls.create_pg_interfaces_internal(
interfaces, gso, gso_size, pgmode.PG_API_MODE_ETHERNET
:param count: number of interfaces created.
:returns: List of created interfaces.
"""
+ if not hasattr(cls, "vpp"):
+ cls.lo_interfaces = []
+ return cls.lo_interfaces
result = [VppLoInterface(cls) for i in range(count)]
for intf in result:
setattr(cls, intf.name, intf)
:param count: number of interfaces created.
:returns: List of created interfaces.
"""
+ if not hasattr(cls, "vpp"):
+ cls.bvi_interfaces = []
+ return cls.bvi_interfaces
result = [VppBviInterface(cls) for i in range(count)]
for intf in result:
setattr(cls, intf.name, intf)
test_title = colorize(f"FIXME on Ubuntu-22.04: {test_title}", RED)
test.skip_fixme_ubuntu2204()
+ if is_distro_debian11 == True and test.has_tag(TestCaseTag.FIXME_DEBIAN11):
+ test_title = colorize(f"FIXME on Debian-11: {test_title}", RED)
+ test.skip_fixme_debian11()
+
if hasattr(test, "vpp_worker_count"):
if test.vpp_worker_count == 0:
test_title += " [main thread only]"
Poll the vpp status and throw an exception if it's not running
:raises VppDiedError: exception if VPP is not running anymore
"""
- if self.test.vpp_dead:
+ if not hasattr(self.test, "vpp") or self.test.vpp_dead:
# already dead, nothing to do
return
BFDState,
BFD_vpp_echo,
)
-from framework import tag_fixme_vpp_workers
+from framework import tag_fixme_vpp_workers, tag_fixme_ubuntu2204, tag_fixme_debian11
+from framework import is_distro_ubuntu2204, is_distro_debian11
from framework import VppTestCase, VppTestRunner
from framework import tag_run_solo
from util import ppp
@tag_run_solo
+@tag_fixme_ubuntu2204
+@tag_fixme_debian11
class BFD4TestCase(VppTestCase):
"""Bidirectional Forwarding Detection (BFD)"""
@classmethod
def setUpClass(cls):
+ if (is_distro_ubuntu2204 == True or is_distro_debian11 == True) and not hasattr(
+ cls, "vpp"
+ ):
+ return
super(BFD4TestCase, cls).setUpClass()
cls.vapi.cli("set log class bfd level debug")
try:
@tag_run_solo
@tag_fixme_vpp_workers
+@tag_fixme_ubuntu2204
class BFD6TestCase(VppTestCase):
"""Bidirectional Forwarding Detection (BFD) (IPv6)"""
@classmethod
def setUpClass(cls):
+ if is_distro_ubuntu2204 == True and not hasattr(cls, "vpp"):
+ return
super(BFD6TestCase, cls).setUpClass()
cls.vapi.cli("set log class bfd level debug")
try:
def setUp(self):
super(BFD6TestCase, self).setUp()
+ if is_distro_ubuntu2204 == True and not hasattr(self, "vpp"):
+ return
self.factory = AuthKeyFactory()
self.vapi.want_bfd_events()
self.pg0.enable_capture()
from scapy.layers.inet6 import IPv6
from config import config
-from framework import tag_fixme_vpp_workers
+from framework import tag_fixme_vpp_workers, tag_fixme_ubuntu2204, tag_fixme_debian11
+from framework import is_distro_ubuntu2204, is_distro_debian11
from framework import VppTestCase, VppTestRunner
from framework import tag_run_solo
from vpp_object import VppObject
variables and configure VPP.
"""
super(MethodHolder, cls).setUpClass()
+ if (is_distro_ubuntu2204 == True or is_distro_debian11 == True) and not hasattr(
+ cls, "vpp"
+ ):
+ return
try:
# Create pg interfaces
cls.create_pg_interfaces(range(9))
@tag_run_solo
@tag_fixme_vpp_workers
+@tag_fixme_ubuntu2204
+@tag_fixme_debian11
class Flowprobe(MethodHolder):
"""Template verification, timer tests"""
from scapy.layers.inet6 import IPv6
from scapy.packet import raw, Raw
from scapy.utils import long_converter
-from framework import tag_fixme_vpp_workers
+from framework import tag_fixme_vpp_workers, tag_fixme_ubuntu2204, tag_fixme_debian11
+from framework import is_distro_ubuntu2204, is_distro_debian11
from framework import VppTestCase, VppTestRunner
from vpp_ikev2 import Profile, IDType, AuthMethod
from vpp_papi import VppEnum
WITH_KEX = True
+@tag_fixme_ubuntu2204
+@tag_fixme_debian11
class TestResponderVrf(TestResponderPsk, Ikev2Params):
"""test ikev2 responder - non-default table id"""
globals()["ikev2"] = _ikev2
super(IkePeer, cls).setUpClass()
+ if (is_distro_ubuntu2204 == True or is_distro_debian11 == True) and not hasattr(
+ cls, "vpp"
+ ):
+ return
cls.create_pg_interfaces(range(1))
cls.vapi.cli("ip table add 1")
cls.vapi.cli("set interface ip table pg0 1")
from scapy.layers.inet import IP, ICMP
from framework import VppTestCase, VppTestRunner
-from framework import tag_run_solo
+from framework import tag_run_solo, tag_fixme_debian11, is_distro_debian11
from remote_test import RemoteClass, RemoteVppTestCase
from vpp_memif import remove_all_memif_vpp_config, VppSocketFilename, VppMemif
from vpp_ip_route import VppIpRoute, VppRoutePath
@tag_run_solo
+@tag_fixme_debian11
class TestMemif(VppTestCase):
"""Memif Test Case"""
cls.remote_test.start_remote()
cls.remote_test.set_request_timeout(10)
super(TestMemif, cls).setUpClass()
+ if is_distro_debian11 == True and not hasattr(cls, "vpp"):
+ cls.remote_test.quit_remote()
+ return
cls.remote_test.setUpClass(cls.tempdir)
cls.create_pg_interfaces(range(1))
for pg in cls.pg_interfaces:
from random import randint, choice
import scapy.compat
+from framework import tag_fixme_ubuntu2204, is_distro_ubuntu2204
from framework import VppTestCase, VppTestRunner
from scapy.data import IP_PROTOS
from scapy.layers.inet import IP, TCP, UDP, ICMP, GRE
@classmethod
def setUpClass(cls):
super().setUpClass()
+ if is_distro_ubuntu2204 == True and not hasattr(cls, "vpp"):
+ return
cls.create_pg_interfaces(range(12))
cls.interfaces = list(cls.pg_interfaces[:4])
out_if.unconfig()
+@tag_fixme_ubuntu2204
class TestNAT44EDMW(TestNAT44ED):
"""NAT44ED MW Test Case"""
from io import BytesIO
import scapy.compat
+from framework import tag_fixme_debian11, is_distro_debian11
from framework import VppTestCase, VppTestRunner
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from scapy.all import (
return 1 + h % vpp_worker_count
+@tag_fixme_debian11
class TestNAT44EI(MethodHolder):
"""NAT44EI Test Cases"""
@classmethod
def setUpClass(cls):
super(TestNAT44EI, cls).setUpClass()
+ if is_distro_debian11 == True and not hasattr(cls, "vpp"):
+ return
cls.vapi.cli("set log class nat44-ei level debug")
cls.tcp_port_in = 6303
import scapy.compat
from config import config
-from framework import tag_fixme_vpp_workers
+from framework import tag_fixme_vpp_workers, tag_fixme_ubuntu2204, is_distro_ubuntu2204
from framework import VppTestCase, VppTestRunner
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from scapy.data import IP_PROTOS
@tag_fixme_vpp_workers
+@tag_fixme_ubuntu2204
class TestNAT64(VppTestCase):
"""NAT64 Test Cases"""
def setUpClass(cls):
super(TestNAT64, cls).setUpClass()
+ if is_distro_ubuntu2204 == True and not hasattr(cls, "vpp"):
+ return
cls.tcp_port_in = 6303
cls.tcp_port_out = 6303
cls.udp_port_in = 6304
import os
from socket import AF_INET, AF_INET6, inet_pton
-from framework import tag_fixme_vpp_workers
+from framework import tag_fixme_vpp_workers, tag_fixme_ubuntu2204, tag_fixme_debian11
from framework import VppTestCase, VppTestRunner
from vpp_neighbor import VppNeighbor, find_nbr
from vpp_ip_route import (
self.assertEqual(NUM_PKTS + 16, nd1.get_stats()["packets"])
+@tag_fixme_ubuntu2204
class NeighborAgeTestCase(VppTestCase):
"""ARP/ND Aging"""
from vpp_ip_route import VppIpRoute, VppRoutePath
from vpp_object import VppObject
from vpp_papi import VppEnum
+from framework import tag_fixme_ubuntu2204, tag_fixme_debian11
+from framework import is_distro_ubuntu2204, is_distro_debian11
from framework import VppTestCase
from re import compile
import unittest
self._test.assertEqual(rv.peer_index, self.index)
+@tag_fixme_ubuntu2204
+@tag_fixme_debian11
class TestWg(VppTestCase):
"""Wireguard Test Case"""
@classmethod
def setUpClass(cls):
super(TestWg, cls).setUpClass()
+ if (is_distro_ubuntu2204 == True or is_distro_debian11 == True) and not hasattr(
+ cls, "vpp"
+ ):
+ return
try:
cls.create_pg_interfaces(range(3))
for i in cls.pg_interfaces:
wg1.remove_vpp_config()
+@tag_fixme_ubuntu2204
+@tag_fixme_debian11
class WireguardHandoffTests(TestWg):
"""Wireguard Tests in multi worker setup"""
Code Review / vpp.git / commitdiff