import time
import venv
import datetime
+import re
# Required Std. Path Variables
signal.signal(signal.SIGTERM, handler)
-def show_progress(stream):
+def show_progress(stream, exclude_pattern=None):
"""
Read lines from a subprocess stdout/stderr streams and write
to sys.stdout & the logfile
+
+ arguments:
+ stream - subprocess stdout or stderr data stream
+ exclude_pattern - lines matching this reg-ex will be excluded
+ from stdout.
"""
while True:
s = stream.readline()
# Filter the annoying SIGTERM signal from the output when VPP is
# terminated after a test run
if "SIGTERM" not in data:
- sys.stdout.write(data)
+ if exclude_pattern is not None:
+ if bool(re.search(exclude_pattern, data)) is False:
+ sys.stdout.write(data)
+ else:
+ sys.stdout.write(data)
logging.debug(data)
sys.stdout.flush()
stream.close()
p = Popen(
[script, test_name, kernel_image, test_data_dir, cpu_mask, mem],
stdout=PIPE,
- stderr=STDOUT,
cwd=ws_root,
)
- show_progress(p.stdout)
+ # Show only the test result without clobbering the stdout.
+ # The VM console displays VPP stderr & Linux IPv6 netdev change
+ # messages, which is logged by default and can be excluded.
+ exclude_pattern = r"vpp\[\d+\]:|ADDRCONF\(NETDEV_CHANGE\):"
+ show_progress(p.stdout, exclude_pattern)
post_vm_test_run()
f"--jobs={jobs}",
f"--log-dir={log_dir}",
f"--tmp-dir={log_dir}",
+ f"--cache-vpp-output",
]
if running_vpp:
args = args + [f"--use-running-vpp"]
mount -t 9p /dev/vpp9p ${WS_ROOT}
mount -t 9p tmp9p /tmp
modprobe -a vhost_net
-env SOCKET=1 SANITY=no \
-FAILED_DIR=${FAILED_DIR} RND_SEED=${RND_SEED} BR=${BR} \
-VENV_PATH=${VENV_PATH} TEST=${TEST} TEST_JOBS=${TEST_JOBS} \
-VPP_BUILD_DIR=${VPP_BUILD_DIR} VPP_BIN=${VPP_BIN} VPP_PLUGIN_PATH=${VPP_PLUGIN_PATH} \
-VPP_TEST_PLUGIN_PATH=${VPP_TEST_PLUGIN_PATH} VPP_INSTALL_PATH=${VPP_INSTALL_PATH} \
-LD_LIBRARY_PATH=${LD_LIBRARY_PATH} TEST_DATA_DIR=${TEST_DATA_DIR} INITRD=${INITRD} \
-bash -c "${WS_ROOT}/test/scripts/run.sh --filter=${TEST} --jobs=${TEST_JOBS} --failed-dir=${FAILED_DIR} \
---venv-dir=${VENV_PATH} --vpp-ws-dir=${WS_ROOT} --extended"
+${VENV_PATH}/bin/python3 ${WS_ROOT}/test/run_tests.py --filter=${TEST} --jobs=${TEST_JOBS} \
+--failed-dir=${FAILED_DIR} --venv-dir=${VENV_PATH} --vpp-ws-dir=${WS_ROOT} --extended \
+--vpp-tag=vpp_debug --cache-vpp-output
poweroff -f
_EOF_
+++ /dev/null
-#!/usr/bin/env python3
-import unittest
-from ipaddress import ip_interface
-from vpp_qemu_utils import create_namespace
-from vpp_iperf import VppIperf
-from framework import VppTestCase, VppTestRunner
-from config import config
-
-
-class TestTapQemu(VppTestCase):
- """Test Tap interfaces inside a QEMU VM.
-
- Start an iPerf connection stream between QEMU and VPP via
- tap v2 interfaces.
-
- Linux_ns1 -- iperf_client -- tap1 -- VPP-BD -- tap2 --
- -- iperfServer -- Linux_ns2
- """
-
- @classmethod
- def setUpClass(cls):
- super(TestTapQemu, cls).setUpClass()
-
- @classmethod
- def tearDownClass(cls):
- super(TestTapQemu, cls).tearDownClass()
-
- def setUp(self):
- """Perform test setup before running QEMU tests.
-
- 1. Create a namespace for the iPerf Server & Client.
- 2. Create 2 tap interfaces in VPP & add them to each namespace.
- 3. Add the tap interfaces to a bridge-domain.
- """
- super(TestTapQemu, self).setUp()
- self.client_namespace = "iprf_client_ns"
- self.server_namespace = "iprf_server_ns"
- self.client_ip4_prefix = "10.0.0.101/24"
- self.server_ip4_prefix = "10.0.0.102/24"
- create_namespace(self.client_namespace)
- create_namespace(self.server_namespace)
- tap1_if_idx = self.create_tap(
- 101, self.client_namespace, self.client_ip4_prefix
- )
- tap2_if_idx = self.create_tap(
- 102, self.server_namespace, self.server_ip4_prefix
- )
- self.l2_connect_interfaces(tap1_if_idx, tap2_if_idx)
-
- def create_tap(self, id, host_namespace, host_ip4_prefix):
- result = self.vapi.api(
- self.vapi.papi.tap_create_v2,
- {
- "id": id,
- "use_random_mac": True,
- "host_namespace_set": True,
- "host_namespace": host_namespace,
- "host_if_name_set": False,
- "host_bridge_set": False,
- "host_mac_addr_set": False,
- "host_ip4_prefix": ip_interface(host_ip4_prefix),
- "host_ip4_prefix_set": True,
- },
- )
- sw_if_index = result.sw_if_index
- self.vapi.api(
- self.vapi.papi.sw_interface_set_flags,
- {"sw_if_index": sw_if_index, "flags": 1},
- )
- return sw_if_index
-
- def dump_vpp_tap_interfaces(self):
- return self.vapi.api(self.vapi.papi.sw_interface_tap_v2_dump, {})
-
- def dump_bridge_domain_details(self):
- return self.vapi.api(self.vapi.papi.bridge_domain_dump, {"bd_id": 1})
-
- def l2_connect_interfaces(self, *sw_if_idxs):
- for if_idx in sw_if_idxs:
- self.vapi.api(
- self.vapi.papi.sw_interface_set_l2_bridge,
- {
- "rx_sw_if_index": if_idx,
- "bd_id": 1,
- "shg": 0,
- "port_type": 0,
- "enable": True,
- },
- )
-
- @unittest.skipUnless(config.extended, "part of extended tests")
- def test_tap_iperf(self):
- """Start an iperf connection stream between QEMU & VPP via tap."""
- iperf = VppIperf()
- iperf.client_ns = self.client_namespace
- iperf.server_ns = self.server_namespace
- iperf.server_ip = str(ip_interface(self.server_ip4_prefix).ip)
- iperf.start()
-
-
-if __name__ == "__main__":
- unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+import unittest
+from ipaddress import ip_address, ip_interface
+from vpp_qemu_utils import (
+ create_namespace,
+ delete_namespace,
+ create_host_interface,
+ delete_host_interfaces,
+ set_interface_mtu,
+ disable_interface_gso,
+ add_namespace_route,
+)
+from vpp_iperf import start_iperf, stop_iperf
+from framework import VppTestCase, VppTestRunner
+from config import config
+from vpp_papi import VppEnum
+import time
+import sys
+from vm_test_config import test_config
+
+#
+# Tests for:
+# - tapv2, tunv2 & af_packet_v2 & v3 interfaces.
+# - reads test config from the file vm_test_config.py
+# - Uses iPerf to send TCP/IP streams to VPP
+# - VPP ingress interface runs the iperf client
+# - VPP egress interface runs the iperf server
+# - Runs tests specified in the vm_test_config module and verifies that:
+# - TCP over IPv4 and IPv6 is enabled correctly for Bridged and Routed topologies.
+# sending jumbo frames (9000/9001 MTUs) with GSO/GRO is enabled correctly.
+# sending VPP buffer-sized frames(2048 MTU) with GSO/GRO is enabled correctly.
+# sending standard frames (1500 MTU) with GSO/GRO is enabled correctly.
+# sending smaller frames (512 MTU) with GSO/GRO is enabled correctly for IPv4
+# sending odd sized frames (9001, 2049 MTU) with GSO/GRO is enabled correctly.
+#
+
+
+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:
+ test_ids_to_run = []
+ for test_id in selection.split(","):
+ if "-" in test_id.strip():
+ start, end = map(int, test_id.split("-"))
+ test_ids_to_run.extend(list(range(start, end + 1)))
+ elif test_id.strip():
+ test_ids_to_run.append(int(test_id))
+ return test["id"] in test_ids_to_run
+
+
+# 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"]
+
+
+def create_test(test_name, test, ip_version, mtu):
+ """Create and return a unittest method for a test."""
+
+ @unittest.skipUnless(config.extended, "part of extended tests")
+ def test_func(self):
+ self.logger.debug(f"Starting unittest:{test_name}")
+ self.setUpTestToplogy(test=test, ip_version=ip_version)
+ result = self.set_interfaces_mtu(
+ mtu=mtu,
+ ip_version=ip_version,
+ vpp_interfaces=self.vpp_interfaces,
+ linux_interfaces=self.linux_interfaces,
+ )
+ # Start the Iperf server in dual stack mode & run iperf client
+ if result is True:
+ start_iperf(ip_version=6, server_only=True, logger=self.logger)
+ self.assertTrue(
+ start_iperf(
+ ip_version=ip_version,
+ server_ipv4_address=self.server_ip4_address,
+ server_ipv6_address=self.server_ip6_address,
+ client_only=True,
+ duration=2,
+ logger=self.logger,
+ )
+ )
+ else:
+ print(
+ f"Skipping test:{test_name} as mtu:{mtu} is "
+ f"invalid for TCP/IPv{ip_version}"
+ )
+
+ test_func.__name__ = test_name
+ return test_func
+
+
+def generate_vpp_interface_tests():
+ """Generate unittests for testing vpp interfaces."""
+ for test in tests:
+ for ip_version in test_config["ip_versions"]:
+ for mtu in test_config["mtus"]:
+ test_name = (
+ f"test_id_{test['id']}_"
+ + f"client_{test['client_if_type']}"
+ + f"_v{test['client_if_version']}_"
+ + f"gso_{test.get('client_if_gso', 0)}_"
+ + f"gro_{test.get('client_if_gro', 0)}_"
+ + f"to_server_{test['server_if_type']}"
+ + f"_v{test['server_if_version']}_"
+ + f"gso_{test.get('server_if_gso', 0)}_"
+ + f"gro_{test.get('server_if_gro', 0)}_"
+ + f"mtu_{mtu}_mode_{test['x_connect_mode']}_"
+ + f"tcp_ipv{ip_version}"
+ )
+ test_func = create_test(
+ test_name=test_name, test=test, ip_version=ip_version, mtu=mtu
+ )
+ setattr(TestVPPInterfacesQemu, test_name, test_func)
+
+
+class TestVPPInterfacesQemu(VppTestCase):
+ """Test VPP interfaces inside a QEMU VM for IPv4/v6.
+
+ Test Setup:
+ Linux_ns1--iperfClient--host-int1--vpp-af_packet-int1--VPP-BD
+ --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.
+
+ 1. Create Linux Namespaces for iPerf Client & Server.
+ 2. Create VPP iPerf client and server virtual interfaces.
+ 3. Enable desired vif features such as GSO & GRO.
+ 3. Cross-Connect interfaces in VPP using L2 or L3.
+ """
+ super(TestVPPInterfacesQemu, self).setUp()
+ client_if_type = test["client_if_type"]
+ server_if_type = test["server_if_type"]
+ client_if_version = test["client_if_version"]
+ server_if_version = test["server_if_version"]
+ x_connect_mode = test["x_connect_mode"]
+ # server ip4/ip6 addresses required by iperf client
+ server_ip4_prefix = (
+ layer2["server_ip4_prefix"]
+ if x_connect_mode == "L2"
+ else layer3["server_ip4_prefix"]
+ )
+ server_ip6_prefix = (
+ layer2["server_ip6_prefix"]
+ if x_connect_mode == "L2"
+ else layer3["server_ip6_prefix"]
+ )
+ self.server_ip4_address = str(ip_interface(server_ip4_prefix).ip)
+ self.server_ip6_address = str(ip_interface(server_ip6_prefix).ip)
+ # next-hop IP address on VPP for routing from client & server namespaces
+ vpp_client_prefix = (
+ layer3["vpp_client_ip4_prefix"]
+ if ip_version == 4
+ else layer3["vpp_client_ip6_prefix"]
+ )
+ vpp_client_nexthop = str(ip_interface(vpp_client_prefix).ip)
+ vpp_server_prefix = (
+ layer3["vpp_server_ip4_prefix"]
+ if ip_version == 4
+ else layer3["vpp_server_ip6_prefix"]
+ )
+ vpp_server_nexthop = str(ip_interface(vpp_server_prefix).ip)
+ create_namespace([client_namespace, server_namespace])
+ self.vpp_interfaces = []
+ self.linux_interfaces = []
+ enable_client_if_gso = test.get("client_if_gso", 0)
+ enable_server_if_gso = test.get("server_if_gso", 0)
+ enable_client_if_gro = test.get("client_if_gro", 0)
+ enable_server_if_gro = test.get("server_if_gro", 0)
+ ## Handle client interface 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,
+ layer2["client_ip4_prefix"]
+ if x_connect_mode == "L2"
+ else layer3["client_ip4_prefix"],
+ layer2["client_ip6_prefix"]
+ if x_connect_mode == "L2"
+ else layer3["client_ip6_prefix"],
+ )
+ self.ingress_if_idx = self.create_af_packet(
+ version=client_if_version,
+ host_if_name=af_packet_config["iprf_client_interface_on_vpp"],
+ enable_gso=enable_client_if_gso,
+ )
+ self.vpp_interfaces.append(self.ingress_if_idx)
+ self.linux_interfaces.append(
+ ["", af_packet_config["iprf_client_interface_on_vpp"]]
+ )
+ self.linux_interfaces.append(
+ [client_namespace, af_packet_config["iprf_client_interface_on_linux"]]
+ )
+ if enable_client_if_gso == 0:
+ disable_interface_gso(
+ "", af_packet_config["iprf_client_interface_on_vpp"]
+ )
+ disable_interface_gso(
+ client_namespace, af_packet_config["iprf_client_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,
+ ip_version=ip_version,
+ host_ip4_prefix=layer2["client_ip4_prefix"]
+ if x_connect_mode == "L2"
+ else layer3["client_ip4_prefix"],
+ host_ip6_prefix=layer2["client_ip6_prefix"]
+ if x_connect_mode == "L2"
+ else layer3["client_ip6_prefix"],
+ host_ip4_gw=vpp_client_nexthop
+ if x_connect_mode == "L3" and ip_version == 4
+ else None,
+ host_ip6_gw=vpp_client_nexthop
+ if x_connect_mode == "L3" and ip_version == 6
+ else None,
+ int_type=client_if_type,
+ host_if_name=f"{client_if_type}0",
+ enable_gso=enable_client_if_gso,
+ enable_gro=enable_client_if_gro,
+ )
+ self.vpp_interfaces.append(self.ingress_if_idx)
+ self.linux_interfaces.append([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")
+ else:
+ print(
+ f"Unsupported client interface type: {client_if_type} "
+ f"for test - ID={test['id']}"
+ )
+ sys.exit(1)
+
+ 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,
+ 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"],
+ enable_gso=enable_server_if_gso,
+ )
+ self.vpp_interfaces.append(self.egress_if_idx)
+ self.linux_interfaces.append(
+ ["", af_packet_config["iprf_server_interface_on_vpp"]]
+ )
+ self.linux_interfaces.append(
+ [server_namespace, af_packet_config["iprf_server_interface_on_linux"]]
+ )
+ if enable_server_if_gso == 0:
+ disable_interface_gso(
+ "", af_packet_config["iprf_server_interface_on_vpp"]
+ )
+ disable_interface_gso(
+ server_namespace, af_packet_config["iprf_server_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,
+ ip_version=ip_version,
+ host_ip4_prefix=layer2["server_ip4_prefix"]
+ if x_connect_mode == "L2"
+ else layer3["server_ip4_prefix"],
+ host_ip6_prefix=layer2["server_ip6_prefix"]
+ if x_connect_mode == "L2"
+ else layer3["server_ip6_prefix"],
+ int_type=server_if_type,
+ host_if_name=f"{server_if_type}0",
+ enable_gso=enable_server_if_gso,
+ enable_gro=enable_server_if_gro,
+ )
+ self.vpp_interfaces.append(self.egress_if_idx)
+ self.linux_interfaces.append([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")
+ else:
+ print(
+ f"Unsupported server interface type: {server_if_type} "
+ f"for test - ID={test['id']}"
+ )
+ sys.exit(1)
+
+ if x_connect_mode == "L2":
+ self.l2_connect_interfaces(1, self.ingress_if_idx, self.egress_if_idx)
+ elif x_connect_mode == "L3":
+ # L3 connect client & server side
+ vrf_id = layer3["ip4_vrf"] if ip_version == 4 else layer3["ip6_vrf"]
+ self.l3_connect_interfaces(
+ ip_version,
+ vrf_id,
+ (self.ingress_if_idx, vpp_client_prefix),
+ (self.egress_if_idx, vpp_server_prefix),
+ )
+ # Setup namespace routing
+ if ip_version == 4:
+ add_namespace_route(client_namespace, "0.0.0.0/0", vpp_client_nexthop)
+ add_namespace_route(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)
+ # Wait for Linux IPv6 stack to become ready
+ if ip_version == 6:
+ time.sleep(2)
+
+ def tearDown(self):
+ try:
+ self.vapi.tap_delete_v2(self.ingress_if_idx)
+ except Exception:
+ pass
+ try:
+ self.vapi.tap_delete_v2(self.egress_if_idx)
+ except Exception:
+ 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"]
+ )
+ 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"],
+ )
+ except Exception:
+ pass
+ try:
+ self.vapi.ip_table_add_del(is_add=0, table={"table_id": layer3["ip4_vrf"]})
+ except Exception:
+ pass
+ try:
+ self.vapi.ip_table_add_del(is_add=0, table={"table_id": layer3["ip6_vrf"]})
+ except Exception:
+ pass
+ try:
+ delete_namespace(
+ [
+ client_namespace,
+ server_namespace,
+ ]
+ )
+ except Exception:
+ pass
+ try:
+ stop_iperf()
+ except Exception:
+ pass
+
+ def create_af_packet(self, version, host_if_name, enable_gso=0):
+ """Create an af_packetv3 interface in VPP.
+
+ Parameters:
+ version -- 2 for af_packet_create_v2
+ -- 3 for af_packet_create_v3
+ host_if_name -- host interface name
+ enable_gso -- Enable GSO on the interface when True
+ """
+ af_packet_mode = VppEnum.vl_api_af_packet_mode_t
+ af_packet_interface_mode = af_packet_mode.AF_PACKET_API_MODE_ETHERNET
+ af_packet_flags = VppEnum.vl_api_af_packet_flags_t
+ af_packet_interface_flags = af_packet_flags.AF_PACKET_API_FLAG_QDISC_BYPASS
+ if enable_gso:
+ af_packet_interface_flags = (
+ af_packet_interface_flags | af_packet_flags.AF_PACKET_API_FLAG_CKSUM_GSO
+ )
+ if version == 2:
+ af_packet_interface_flags = (
+ af_packet_interface_flags | af_packet_flags.AF_PACKET_API_FLAG_VERSION_2
+ )
+ api_args = {
+ "use_random_hw_addr": True,
+ "host_if_name": host_if_name,
+ "flags": af_packet_interface_flags,
+ }
+ api_args["mode"] = af_packet_interface_mode
+ result = self.vapi.af_packet_create_v3(**api_args)
+ sw_if_index = result.sw_if_index
+ # Enable software GSO chunking when interface doesn't support GSO offload
+ if enable_gso == 0:
+ self.vapi.feature_gso_enable_disable(
+ sw_if_index=sw_if_index, enable_disable=1
+ )
+ else:
+ self.vapi.feature_gso_enable_disable(
+ sw_if_index=sw_if_index, enable_disable=0
+ )
+ self.vapi.sw_interface_set_flags(sw_if_index=sw_if_index, flags=1)
+ return sw_if_index
+
+ def create_tap_tun(
+ self,
+ id,
+ host_namespace,
+ ip_version,
+ host_ip4_prefix=None,
+ host_ip6_prefix=None,
+ host_ip4_gw=None,
+ host_ip6_gw=None,
+ int_type="tap",
+ host_if_name=None,
+ enable_gso=0,
+ enable_gro=0,
+ ):
+ """Create a tapv2 or tunv2 interface in VPP and attach to host.
+
+ Parameters:
+ id -- interface ID
+ host_namespace -- host namespace to attach the tap/tun interface to
+ ip_version -- 4 or 6
+ host_ip4_prefix -- ipv4 host interface address in CIDR notation
+ if ip_version=4
+ host_ip6_prefix -- ipv6 host interface address in CIDR notation
+ if ip_version=6
+ host_ip4_gw -- host IPv4 default gateway IP Address
+ host_ip6_gw -- host IPv6 default gateway IP address
+ int_type -- "tap" for tapv2 & "tun" for tunv2 interface
+ host_if_name -- host side interface name
+ enable_gso -- enable GSO
+ enable_gro -- enable GSO/GRO-Coalesce
+ """
+ TapFlags = VppEnum.vl_api_tap_flags_t
+ tap_flags = 0
+ if int_type == "tun":
+ tap_flags = TapFlags.TAP_API_FLAG_TUN
+ if enable_gro:
+ tap_flags = tap_flags | (
+ TapFlags.TAP_API_FLAG_GSO | TapFlags.TAP_API_FLAG_GRO_COALESCE
+ )
+ elif enable_gso:
+ tap_flags = tap_flags | TapFlags.TAP_API_FLAG_GSO
+ elif int_type == "tap":
+ if enable_gro:
+ tap_flags = (
+ TapFlags.TAP_API_FLAG_GSO | TapFlags.TAP_API_FLAG_GRO_COALESCE
+ )
+ elif enable_gso:
+ tap_flags = TapFlags.TAP_API_FLAG_GSO
+
+ api_args = {
+ "id": id,
+ "host_namespace_set": True,
+ "host_namespace": host_namespace,
+ "host_if_name_set": False,
+ "host_bridge_set": False,
+ "host_mac_addr_set": False,
+ }
+ if tap_flags != 0:
+ api_args["tap_flags"] = tap_flags
+ if ip_version == 4:
+ api_args["host_ip4_prefix"] = ip_interface(host_ip4_prefix)
+ api_args["host_ip4_prefix_set"] = True
+ if host_ip4_gw:
+ api_args["host_ip4_gw"] = ip_address(host_ip4_gw)
+ api_args["host_ip4_gw_set"] = True
+ if ip_version == 6:
+ api_args["host_ip6_prefix"] = ip_interface(host_ip6_prefix)
+ api_args["host_ip6_prefix_set"] = True
+ if host_ip6_gw:
+ api_args["host_ip6_gw"] = ip_address(host_ip6_gw)
+ api_args["host_ip6_gw_set"] = True
+ if host_if_name:
+ api_args["host_if_name"] = host_if_name
+ api_args["host_if_name_set"] = True
+
+ result = self.vapi.tap_create_v2(**api_args)
+ sw_if_index = result.sw_if_index
+ # Enable software GSO chunking when interface doesn't support GSO offload and
+ # GRO coalesce
+ if enable_gso == 0 and enable_gro == 0:
+ self.vapi.feature_gso_enable_disable(
+ sw_if_index=sw_if_index, enable_disable=1
+ )
+ else:
+ self.vapi.feature_gso_enable_disable(
+ sw_if_index=sw_if_index, enable_disable=0
+ )
+ # Admin up
+ self.vapi.sw_interface_set_flags(sw_if_index=sw_if_index, flags=1)
+ return sw_if_index
+
+ def dump_bridge_domain_details(self, bd_id):
+ return self.vapi.bridge_domain_dump(bd_id=bd_id)
+
+ def l2_connect_interfaces(self, bridge_id, *sw_if_idxs):
+ for if_idx in sw_if_idxs:
+ self.vapi.sw_interface_set_l2_bridge(
+ rx_sw_if_index=if_idx, bd_id=bridge_id, shg=0, port_type=0, enable=True
+ )
+
+ def l3_connect_interfaces(self, ip_version, vrf_id, *if_idx_ip_prefixes):
+ """Setup routing for (if_idx, ip_prefix) inside VPP.
+
+ arguments:
+ if_idx_ip_prefixes -- sequence of (if_idx, ip_prefix) tuples
+ ip_version -- 4 or 6
+ vrf_id -- vrf_id
+ """
+ is_ipv6 = 0 if ip_version == 4 else 1
+ self.vapi.ip_table_add_del(
+ is_add=1, table={"table_id": vrf_id, "is_ip6": is_ipv6}
+ )
+ for sw_if_index, ip_prefix in if_idx_ip_prefixes:
+ self.vapi.sw_interface_set_table(
+ sw_if_index=sw_if_index, is_ipv6=is_ipv6, vrf_id=vrf_id
+ )
+ self.vapi.sw_interface_add_del_address(
+ sw_if_index=sw_if_index, is_add=1, prefix=ip_interface(ip_prefix)
+ )
+
+ def set_interfaces_mtu(self, mtu, ip_version, **kwargs):
+ """Set MTUs on VPP and Linux interfaces.
+
+ arguments --
+ mtu -- mtu value
+ ip_version - 4 or 6
+ kwargs['vpp_interfaces'] -- list of vpp interface if indexes
+ kwargs['linux_interfaces'] -- list of tuples (namespace, interface_names)
+ return True if mtu is set, else False
+ """
+ vpp_interfaces = kwargs.get("vpp_interfaces")
+ linux_interfaces = kwargs.get("linux_interfaces")
+ # IPv6 on Linux requires an MTU value >=1280
+ if (ip_version == 6 and mtu >= 1280) or ip_version == 4:
+ for sw_if_idx in vpp_interfaces:
+ self.vapi.sw_interface_set_mtu(
+ sw_if_index=sw_if_idx, mtu=[mtu, 0, 0, 0]
+ )
+ for namespace, interface_name in linux_interfaces:
+ set_interface_mtu(
+ namespace=namespace,
+ interface=interface_name,
+ mtu=mtu,
+ logger=self.logger,
+ )
+ return True
+ else:
+ return False
+
+
+generate_vpp_interface_tests()
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)
--- /dev/null
+#!/usr/bin/env python3
+
+## Virtual Interface Test configuration for VM tests ##
+
+test_config = {
+ "client_namespace": "iprf_client_ns",
+ "server_namespace": "iprf_server_ns",
+ "mtus": [9001, 9000, 2048, 2049, 1500, 512],
+ "ip_versions": [4, 6],
+ "af_packet": {
+ "iprf_client_interface_on_linux": "hostintclient",
+ "iprf_server_interface_on_linux": "hostintserver",
+ "iprf_client_interface_on_vpp": "vppclientout",
+ "iprf_server_interface_on_vpp": "vppserverout",
+ },
+ "L2": {
+ "client_ip4_prefix": "10.0.0.101/24",
+ "server_ip4_prefix": "10.0.0.102/24",
+ "client_ip6_prefix": "2001:1::1/64",
+ "server_ip6_prefix": "2001:1::2/64",
+ },
+ "L3": {
+ "client_ip4_prefix": "10.0.0.101/24",
+ "vpp_client_ip4_prefix": "10.0.0.102/24",
+ "server_ip4_prefix": "10.0.1.102/24",
+ "vpp_server_ip4_prefix": "10.0.1.101/24",
+ "ip4_vrf": 1,
+ "client_ip6_prefix": "2001:1::1/64",
+ "vpp_client_ip6_prefix": "2001:1::2/64",
+ "server_ip6_prefix": "2001:2::2/64",
+ "vpp_server_ip6_prefix": "2001:2::1/64",
+ "ip6_vrf": 2,
+ },
+ # Test Filter
+ # Comma separated test id's or range(s) of test id's to run (default=all)
+ # e.g. "1,3-4,19-23" runs tests with ID's 1, 3, 4, 19, 20, 21, 22 & 23 only
+ "tests_to_run": "",
+ "tests": [
+ {
+ "id": 1,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "client_if_gro": 0,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "server_if_gro": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 2,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 1,
+ "client_if_gro": 0,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "server_if_gro": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 3,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "client_if_gro": 0,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 1,
+ "server_if_gro": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 4,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "client_if_gro": 1,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "server_if_gro": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 5,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "client_if_gro": 0,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "server_if_gro": 1,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 6,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 1,
+ "client_if_gro": 0,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "server_if_gro": 1,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 7,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "client_if_gro": 1,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 1,
+ "server_if_gro": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 8,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 1,
+ "client_if_gro": 0,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 1,
+ "server_if_gro": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 9,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "client_if_gro": 1,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "server_if_gro": 1,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 10,
+ "client_if_type": "tap",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "client_if_gro": 0,
+ "server_if_type": "tun",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "server_if_gro": 1,
+ "x_connect_mode": "L3",
+ },
+ {
+ "id": 11,
+ "client_if_type": "tun",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "client_if_gro": 0,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "server_if_gro": 1,
+ "x_connect_mode": "L3",
+ },
+ {
+ "id": 12,
+ "client_if_type": "af_packet",
+ "client_if_version": 3,
+ "client_if_gso": 0,
+ "server_if_type": "af_packet",
+ "server_if_version": 3,
+ "server_if_gso": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 13,
+ "client_if_type": "af_packet",
+ "client_if_version": 3,
+ "client_if_gso": 0,
+ "server_if_type": "af_packet",
+ "server_if_version": 3,
+ "server_if_gso": 0,
+ "x_connect_mode": "L3",
+ },
+ {
+ "id": 14,
+ "client_if_type": "af_packet",
+ "client_if_version": 3,
+ "client_if_gso": 1,
+ "server_if_type": "af_packet",
+ "server_if_version": 3,
+ "server_if_gso": 1,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 15,
+ "client_if_type": "af_packet",
+ "client_if_version": 3,
+ "client_if_gso": 1,
+ "server_if_type": "af_packet",
+ "server_if_version": 3,
+ "server_if_gso": 1,
+ "x_connect_mode": "L3",
+ },
+ {
+ "id": 16,
+ "client_if_type": "af_packet",
+ "client_if_version": 3,
+ "client_if_gso": 1,
+ "server_if_type": "af_packet",
+ "server_if_version": 3,
+ "server_if_gso": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 17,
+ "client_if_type": "af_packet",
+ "client_if_version": 2,
+ "client_if_gso": 1,
+ "server_if_type": "tap",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 18,
+ "client_if_type": "af_packet",
+ "client_if_version": 2,
+ "client_if_gso": 1,
+ "server_if_type": "tun",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "x_connect_mode": "L3",
+ },
+ {
+ "id": 19,
+ "client_if_type": "af_packet",
+ "client_if_version": 2,
+ "client_if_gso": 1,
+ "server_if_type": "af_packet",
+ "server_if_version": 2,
+ "server_if_gso": 1,
+ "x_connect_mode": "L3",
+ },
+ {
+ "id": 20,
+ "client_if_type": "af_packet",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "server_if_type": "af_packet",
+ "server_if_version": 2,
+ "server_if_gso": 0,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 21,
+ "client_if_type": "af_packet",
+ "client_if_version": 2,
+ "client_if_gso": 0,
+ "server_if_type": "tun",
+ "server_if_version": 2,
+ "server_if_gro": 1,
+ "x_connect_mode": "L3",
+ },
+ {
+ "id": 22,
+ "client_if_type": "af_packet",
+ "client_if_version": 3,
+ "client_if_gso": 0,
+ "server_if_type": "af_packet",
+ "server_if_version": 2,
+ "server_if_gso": 1,
+ "x_connect_mode": "L2",
+ },
+ {
+ "id": 23,
+ "client_if_type": "af_packet",
+ "client_if_version": 2,
+ "client_if_gso": 1,
+ "server_if_type": "af_packet",
+ "server_if_version": 2,
+ "server_if_gso": 1,
+ "x_connect_mode": "L2",
+ },
+ ],
+}
import subprocess
import os
+import sys
class VppIperf:
iperf.server_args='-p 5202'
"""
- def __init__(self, server_ns=None, client_ns=None, server_ip=None):
+ def __init__(self, server_ns=None, client_ns=None, server_ip=None, logger=None):
self.server_ns = server_ns
self.client_ns = client_ns
self.server_ip = server_ip
self.duration = 10
self.client_args = ""
self.server_args = ""
+ self.logger = logger
# Set the iperf executable
self.iperf = os.path.join(os.getenv("TEST_DATA_DIR") or "/", "usr/bin/iperf")
)
def start_iperf_server(self):
- print("Starting iPerf Server Daemon in Namespace ", self.server_ns)
args = [
"ip",
"netns",
self.iperf,
"-s",
"-D",
- "-B",
- self.server_ip,
]
args.extend(self.server_args.split())
+ args = " ".join(args)
try:
- subprocess.run(
+ return subprocess.run(
args,
- stderr=subprocess.STDOUT,
timeout=self.duration + 5,
encoding="utf-8",
+ shell=True,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
)
except subprocess.TimeoutExpired as e:
raise Exception("Error: Timeout expired for iPerf", e.output)
def start_iperf_client(self):
- print("Starting iPerf Client in Namespace ", self.client_ns)
args = [
"ip",
"netns",
str(self.duration),
]
args.extend(self.client_args.split())
+ args = " ".join(args)
try:
- subprocess.run(
+ return subprocess.run(
args,
- stderr=subprocess.STDOUT,
timeout=self.duration + 5,
encoding="utf-8",
+ capture_output=True,
+ shell=True,
)
except subprocess.TimeoutExpired as e:
raise Exception("Error: Timeout expired for iPerf", e.output)
- def start(self):
- """Run iPerf and return True if successful"""
+ def start(self, server_only=False, client_only=False):
+ """Runs iPerf.
+
+ Starts the iperf server daemon & runs the iperf client.
+ arguments:-
+ server_only -- start the iperf server daemon only
+ client_only -- run the iperf client only
+ Return True if we have no errors in iPerf client, else False.
+ """
self.ensure_init()
- try:
+ if not client_only:
self.start_iperf_server()
- except Exception as e:
- subprocess.run(["pkill", "iperf"])
- raise Exception("Error starting iPerf Server", e)
+ if not server_only:
+ result = self.start_iperf_client()
+ self.logger.debug(f"Iperf client args: {result.args}")
+ self.logger.debug(result.stdout)
+ if result.stderr:
+ self.logger.error(
+ f"Error starting Iperf Client in Namespace: {self.client_ns}"
+ )
+ self.logger.error(f"Iperf client args: {result.args}")
+ self.logger.error(f"Iperf client has errors: {result.stderr}")
+ return False
+ else:
+ return True
- try:
- self.start_iperf_client()
- except Exception as e:
- raise Exception("Error starting iPerf Client", e)
- subprocess.run(["pkill", "iperf"])
+
+## Functions to start and stop iPerf using the iPerf object
+def start_iperf(
+ ip_version,
+ client_ns="iprf_client_ns",
+ server_ns="iprf_server_ns",
+ server_ipv4_address="10.0.0.102",
+ server_ipv6_address="2001:1::2",
+ client_args="",
+ server_args="",
+ duration=10,
+ server_only=False,
+ client_only=False,
+ logger=None,
+):
+ """Start an iperf connection stream using the iPerf object.
+
+ Starts iPerf an connection stream between an iPerf client in the
+ client namespace (client_ns) and a server in another
+ namespace (server_ns).
+ Parameters:
+ ip_version - 4 or 6
+ client_ns - iPerf client namespace
+ server_ns - iPerf server namespace
+ server_ipv4_address - ipv4 address of the server, if ip_version=4
+ server_ipv6_address - ipv6 address of the server, if ip_version=6
+ client_args - Additonal iperf control arguments to be passed
+ to the iperf client from the test (str)
+ server_args - Additonal iperf control arguments to be passed
+ to the iperf server from the test (str)
+ duration - Iperf duration in seconds
+ server_only - start iperf server only
+ client_only - start the iperf client only
+ logger - test logger
+ """
+ if ip_version == 4:
+ iperf_server_ip = server_ipv4_address
+ elif ip_version == 6:
+ iperf_server_ip = server_ipv6_address
+ client_args = "-V" + " " + client_args
+ server_args = "-V" + " " + server_args
+ iperf = VppIperf()
+ iperf.client_ns = client_ns
+ iperf.server_ns = server_ns
+ iperf.server_ip = iperf_server_ip
+ iperf.client_args = client_args
+ iperf.server_args = server_args
+ iperf.duration = duration
+ iperf.logger = logger
+ 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
if __name__ == "__main__":
# Utility functions for QEMU tests ##
import subprocess
+import sys
def create_namespace(ns):
+ """create one or more namespaces.
+
+ arguments:
+ ns -- a string value or an iterable of namespace names
+ """
+ if isinstance(ns, str):
+ namespaces = [ns]
+ else:
+ namespaces = ns
try:
- subprocess.run(["ip", "netns", "add", ns])
+ for namespace in namespaces:
+ subprocess.run(["ip", "netns", "add", namespace])
except subprocess.CalledProcessError as e:
raise Exception("Error creating namespace:", e.output)
+def add_namespace_route(ns, prefix, gw_ip):
+ """Add a route to a namespace.
+
+ arguments:
+ ns -- namespace string value
+ prefix -- NETWORK/MASK or "default"
+ gw_ip -- Gateway IP
+ """
+ try:
+ subprocess.run(
+ ["ip", "netns", "exec", ns, "ip", "route", "add", prefix, "via", gw_ip],
+ capture_output=True,
+ )
+ except subprocess.CalledProcessError as e:
+ raise Exception("Error adding route to namespace:", e.output)
+
+
+def delete_host_interfaces(*host_interface_names):
+ """Delete host interfaces.
+
+ arguments:
+ host_interface_names - sequence of host interface names to be deleted
+ """
+ for host_interface_name in host_interface_names:
+ try:
+ subprocess.run(
+ ["ip", "link", "del", host_interface_name], capture_output=True
+ )
+ except subprocess.CalledProcessError as e:
+ raise Exception("Error deleting host interface:", e.output)
+
+
+def create_host_interface(
+ host_interface_name, vpp_interface_name, host_namespace, *host_ip_prefixes
+):
+ """Create a host interface of type veth.
+
+ arguments:
+ host_interface_name -- name of the veth interface on the host side
+ vpp_interface_name -- name of the veth interface on the VPP side
+ host_namespace -- host namespace into which the host_interface needs to be set
+ host_ip_prefixes -- a sequence of ip/prefix-lengths to be set
+ on the host_interface
+ """
+ try:
+ process = subprocess.run(
+ [
+ "ip",
+ "link",
+ "add",
+ "name",
+ vpp_interface_name,
+ "type",
+ "veth",
+ "peer",
+ "name",
+ host_interface_name,
+ ],
+ capture_output=True,
+ )
+ if process.returncode != 0:
+ print(f"Error creating host interface: {process.stderr}")
+ sys.exit(1)
+
+ process = subprocess.run(
+ ["ip", "link", "set", host_interface_name, "netns", host_namespace],
+ capture_output=True,
+ )
+ if process.returncode != 0:
+ print(f"Error setting host interface namespace: {process.stderr}")
+ sys.exit(1)
+
+ process = subprocess.run(
+ ["ip", "link", "set", "dev", vpp_interface_name, "up"], capture_output=True
+ )
+ if process.returncode != 0:
+ print(f"Error bringing up the host interface: {process.stderr}")
+ sys.exit(1)
+
+ process = subprocess.run(
+ [
+ "ip",
+ "netns",
+ "exec",
+ host_namespace,
+ "ip",
+ "link",
+ "set",
+ "dev",
+ host_interface_name,
+ "up",
+ ],
+ capture_output=True,
+ )
+ if process.returncode != 0:
+ print(
+ f"Error bringing up the host interface in namespace: "
+ f"{process.stderr}"
+ )
+ sys.exit(1)
+
+ for host_ip_prefix in host_ip_prefixes:
+ process = subprocess.run(
+ [
+ "ip",
+ "netns",
+ "exec",
+ host_namespace,
+ "ip",
+ "addr",
+ "add",
+ host_ip_prefix,
+ "dev",
+ host_interface_name,
+ ],
+ capture_output=True,
+ )
+ if process.returncode != 0:
+ print(
+ f"Error setting ip prefix on the host interface: "
+ f"{process.stderr}"
+ )
+ sys.exit(1)
+ except subprocess.CalledProcessError as e:
+ raise Exception("Error adding route to namespace:", e.output)
+
+
+def set_interface_mtu(namespace, interface, mtu, logger):
+ """set an mtu number on a linux device interface."""
+ args = ["ip", "link", "set", "mtu", str(mtu), "dev", interface]
+ if namespace:
+ args = ["ip", "netns", "exec", namespace] + args
+ try:
+ logger.debug(
+ f"Setting mtu:{mtu} on linux interface:{interface} "
+ f"in namespace:{namespace}"
+ )
+ subprocess.run(args)
+ except subprocess.CalledProcessError as e:
+ raise Exception("Error updating mtu:", e.output)
+
+
+def enable_interface_gso(namespace, interface):
+ """enable gso offload on a linux device interface."""
+ args = ["ethtool", "-K", interface, "rx", "on", "tx", "on"]
+ if namespace:
+ args = ["ip", "netns", "exec", namespace] + args
+ try:
+ process = subprocess.run(args, capture_output=True)
+ if process.returncode != 0:
+ print(
+ f"Error enabling GSO offload on linux device interface: "
+ f"{process.stderr}"
+ )
+ sys.exit(1)
+ except subprocess.CalledProcessError as e:
+ raise Exception("Error enabling gso:", e.output)
+
+
+def disable_interface_gso(namespace, interface):
+ """disable gso offload on a linux device interface."""
+ args = ["ethtool", "-K", interface, "rx", "off", "tx", "off"]
+ if namespace:
+ args = ["ip", "netns", "exec", namespace] + args
+ try:
+ process = subprocess.run(args, capture_output=True)
+ if process.returncode != 0:
+ print(
+ f"Error disabling GSO offload on linux device interface: "
+ f"{process.stderr}"
+ )
+ sys.exit(1)
+ except subprocess.CalledProcessError as e:
+ raise Exception("Error disabling gso:", e.output)
+
+
+def delete_namespace(namespaces):
+ """delete one or more namespaces.
+
+ arguments:
+ namespaces -- a list of namespace names
+ """
+ try:
+ for namespace in namespaces:
+ subprocess.run(["ip", "netns", "del", namespace], capture_output=True)
+ except subprocess.CalledProcessError as e:
+ raise Exception("Error deleting namespace:", e.output)
+
+
def list_namespace(ns):
"""List the IP address of a namespace"""
try:
Code Review / vpp.git / commitdiff