test/test_vm_vpp_interfaces.py

#!/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,
    libmemif_test_app,
)
from vpp_iperf import start_iperf, stop_iperf
from framework import VppTestCase
from asfframework import VppTestRunner, tag_fixme_debian11, is_distro_debian11
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 & memif 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.skipIf(
        is_distro_debian11, "FIXME intermittent test failures on debian11 distro"
    )
    @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)
        result = self.set_interfaces_mtu(
            mtu=mtu,
            ip_version=ip_version,
            vpp_interfaces=self.vpp_interfaces,
            linux_interfaces=self.linux_interfaces,
        )
        if "memif" in self.if_types:
            self.logger.debug("Starting libmemif test_app for memif test")
            self.memif_process = libmemif_test_app(
                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)
            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."""
    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"]:
                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"checksum_{test.get('client_if_checksum_offload', 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"checksum_{test.get('server_if_checksum_offload', 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)


@tag_fixme_debian11
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()
        # 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.
        client_if_types = test["client_if_type"].split(",")
        server_if_types = test["server_if_type"].split(",")
        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])
        # IPerf client & server ingress/egress interface indexes in VPP
        self.tap_interfaces = []
        self.memif_interfaces = []
        self.ingress_if_idxes = []
        self.egress_if_idxes = []
        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)
        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
        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,
                    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.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.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,
                    enable_checksum_offload=enable_client_if_checksum_offload,
                )
                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"])
                # Seeing TCP timeouts if tx=on & rx=on Linux tap & tun interfaces
                disable_interface_gso(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
                )
                self.memif_interfaces.append(self.ingress_if_idx)
                self.ingress_if_idxes.append(self.ingress_if_idx)
                self.vpp_interfaces.append(self.ingress_if_idx)
            else:
                print(
                    f"Unsupported client interface type: {client_if_type} "
                    f"for test - ID={test['id']}"
                )
                sys.exit(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,
                    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.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.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,
                    enable_checksum_offload=enable_server_if_checksum_offload,
                )
                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"])
                # Seeing TCP timeouts if tx=on & rx=on Linux tap & tun interfaces
                disable_interface_gso(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
                )
                self.memif_interfaces.append(self.egress_if_idx)
                self.egress_if_idxes.append(self.egress_if_idx)
                self.vpp_interfaces.append(self.egress_if_idx)
            else:
                print(
                    f"Unsupported server interface type: {server_if_type} "
                    f"for test - ID={test['id']}"
                )
                sys.exit(1)
        self.if_types = set(client_if_types).union(set(server_if_types))
        # for memif testing: tapv2, memif & libmemif_app are connected
        if "memif" not in self.if_types:
            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)
        else:
            # connect: ingress tap & memif & egress tap and memif interfaces
            if x_connect_mode == "L2":
                self.l2_connect_interfaces(1, *self.ingress_if_idxes)
                self.l2_connect_interfaces(2, *self.egress_if_idxes)
        # Wait for Linux IPv6 stack to become ready
        if ip_version == 6:
            time.sleep(2)

    def tearDown(self):
        try:
            for interface_if_idx in self.tap_interfaces:
                self.vapi.tap_delete_v2(sw_if_index=interface_if_idx)
        except Exception:
            pass
        try:
            for interface_if_idx in self.memif_interfaces:
                self.vapi.memif_delete(sw_if_index=interface_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:
            self.vapi.bridge_domain_add_del_v2(bd_id=1, is_add=0)
            self.vapi.bridge_domain_add_del_v2(bd_id=2, is_add=0)
        except Exception:
            pass
        try:
            delete_namespace(
                [
                    client_namespace,
                    server_namespace,
                ]
            )
        except Exception:
            pass
        try:
            stop_iperf()
        except Exception:
            pass
        try:
            if self.memif_process:
                self.memif_process.terminate()
                self.memif_process.join()
        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,
        enable_checksum_offload=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
        enable_checksum_offload -- enable checksum offload without gso
        """
        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 enable_checksum_offload:
                tap_flags = tap_flags | TapFlags.TAP_API_FLAG_CSUM_OFFLOAD
        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
            elif enable_checksum_offload:
                tap_flags = tap_flags | TapFlags.TAP_API_FLAG_CSUM_OFFLOAD

        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 create_memif(self, memif_id, mode):
        """Create memif interface in VPP.

        Parameters:
        memif_id: A unique ID for the memif interface
        mode: 0 = ethernet, 1 = ip, 2 = punt/inject
        """
        # create memif interface with role=0 (i.e. master)
        result = self.vapi.memif_create_v2(
            role=0, mode=mode, id=memif_id, buffer_size=9216
        )
        sw_if_index = result.sw_if_index
        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)