--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuAfPacketGsoL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test af_packet interfaces with GSO in L2 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "14,16,22,23"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuAfPacketGsoL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuAfPacketGsoL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuAfPacketGsoL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuAfPacketGsoL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuAfPacketGsoL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuAfPacketGsoL3(TestVPPInterfacesQemu, VppTestCase):
+ """Test af_packet interfaces with GSO in L3 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "15,19"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuAfPacketGsoL3, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuAfPacketGsoL3, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuAfPacketGsoL3, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuAfPacketGsoL3.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuAfPacketGsoL3)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuAfPacketL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test af_packet interfaces in L2 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "12,20"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuAfPacketL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuAfPacketL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuAfPacketL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuAfPacketL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuAfPacketL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuAfpacketL3(TestVPPInterfacesQemu, VppTestCase):
+ """Test af_packet interfaces in L3 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "13"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuAfpacketL3, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuAfpacketL3, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuAfpacketL3, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuAfpacketL3.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuAfpacketL3)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuAfPacketTapGsoL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test af_packet & tap interfaces with GSO in L2 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "17"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuAfPacketTapGsoL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuAfPacketTapGsoL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuAfPacketTapGsoL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuAfPacketTapGsoL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuAfPacketTapGsoL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuAfPacketTunGroL3(TestVPPInterfacesQemu, VppTestCase):
+ """Test af_packet & tun interfaces with GRO in L3 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "21"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuAfPacketTunGroL3, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuAfPacketTunGroL3, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuAfPacketTunGroL3, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuAfPacketTunGroL3.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuAfPacketTunGroL3)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuAfPacketTunGsoL3(TestVPPInterfacesQemu, VppTestCase):
+ """Test af_packet & tun interfaces with GSO in L3 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "18"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuAfPacketTunGsoL3, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuAfPacketTunGsoL3, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuAfPacketTunGsoL3, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuAfPacketTunGsoL3.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuAfPacketTunGsoL3)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuMemifL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test Memif interfaces in L2 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "27"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuMemifL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuMemifL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuMemifL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuMemifL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuMemifL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuTapChecksumOffloadL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test tap interfaces in L2 mode with checksum offload for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "24"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuTapChecksumOffloadL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuTapChecksumOffloadL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuTapChecksumOffloadL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuTapChecksumOffloadL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuTapChecksumOffloadL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuTapChecksumOffloadL3(TestVPPInterfacesQemu, VppTestCase):
+ """Test tap interfaces in L3 mode with checksum offload for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "25,26"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuTapChecksumOffloadL3, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuTapChecksumOffloadL3, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuTapChecksumOffloadL3, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuTapChecksumOffloadL3.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuTapChecksumOffloadL3)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuTapGroL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test tap interfaces with GRO in L2 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "4,5,9"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuTapGroL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuTapGroL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuTapGroL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuTapGroL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuTapGroL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuTapGsoGroL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test tap interfaces with GSO & GRO in L2 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "6,7"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuTapGsoGroL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuTapGsoGroL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuTapGsoGroL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuTapGsoGroL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuTapGsoGroL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuTapGsoL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test tap interfaces with GSO in L2 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "2,3,8"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuTapGsoL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuTapGsoL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuTapGsoL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuTapGsoL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuTapGsoL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuTapL2(TestVPPInterfacesQemu, VppTestCase):
+ """Test VPP tap interfaces in L2 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "1"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuTapL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuTapL2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuTapL2, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuTapL2.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuTapL2)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+import unittest
+from framework import VppTestCase
+from vm_vpp_interfaces import (
+ TestSelector,
+ TestVPPInterfacesQemu,
+ generate_vpp_interface_tests,
+)
+from asfframework import VppTestRunner
+from vm_test_config import test_config
+
+
+class TestVPPInterfacesQemuTapTunGroL3(TestVPPInterfacesQemu, VppTestCase):
+ """Test tap & tun interfaces with GRO in L3 mode for IPv4/v6."""
+
+ # Set test_id(s) to run from vm_test_config
+ # The expansion of these numbers are included in the test docstring
+ tests_to_run = "10,11"
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestVPPInterfacesQemuTapTunGroL3, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestVPPInterfacesQemuTapTunGroL3, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestVPPInterfacesQemuTapTunGroL3, self).tearDown()
+
+
+SELECTED_TESTS = TestVPPInterfacesQemuTapTunGroL3.tests_to_run
+tests = filter(TestSelector(SELECTED_TESTS).filter_tests, test_config["tests"])
+generate_vpp_interface_tests(tests, TestVPPInterfacesQemuTapTunGroL3)
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
#
-def filter_tests(test):
- """Filter test IDs to include only those selected to run."""
- selection = test_config["tests_to_run"]
- if not selection or selection == " ":
- return True
- else:
+class TestSelector:
+ """Selects specified test(s) from vm_test_config to run
+
+ The selected_test field specifies a comma separated or range(s) of
+ tests to run (default='' i.e all_tests) e.g. setting the selected_tests
+ attribute to "1,3-4,19-23" runs tests with ID's 1, 3, 4, 19, 20, 21,
+ 22 & 23 from the spec file vm_test_config
+ """
+
+ def __init__(self, selected_tests="") -> None:
+ self.selected_tests = selected_tests
+
+ def filter_tests(self, test):
+ """Works with the filter fn. to include only selected tests."""
+
+ if self.selected_tests:
+ selection = self.selected_tests
+ else:
+ selection = test_config["tests_to_run"]
+
+ if not selection or selection == " ":
+ return True
+
test_ids_to_run = []
for test_id in selection.split(","):
if "-" in test_id.strip():
# Test Config variables
-client_namespace = test_config["client_namespace"]
-server_namespace = test_config["server_namespace"]
-tests = filter(filter_tests, test_config["tests"])
af_packet_config = test_config["af_packet"]
layer2 = test_config["L2"]
layer3 = test_config["L3"]
memif_sock_path=self.get_memif_sock_path(), logger=self.logger
)
if result is True:
- start_iperf(ip_version=6, server_only=True, logger=self.logger)
+ # Start an instance of an iperf server using
+ # a unique port. Save the iperf cmdline for
+ # terminating the iperf_server process after the test.
+ self.iperf_cmd = start_iperf(
+ ip_version=6,
+ client_ns=self.client_namespace,
+ server_ns=self.server_namespace,
+ server_only=True,
+ server_args=f"-p {self.iperf_port}",
+ logger=self.logger,
+ )
+ # Send traffic between iperf client & server
self.assertTrue(
start_iperf(
ip_version=ip_version,
+ client_ns=self.client_namespace,
+ server_ns=self.server_namespace,
server_ipv4_address=self.server_ip4_address,
server_ipv6_address=self.server_ip6_address,
+ client_args=f"-p {self.iperf_port}",
client_only=True,
duration=2,
logger=self.logger,
return test_func
-def generate_vpp_interface_tests():
- """Generate unittests for testing vpp interfaces."""
+def generate_vpp_interface_tests(tests, test_class):
+ """Generate unittests for testing vpp interfaces
+
+ Generates unittests from test spec. and sets them as attributes
+ to the test_class.
+ Args:
+ tests : list of test specs from vm_test_config['tests']
+ test_class : the name of the test class to which the
+ generated tests are set as attributes.
+ """
+
if config.skip_netns_tests:
print("Skipping netns tests")
for test in tests:
test_func = create_test(
test_name=test_name, test=test, ip_version=ip_version, mtu=mtu
)
- setattr(TestVPPInterfacesQemu, test_name, test_func)
+ setattr(test_class, test_name, test_func)
@tag_fixme_debian11
-class TestVPPInterfacesQemu(VppTestCase):
+class TestVPPInterfacesQemu:
"""Test VPP interfaces inside a QEMU VM for IPv4/v6.
Test Setup:
--vppaf_packet_int2--host-int2--iperfServer--Linux_ns2
"""
- @classmethod
- def setUpClass(cls):
- super(TestVPPInterfacesQemu, cls).setUpClass()
-
- @classmethod
- def tearDownClass(cls):
- super(TestVPPInterfacesQemu, cls).tearDownClass()
-
def setUpTestToplogy(self, test, ip_version):
"""Setup the test topology.
3. Enable desired vif features such as GSO & GRO.
3. Cross-Connect interfaces in VPP using L2 or L3.
"""
- super(TestVPPInterfacesQemu, self).setUp()
+
# Need to support multiple interface types as the memif interface
# in VPP is connected to the iPerf client & server by x-connecting
# to a tap interface in their respective namespaces.
else layer3["vpp_server_ip6_prefix"]
)
vpp_server_nexthop = str(ip_interface(vpp_server_prefix).ip)
- create_namespace([client_namespace, server_namespace])
+ # Create unique namespaces for iperf client & iperf server to
+ # prevent conflicts when TEST_JOBS > 1
+ self.client_namespace = test_config["client_namespace"] + str(test["id"])
+ self.server_namespace = test_config["server_namespace"] + str(test["id"])
+ create_namespace([self.client_namespace, self.server_namespace])
+ # Set a unique iPerf port for parallel server and client runs
+ self.iperf_port = 5000 + test["id"]
# IPerf client & server ingress/egress interface indexes in VPP
self.tap_interfaces = []
self.memif_interfaces = []
enable_server_if_gro = test.get("server_if_gro", 0)
enable_client_if_checksum_offload = test.get("client_if_checksum_offload", 0)
enable_server_if_checksum_offload = test.get("server_if_checksum_offload", 0)
- ## Handle client interface types
+
+ # Create unique host interfaces in Linux and VPP for connecting to iperf
+ # client & iperf server to prevent conflicts when TEST_JOBS > 1
+ self.iprf_client_host_interface_on_linux = af_packet_config[
+ "iprf_client_interface_on_linux"
+ ] + str(test["id"])
+ self.iprf_client_host_interface_on_vpp = af_packet_config[
+ "iprf_client_interface_on_vpp"
+ ] + str(test["id"])
+ self.iprf_server_host_interface_on_linux = af_packet_config[
+ "iprf_server_interface_on_linux"
+ ] + str(test["id"])
+ self.iprf_server_host_interface_on_vpp = af_packet_config[
+ "iprf_server_interface_on_vpp"
+ ] + str(test["id"])
+ # Handle client interface types
for client_if_type in client_if_types:
if client_if_type == "af_packet":
create_host_interface(
- af_packet_config["iprf_client_interface_on_linux"],
- af_packet_config["iprf_client_interface_on_vpp"],
- client_namespace,
+ self.iprf_client_host_interface_on_linux,
+ self.iprf_client_host_interface_on_vpp,
+ self.client_namespace,
layer2["client_ip4_prefix"]
if x_connect_mode == "L2"
else layer3["client_ip4_prefix"],
)
self.ingress_if_idx = self.create_af_packet(
version=client_if_version,
- host_if_name=af_packet_config["iprf_client_interface_on_vpp"],
+ host_if_name=self.iprf_client_host_interface_on_vpp,
enable_gso=enable_client_if_gso,
)
self.ingress_if_idxes.append(self.ingress_if_idx)
self.vpp_interfaces.append(self.ingress_if_idx)
self.linux_interfaces.append(
- ["", af_packet_config["iprf_client_interface_on_vpp"]]
+ ["", self.iprf_client_host_interface_on_vpp]
)
self.linux_interfaces.append(
[
- client_namespace,
- af_packet_config["iprf_client_interface_on_linux"],
+ self.client_namespace,
+ self.iprf_client_host_interface_on_linux,
]
)
if enable_client_if_gso == 0:
+ disable_interface_gso("", self.iprf_client_host_interface_on_vpp)
disable_interface_gso(
- "", af_packet_config["iprf_client_interface_on_vpp"]
- )
- disable_interface_gso(
- client_namespace,
- af_packet_config["iprf_client_interface_on_linux"],
+ self.client_namespace,
+ self.iprf_client_host_interface_on_linux,
)
elif client_if_type == "tap" or client_if_type == "tun":
self.ingress_if_idx = self.create_tap_tun(
id=101,
- host_namespace=client_namespace,
+ host_namespace=self.client_namespace,
ip_version=ip_version,
host_ip4_prefix=layer2["client_ip4_prefix"]
if x_connect_mode == "L2"
self.tap_interfaces.append(self.ingress_if_idx)
self.ingress_if_idxes.append(self.ingress_if_idx)
self.vpp_interfaces.append(self.ingress_if_idx)
- self.linux_interfaces.append([client_namespace, f"{client_if_type}0"])
+ self.linux_interfaces.append(
+ [self.client_namespace, f"{client_if_type}0"]
+ )
# Seeing TCP timeouts if tx=on & rx=on Linux tap & tun interfaces
- disable_interface_gso(client_namespace, f"{client_if_type}0")
+ disable_interface_gso(self.client_namespace, f"{client_if_type}0")
elif client_if_type == "memif":
self.ingress_if_idx = self.create_memif(
memif_id=0, mode=0 if x_connect_mode == "L2" else 1
for server_if_type in server_if_types:
if server_if_type == "af_packet":
create_host_interface(
- af_packet_config["iprf_server_interface_on_linux"],
- af_packet_config["iprf_server_interface_on_vpp"],
- server_namespace,
+ self.iprf_server_host_interface_on_linux,
+ self.iprf_server_host_interface_on_vpp,
+ self.server_namespace,
server_ip4_prefix,
server_ip6_prefix,
)
self.egress_if_idx = self.create_af_packet(
version=server_if_version,
- host_if_name=af_packet_config["iprf_server_interface_on_vpp"],
+ host_if_name=self.iprf_server_host_interface_on_vpp,
enable_gso=enable_server_if_gso,
)
self.egress_if_idxes.append(self.egress_if_idx)
self.vpp_interfaces.append(self.egress_if_idx)
self.linux_interfaces.append(
- ["", af_packet_config["iprf_server_interface_on_vpp"]]
+ ["", self.iprf_server_host_interface_on_vpp]
)
self.linux_interfaces.append(
[
- server_namespace,
- af_packet_config["iprf_server_interface_on_linux"],
+ self.server_namespace,
+ self.iprf_server_host_interface_on_linux,
]
)
if enable_server_if_gso == 0:
+ disable_interface_gso("", self.iprf_server_host_interface_on_vpp)
disable_interface_gso(
- "", af_packet_config["iprf_server_interface_on_vpp"]
- )
- disable_interface_gso(
- server_namespace,
- af_packet_config["iprf_server_interface_on_linux"],
+ self.server_namespace,
+ self.iprf_server_host_interface_on_linux,
)
elif server_if_type == "tap" or server_if_type == "tun":
self.egress_if_idx = self.create_tap_tun(
id=102,
- host_namespace=server_namespace,
+ host_namespace=self.server_namespace,
ip_version=ip_version,
host_ip4_prefix=layer2["server_ip4_prefix"]
if x_connect_mode == "L2"
self.tap_interfaces.append(self.egress_if_idx)
self.egress_if_idxes.append(self.egress_if_idx)
self.vpp_interfaces.append(self.egress_if_idx)
- self.linux_interfaces.append([server_namespace, f"{server_if_type}0"])
+ self.linux_interfaces.append(
+ [self.server_namespace, f"{server_if_type}0"]
+ )
# Seeing TCP timeouts if tx=on & rx=on Linux tap & tun interfaces
- disable_interface_gso(server_namespace, f"{server_if_type}0")
+ disable_interface_gso(self.server_namespace, f"{server_if_type}0")
elif server_if_type == "memif":
self.egress_if_idx = self.create_memif(
memif_id=1, mode=0 if x_connect_mode == "L2" else 1
# Setup namespace routing
if ip_version == 4:
add_namespace_route(
- client_namespace, "0.0.0.0/0", vpp_client_nexthop
+ self.client_namespace, "0.0.0.0/0", vpp_client_nexthop
)
add_namespace_route(
- server_namespace, "0.0.0.0/0", vpp_server_nexthop
+ self.server_namespace, "0.0.0.0/0", vpp_server_nexthop
)
else:
- add_namespace_route(client_namespace, "::/0", vpp_client_nexthop)
- add_namespace_route(server_namespace, "::/0", vpp_server_nexthop)
+ add_namespace_route(
+ self.client_namespace, "::/0", vpp_client_nexthop
+ )
+ add_namespace_route(
+ self.server_namespace, "::/0", vpp_server_nexthop
+ )
else:
# connect: ingress tap & memif & egress tap and memif interfaces
if x_connect_mode == "L2":
pass
try:
for interface in self.vapi.af_packet_dump():
- if (
- interface.host_if_name
- == af_packet_config["iprf_client_interface_on_vpp"]
- ):
- self.vapi.af_packet_delete(
- af_packet_config["iprf_client_interface_on_vpp"]
- )
- elif (
- interface.host_if_name
- == af_packet_config["iprf_server_interface_on_vpp"]
- ):
- self.vapi.af_packet_delete(
- af_packet_config["iprf_server_interface_on_vpp"]
- )
+ if interface.host_if_name == self.iprf_client_host_interface_on_vpp:
+ self.vapi.af_packet_delete(self.iprf_client_host_interface_on_vpp)
+ elif interface.host_if_name == self.iprf_server_host_interface_on_vpp:
+ self.vapi.af_packet_delete(self.iprf_server_host_interface_on_vpp)
except Exception:
pass
try:
delete_host_interfaces(
- af_packet_config["iprf_client_interface_on_linux"],
- af_packet_config["iprf_server_interface_on_linux"],
- af_packet_config["iprf_client_interface_on_vpp"],
- af_packet_config["iprf_server_interface_on_vpp"],
+ self.iprf_client_host_interface_on_linux,
+ self.iprf_server_host_interface_on_linux,
+ self.iprf_client_host_interface_on_vpp,
+ self.iprf_server_host_interface_on_vpp,
)
except Exception:
pass
try:
delete_namespace(
[
- client_namespace,
- server_namespace,
+ self.client_namespace,
+ self.server_namespace,
]
)
except Exception:
pass
try:
- stop_iperf()
+ if hasattr(self, "iperf_cmd"):
+ stop_iperf(" ".join(self.iperf_cmd))
except Exception:
pass
try:
return False
-generate_vpp_interface_tests()
-
if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
sys.exit(1)
def start_iperf_server(self):
+ """Starts the iperf server and returns the process cmdline args."""
args = [
"ip",
"netns",
"-D",
]
args.extend(self.server_args.split())
- args = " ".join(args)
- self.logger.debug(f"Starting iperf server: {args}")
+ cmd = " ".join(args)
+ self.logger.debug(f"Starting iperf server: {cmd}")
try:
- return subprocess.run(
- args,
+ subprocess.run(
+ cmd,
timeout=self.duration + 5,
encoding="utf-8",
shell=True,
)
except subprocess.TimeoutExpired as e:
raise Exception("Error: Timeout expired for iPerf", e.output)
+ return args[4:]
def start_iperf_client(self):
args = [
"""
self.ensure_init()
if not client_only:
- self.start_iperf_server()
+ return self.start_iperf_server()
if not server_only:
result = self.start_iperf_client()
self.logger.debug(f"Iperf client args: {result.args}")
return iperf.start(server_only=server_only, client_only=client_only)
-def stop_iperf():
- args = ["pkill", "iperf"]
- args = " ".join(args)
- try:
- return subprocess.run(
- args,
- encoding="utf-8",
- shell=True,
- )
- except Exception:
- pass
+def stop_iperf(iperf_cmd):
+ """Stop the iperf process matching the iperf_cmd string."""
+ args = ["pgrep", "-x", "-f", iperf_cmd]
+ p = subprocess.Popen(
+ args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8"
+ )
+ stdout, _ = p.communicate()
+ for pid in stdout.split():
+ try:
+ subprocess.run(
+ f"kill -9 {pid}",
+ encoding="utf-8",
+ shell=True,
+ )
+ except Exception:
+ pass
if __name__ == "__main__":
Code Review / vpp.git / commitdiff