import os
import psutil
import time
+from vpp_qemu_utils import can_create_namespaces
def positive_int_or_default(default):
)
parser.add_argument("--extended", action="store_true", help="run extended tests")
+parser.add_argument(
+ "--skip-netns-tests",
+ action="store_true",
+ help="skip tests involving netns operations",
+)
parser.add_argument(
"--sanity", action="store_true", help="perform sanity vpp run before running tests"
else:
max_vpp_cpus = num_cpus
+if not config.skip_netns_tests:
+ if not can_create_namespaces():
+ config.skip_netns_tests = True
+
if __name__ == "__main__":
print("Provided arguments:")
for i in config.__dict__:
def create_test(test_name, test, ip_version, mtu):
"""Create and return a unittest method for a test."""
+ @unittest.skipIf(
+ config.skip_netns_tests, "netns not available or disabled from cli"
+ )
def test_func(self):
self.logger.debug(f"Starting unittest:{test_name}")
self.setUpTestToplogy(test=test, ip_version=ip_version)
def generate_vpp_interface_tests():
"""Generate unittests for testing vpp interfaces."""
+ if config.skip_netns_tests:
+ print("Skipping netns tests")
for test in tests:
for ip_version in test_config["ip_versions"]:
for mtu in test_config["mtus"]:
import sys
+def can_create_namespaces():
+ """Check if the environment allows creating the namespaces"""
+
+ try:
+ namespace = "vpp_chk_4212"
+ result = subprocess.run(["ip", "netns", "add", namespace], capture_output=True)
+ if result.returncode != 0:
+ return False
+ result = subprocess.run(["ip", "netns", "del", namespace], capture_output=True)
+ if result.returncode != 0:
+ return False
+ return True
+ except:
+ return False
+
+
def create_namespace(ns):
"""create one or more namespaces.
namespaces = ns
try:
for namespace in namespaces:
- subprocess.run(["ip", "netns", "add", namespace])
+ result = subprocess.run(["ip", "netns", "add", namespace])
+ if result.returncode != 0:
+ raise Exception(f"Error while creating namespace {namespace}")
except subprocess.CalledProcessError as e:
raise Exception("Error creating namespace:", e.output)
"""
try:
for namespace in namespaces:
- subprocess.run(["ip", "netns", "del", namespace], capture_output=True)
+ result = subprocess.run(
+ ["ip", "netns", "del", namespace], capture_output=True
+ )
+ if result.returncode != 0:
+ raise Exception(f"Error while deleting namespace {namespace}")
except subprocess.CalledProcessError as e:
raise Exception("Error deleting namespace:", e.output)