test/test_vm_vpp_interfaces.py

   1 #!/usr/bin/env python3
   2 import unittest
   3 from ipaddress import ip_address, ip_interface
   4 from vpp_qemu_utils import (
   5     create_namespace,
   6     delete_namespace,
   7     create_host_interface,
   8     delete_host_interfaces,
   9     set_interface_mtu,
  10     disable_interface_gso,
  11     add_namespace_route,
  12 )
  13 from vpp_iperf import start_iperf, stop_iperf
  14 from framework import VppTestCase
  15 from asfframework import VppTestRunner, tag_fixme_debian11, is_distro_debian11
  16 from config import config
  17 from vpp_papi import VppEnum
  18 import time
  19 import sys
  20 from vm_test_config import test_config
  21
  22 #
  23 # Tests for:
  24 # - tapv2, tunv2 & af_packet_v2 & v3 interfaces.
  25 # - reads test config from the file vm_test_config.py
  26 # - Uses iPerf to send TCP/IP streams to VPP
  27 #   - VPP ingress interface runs the iperf client
  28 #   - VPP egress interface runs the iperf server
  29 # - Runs tests specified in the vm_test_config module and verifies that:
  30 #   - TCP over IPv4 and IPv6 is enabled correctly for Bridged and Routed topologies.
  31 #     sending jumbo frames (9000/9001 MTUs) with GSO/GRO is enabled correctly.
  32 #     sending VPP buffer-sized frames(2048 MTU) with GSO/GRO is enabled correctly.
  33 #     sending standard frames (1500 MTU) with GSO/GRO is enabled correctly.
  34 #     sending smaller frames (512 MTU) with GSO/GRO is enabled correctly for IPv4
  35 #     sending odd sized frames (9001, 2049 MTU) with GSO/GRO is enabled correctly.
  36 #
  37
  38
  39 def filter_tests(test):
  40     """Filter test IDs to include only those selected to run."""
  41     selection = test_config["tests_to_run"]
  42     if not selection or selection == " ":
  43         return True
  44     else:
  45         test_ids_to_run = []
  46         for test_id in selection.split(","):
  47             if "-" in test_id.strip():
  48                 start, end = map(int, test_id.split("-"))
  49                 test_ids_to_run.extend(list(range(start, end + 1)))
  50             elif test_id.strip():
  51                 test_ids_to_run.append(int(test_id))
  52         return test["id"] in test_ids_to_run
  53
  54
  55 # Test Config variables
  56 client_namespace = test_config["client_namespace"]
  57 server_namespace = test_config["server_namespace"]
  58 tests = filter(filter_tests, test_config["tests"])
  59 af_packet_config = test_config["af_packet"]
  60 layer2 = test_config["L2"]
  61 layer3 = test_config["L3"]
  62
  63
  64 def create_test(test_name, test, ip_version, mtu):
  65     """Create and return a unittest method for a test."""
  66
  67     @unittest.skipIf(
  68         is_distro_debian11, "FIXME intermittent test failures on debian11 distro"
  69     )
  70     @unittest.skipIf(
  71         config.skip_netns_tests, "netns not available or disabled from cli"
  72     )
  73     def test_func(self):
  74         self.logger.debug(f"Starting unittest:{test_name}")
  75         self.setUpTestToplogy(test=test, ip_version=ip_version)
  76         result = self.set_interfaces_mtu(
  77             mtu=mtu,
  78             ip_version=ip_version,
  79             vpp_interfaces=self.vpp_interfaces,
  80             linux_interfaces=self.linux_interfaces,
  81         )
  82         # Start the Iperf server in dual stack mode & run iperf client
  83         if result is True:
  84             start_iperf(ip_version=6, server_only=True, logger=self.logger)
  85             self.assertTrue(
  86                 start_iperf(
  87                     ip_version=ip_version,
  88                     server_ipv4_address=self.server_ip4_address,
  89                     server_ipv6_address=self.server_ip6_address,
  90                     client_only=True,
  91                     duration=2,
  92                     logger=self.logger,
  93                 )
  94             )
  95         else:
  96             print(
  97                 f"Skipping test:{test_name} as mtu:{mtu} is "
  98                 f"invalid for TCP/IPv{ip_version}"
  99             )
 100
 101     test_func.__name__ = test_name
 102     return test_func
 103
 104
 105 def generate_vpp_interface_tests():
 106     """Generate unittests for testing vpp interfaces."""
 107     if config.skip_netns_tests:
 108         print("Skipping netns tests")
 109     for test in tests:
 110         for ip_version in test_config["ip_versions"]:
 111             for mtu in test_config["mtus"]:
 112                 test_name = (
 113                     f"test_id_{test['id']}_"
 114                     + f"client_{test['client_if_type']}"
 115                     + f"_v{test['client_if_version']}_"
 116                     + f"gso_{test.get('client_if_gso', 0)}_"
 117                     + f"gro_{test.get('client_if_gro', 0)}_"
 118                     + f"checksum_{test.get('client_if_checksum_offload', 0)}_"
 119                     + f"to_server_{test['server_if_type']}"
 120                     + f"_v{test['server_if_version']}_"
 121                     + f"gso_{test.get('server_if_gso', 0)}_"
 122                     + f"gro_{test.get('server_if_gro', 0)}_"
 123                     + f"checksum_{test.get('server_if_checksum_offload', 0)}_"
 124                     + f"mtu_{mtu}_mode_{test['x_connect_mode']}_"
 125                     + f"tcp_ipv{ip_version}"
 126                 )
 127                 test_func = create_test(
 128                     test_name=test_name, test=test, ip_version=ip_version, mtu=mtu
 129                 )
 130                 setattr(TestVPPInterfacesQemu, test_name, test_func)
 131
 132
 133 @tag_fixme_debian11
 134 class TestVPPInterfacesQemu(VppTestCase):
 135     """Test VPP interfaces inside a QEMU VM for IPv4/v6.
 136
 137     Test Setup:
 138     Linux_ns1--iperfClient--host-int1--vpp-af_packet-int1--VPP-BD
 139              --vppaf_packet_int2--host-int2--iperfServer--Linux_ns2
 140     """
 141
 142     @classmethod
 143     def setUpClass(cls):
 144         super(TestVPPInterfacesQemu, cls).setUpClass()
 145
 146     @classmethod
 147     def tearDownClass(cls):
 148         super(TestVPPInterfacesQemu, cls).tearDownClass()
 149
 150     def setUpTestToplogy(self, test, ip_version):
 151         """Setup the test topology.
 152
 153         1. Create Linux Namespaces for iPerf Client & Server.
 154         2. Create VPP iPerf client and server virtual interfaces.
 155         3. Enable desired vif features such as GSO & GRO.
 156         3. Cross-Connect interfaces in VPP using L2 or L3.
 157         """
 158         super(TestVPPInterfacesQemu, self).setUp()
 159         client_if_type = test["client_if_type"]
 160         server_if_type = test["server_if_type"]
 161         client_if_version = test["client_if_version"]
 162         server_if_version = test["server_if_version"]
 163         x_connect_mode = test["x_connect_mode"]
 164         # server ip4/ip6 addresses required by iperf client
 165         server_ip4_prefix = (
 166             layer2["server_ip4_prefix"]
 167             if x_connect_mode == "L2"
 168             else layer3["server_ip4_prefix"]
 169         )
 170         server_ip6_prefix = (
 171             layer2["server_ip6_prefix"]
 172             if x_connect_mode == "L2"
 173             else layer3["server_ip6_prefix"]
 174         )
 175         self.server_ip4_address = str(ip_interface(server_ip4_prefix).ip)
 176         self.server_ip6_address = str(ip_interface(server_ip6_prefix).ip)
 177         # next-hop IP address on VPP for routing from client & server namespaces
 178         vpp_client_prefix = (
 179             layer3["vpp_client_ip4_prefix"]
 180             if ip_version == 4
 181             else layer3["vpp_client_ip6_prefix"]
 182         )
 183         vpp_client_nexthop = str(ip_interface(vpp_client_prefix).ip)
 184         vpp_server_prefix = (
 185             layer3["vpp_server_ip4_prefix"]
 186             if ip_version == 4
 187             else layer3["vpp_server_ip6_prefix"]
 188         )
 189         vpp_server_nexthop = str(ip_interface(vpp_server_prefix).ip)
 190         create_namespace([client_namespace, server_namespace])
 191         self.vpp_interfaces = []
 192         self.linux_interfaces = []
 193         enable_client_if_gso = test.get("client_if_gso", 0)
 194         enable_server_if_gso = test.get("server_if_gso", 0)
 195         enable_client_if_gro = test.get("client_if_gro", 0)
 196         enable_server_if_gro = test.get("server_if_gro", 0)
 197         enable_client_if_checksum_offload = test.get("client_if_checksum_offload", 0)
 198         enable_server_if_checksum_offload = test.get("server_if_checksum_offload", 0)
 199         ## Handle client interface types
 200         if client_if_type == "af_packet":
 201             create_host_interface(
 202                 af_packet_config["iprf_client_interface_on_linux"],
 203                 af_packet_config["iprf_client_interface_on_vpp"],
 204                 client_namespace,
 205                 layer2["client_ip4_prefix"]
 206                 if x_connect_mode == "L2"
 207                 else layer3["client_ip4_prefix"],
 208                 layer2["client_ip6_prefix"]
 209                 if x_connect_mode == "L2"
 210                 else layer3["client_ip6_prefix"],
 211             )
 212             self.ingress_if_idx = self.create_af_packet(
 213                 version=client_if_version,
 214                 host_if_name=af_packet_config["iprf_client_interface_on_vpp"],
 215                 enable_gso=enable_client_if_gso,
 216             )
 217             self.vpp_interfaces.append(self.ingress_if_idx)
 218             self.linux_interfaces.append(
 219                 ["", af_packet_config["iprf_client_interface_on_vpp"]]
 220             )
 221             self.linux_interfaces.append(
 222                 [client_namespace, af_packet_config["iprf_client_interface_on_linux"]]
 223             )
 224             if enable_client_if_gso == 0:
 225                 disable_interface_gso(
 226                     "", af_packet_config["iprf_client_interface_on_vpp"]
 227                 )
 228                 disable_interface_gso(
 229                     client_namespace, af_packet_config["iprf_client_interface_on_linux"]
 230                 )
 231         elif client_if_type == "tap" or client_if_type == "tun":
 232             self.ingress_if_idx = self.create_tap_tun(
 233                 id=101,
 234                 host_namespace=client_namespace,
 235                 ip_version=ip_version,
 236                 host_ip4_prefix=layer2["client_ip4_prefix"]
 237                 if x_connect_mode == "L2"
 238                 else layer3["client_ip4_prefix"],
 239                 host_ip6_prefix=layer2["client_ip6_prefix"]
 240                 if x_connect_mode == "L2"
 241                 else layer3["client_ip6_prefix"],
 242                 host_ip4_gw=vpp_client_nexthop
 243                 if x_connect_mode == "L3" and ip_version == 4
 244                 else None,
 245                 host_ip6_gw=vpp_client_nexthop
 246                 if x_connect_mode == "L3" and ip_version == 6
 247                 else None,
 248                 int_type=client_if_type,
 249                 host_if_name=f"{client_if_type}0",
 250                 enable_gso=enable_client_if_gso,
 251                 enable_gro=enable_client_if_gro,
 252                 enable_checksum_offload=enable_client_if_checksum_offload,
 253             )
 254             self.vpp_interfaces.append(self.ingress_if_idx)
 255             self.linux_interfaces.append([client_namespace, f"{client_if_type}0"])
 256             # Seeing TCP timeouts if tx=on & rx=on Linux tap & tun interfaces
 257             disable_interface_gso(client_namespace, f"{client_if_type}0")
 258         else:
 259             print(
 260                 f"Unsupported client interface type: {client_if_type} "
 261                 f"for test - ID={test['id']}"
 262             )
 263             sys.exit(1)
 264
 265         if server_if_type == "af_packet":
 266             create_host_interface(
 267                 af_packet_config["iprf_server_interface_on_linux"],
 268                 af_packet_config["iprf_server_interface_on_vpp"],
 269                 server_namespace,
 270                 server_ip4_prefix,
 271                 server_ip6_prefix,
 272             )
 273             self.egress_if_idx = self.create_af_packet(
 274                 version=server_if_version,
 275                 host_if_name=af_packet_config["iprf_server_interface_on_vpp"],
 276                 enable_gso=enable_server_if_gso,
 277             )
 278             self.vpp_interfaces.append(self.egress_if_idx)
 279             self.linux_interfaces.append(
 280                 ["", af_packet_config["iprf_server_interface_on_vpp"]]
 281             )
 282             self.linux_interfaces.append(
 283                 [server_namespace, af_packet_config["iprf_server_interface_on_linux"]]
 284             )
 285             if enable_server_if_gso == 0:
 286                 disable_interface_gso(
 287                     "", af_packet_config["iprf_server_interface_on_vpp"]
 288                 )
 289                 disable_interface_gso(
 290                     server_namespace, af_packet_config["iprf_server_interface_on_linux"]
 291                 )
 292         elif server_if_type == "tap" or server_if_type == "tun":
 293             self.egress_if_idx = self.create_tap_tun(
 294                 id=102,
 295                 host_namespace=server_namespace,
 296                 ip_version=ip_version,
 297                 host_ip4_prefix=layer2["server_ip4_prefix"]
 298                 if x_connect_mode == "L2"
 299                 else layer3["server_ip4_prefix"],
 300                 host_ip6_prefix=layer2["server_ip6_prefix"]
 301                 if x_connect_mode == "L2"
 302                 else layer3["server_ip6_prefix"],
 303                 int_type=server_if_type,
 304                 host_if_name=f"{server_if_type}0",
 305                 enable_gso=enable_server_if_gso,
 306                 enable_gro=enable_server_if_gro,
 307                 enable_checksum_offload=enable_server_if_checksum_offload,
 308             )
 309             self.vpp_interfaces.append(self.egress_if_idx)
 310             self.linux_interfaces.append([server_namespace, f"{server_if_type}0"])
 311             # Seeing TCP timeouts if tx=on & rx=on Linux tap & tun interfaces
 312             disable_interface_gso(server_namespace, f"{server_if_type}0")
 313         else:
 314             print(
 315                 f"Unsupported server interface type: {server_if_type} "
 316                 f"for test - ID={test['id']}"
 317             )
 318             sys.exit(1)
 319
 320         if x_connect_mode == "L2":
 321             self.l2_connect_interfaces(1, self.ingress_if_idx, self.egress_if_idx)
 322         elif x_connect_mode == "L3":
 323             # L3 connect client & server side
 324             vrf_id = layer3["ip4_vrf"] if ip_version == 4 else layer3["ip6_vrf"]
 325             self.l3_connect_interfaces(
 326                 ip_version,
 327                 vrf_id,
 328                 (self.ingress_if_idx, vpp_client_prefix),
 329                 (self.egress_if_idx, vpp_server_prefix),
 330             )
 331             # Setup namespace routing
 332             if ip_version == 4:
 333                 add_namespace_route(client_namespace, "0.0.0.0/0", vpp_client_nexthop)
 334                 add_namespace_route(server_namespace, "0.0.0.0/0", vpp_server_nexthop)
 335             else:
 336                 add_namespace_route(client_namespace, "::/0", vpp_client_nexthop)
 337                 add_namespace_route(server_namespace, "::/0", vpp_server_nexthop)
 338         # Wait for Linux IPv6 stack to become ready
 339         if ip_version == 6:
 340             time.sleep(2)
 341
 342     def tearDown(self):
 343         try:
 344             self.vapi.tap_delete_v2(self.ingress_if_idx)
 345         except Exception:
 346             pass
 347         try:
 348             self.vapi.tap_delete_v2(self.egress_if_idx)
 349         except Exception:
 350             pass
 351         try:
 352             for interface in self.vapi.af_packet_dump():
 353                 if (
 354                     interface.host_if_name
 355                     == af_packet_config["iprf_client_interface_on_vpp"]
 356                 ):
 357                     self.vapi.af_packet_delete(
 358                         af_packet_config["iprf_client_interface_on_vpp"]
 359                     )
 360                 elif (
 361                     interface.host_if_name
 362                     == af_packet_config["iprf_server_interface_on_vpp"]
 363                 ):
 364                     self.vapi.af_packet_delete(
 365                         af_packet_config["iprf_server_interface_on_vpp"]
 366                     )
 367         except Exception:
 368             pass
 369         try:
 370             delete_host_interfaces(
 371                 af_packet_config["iprf_client_interface_on_linux"],
 372                 af_packet_config["iprf_server_interface_on_linux"],
 373                 af_packet_config["iprf_client_interface_on_vpp"],
 374                 af_packet_config["iprf_server_interface_on_vpp"],
 375             )
 376         except Exception:
 377             pass
 378         try:
 379             self.vapi.ip_table_add_del(is_add=0, table={"table_id": layer3["ip4_vrf"]})
 380         except Exception:
 381             pass
 382         try:
 383             self.vapi.ip_table_add_del(is_add=0, table={"table_id": layer3["ip6_vrf"]})
 384         except Exception:
 385             pass
 386         try:
 387             delete_namespace(
 388                 [
 389                     client_namespace,
 390                     server_namespace,
 391                 ]
 392             )
 393         except Exception:
 394             pass
 395         try:
 396             stop_iperf()
 397         except Exception:
 398             pass
 399
 400     def create_af_packet(self, version, host_if_name, enable_gso=0):
 401         """Create an af_packetv3 interface in VPP.
 402
 403         Parameters:
 404         version -- 2 for af_packet_create_v2
 405                 -- 3 for af_packet_create_v3
 406         host_if_name -- host interface name
 407         enable_gso -- Enable GSO on the interface when True
 408         """
 409         af_packet_mode = VppEnum.vl_api_af_packet_mode_t
 410         af_packet_interface_mode = af_packet_mode.AF_PACKET_API_MODE_ETHERNET
 411         af_packet_flags = VppEnum.vl_api_af_packet_flags_t
 412         af_packet_interface_flags = af_packet_flags.AF_PACKET_API_FLAG_QDISC_BYPASS
 413         if enable_gso:
 414             af_packet_interface_flags = (
 415                 af_packet_interface_flags | af_packet_flags.AF_PACKET_API_FLAG_CKSUM_GSO
 416             )
 417         if version == 2:
 418             af_packet_interface_flags = (
 419                 af_packet_interface_flags | af_packet_flags.AF_PACKET_API_FLAG_VERSION_2
 420             )
 421         api_args = {
 422             "use_random_hw_addr": True,
 423             "host_if_name": host_if_name,
 424             "flags": af_packet_interface_flags,
 425         }
 426         api_args["mode"] = af_packet_interface_mode
 427         result = self.vapi.af_packet_create_v3(**api_args)
 428         sw_if_index = result.sw_if_index
 429         # Enable software GSO chunking when interface doesn't support GSO offload
 430         if enable_gso == 0:
 431             self.vapi.feature_gso_enable_disable(
 432                 sw_if_index=sw_if_index, enable_disable=1
 433             )
 434         else:
 435             self.vapi.feature_gso_enable_disable(
 436                 sw_if_index=sw_if_index, enable_disable=0
 437             )
 438         self.vapi.sw_interface_set_flags(sw_if_index=sw_if_index, flags=1)
 439         return sw_if_index
 440
 441     def create_tap_tun(
 442         self,
 443         id,
 444         host_namespace,
 445         ip_version,
 446         host_ip4_prefix=None,
 447         host_ip6_prefix=None,
 448         host_ip4_gw=None,
 449         host_ip6_gw=None,
 450         int_type="tap",
 451         host_if_name=None,
 452         enable_gso=0,
 453         enable_gro=0,
 454         enable_checksum_offload=0,
 455     ):
 456         """Create a tapv2 or tunv2 interface in VPP and attach to host.
 457
 458         Parameters:
 459         id -- interface ID
 460         host_namespace -- host namespace to attach the tap/tun interface to
 461         ip_version -- 4 or 6
 462         host_ip4_prefix -- ipv4 host interface address in CIDR notation
 463                            if ip_version=4
 464         host_ip6_prefix -- ipv6 host interface address in CIDR notation
 465                            if ip_version=6
 466         host_ip4_gw -- host IPv4 default gateway IP Address
 467         host_ip6_gw -- host IPv6 default gateway IP address
 468         int_type -- "tap" for tapv2  &  "tun" for tunv2 interface
 469         host_if_name -- host side interface name
 470         enable_gso -- enable GSO
 471         enable_gro -- enable GSO/GRO-Coalesce
 472         enable_checksum_offload -- enable checksum offload without gso
 473         """
 474         TapFlags = VppEnum.vl_api_tap_flags_t
 475         tap_flags = 0
 476         if int_type == "tun":
 477             tap_flags = TapFlags.TAP_API_FLAG_TUN
 478             if enable_gro:
 479                 tap_flags = tap_flags | (
 480                     TapFlags.TAP_API_FLAG_GSO | TapFlags.TAP_API_FLAG_GRO_COALESCE
 481                 )
 482             elif enable_gso:
 483                 tap_flags = tap_flags | TapFlags.TAP_API_FLAG_GSO
 484             elif enable_checksum_offload:
 485                 tap_flags = tap_flags | TapFlags.TAP_API_FLAG_CSUM_OFFLOAD
 486         elif int_type == "tap":
 487             if enable_gro:
 488                 tap_flags = (
 489                     TapFlags.TAP_API_FLAG_GSO | TapFlags.TAP_API_FLAG_GRO_COALESCE
 490                 )
 491             elif enable_gso:
 492                 tap_flags = TapFlags.TAP_API_FLAG_GSO
 493             elif enable_checksum_offload:
 494                 tap_flags = tap_flags | TapFlags.TAP_API_FLAG_CSUM_OFFLOAD
 495
 496         api_args = {
 497             "id": id,
 498             "host_namespace_set": True,
 499             "host_namespace": host_namespace,
 500             "host_if_name_set": False,
 501             "host_bridge_set": False,
 502             "host_mac_addr_set": False,
 503         }
 504         if tap_flags != 0:
 505             api_args["tap_flags"] = tap_flags
 506         if ip_version == 4:
 507             api_args["host_ip4_prefix"] = ip_interface(host_ip4_prefix)
 508             api_args["host_ip4_prefix_set"] = True
 509             if host_ip4_gw:
 510                 api_args["host_ip4_gw"] = ip_address(host_ip4_gw)
 511                 api_args["host_ip4_gw_set"] = True
 512         if ip_version == 6:
 513             api_args["host_ip6_prefix"] = ip_interface(host_ip6_prefix)
 514             api_args["host_ip6_prefix_set"] = True
 515             if host_ip6_gw:
 516                 api_args["host_ip6_gw"] = ip_address(host_ip6_gw)
 517                 api_args["host_ip6_gw_set"] = True
 518         if host_if_name:
 519             api_args["host_if_name"] = host_if_name
 520             api_args["host_if_name_set"] = True
 521
 522         result = self.vapi.tap_create_v2(**api_args)
 523         sw_if_index = result.sw_if_index
 524         # Enable software GSO chunking when interface doesn't support GSO offload and
 525         # GRO coalesce
 526         if enable_gso == 0 and enable_gro == 0:
 527             self.vapi.feature_gso_enable_disable(
 528                 sw_if_index=sw_if_index, enable_disable=1
 529             )
 530         else:
 531             self.vapi.feature_gso_enable_disable(
 532                 sw_if_index=sw_if_index, enable_disable=0
 533             )
 534         # Admin up
 535         self.vapi.sw_interface_set_flags(sw_if_index=sw_if_index, flags=1)
 536         return sw_if_index
 537
 538     def dump_bridge_domain_details(self, bd_id):
 539         return self.vapi.bridge_domain_dump(bd_id=bd_id)
 540
 541     def l2_connect_interfaces(self, bridge_id, *sw_if_idxs):
 542         for if_idx in sw_if_idxs:
 543             self.vapi.sw_interface_set_l2_bridge(
 544                 rx_sw_if_index=if_idx, bd_id=bridge_id, shg=0, port_type=0, enable=True
 545             )
 546
 547     def l3_connect_interfaces(self, ip_version, vrf_id, *if_idx_ip_prefixes):
 548         """Setup routing for (if_idx, ip_prefix) inside VPP.
 549
 550         arguments:
 551         if_idx_ip_prefixes -- sequence of (if_idx, ip_prefix) tuples
 552         ip_version -- 4 or 6
 553         vrf_id -- vrf_id
 554         """
 555         is_ipv6 = 0 if ip_version == 4 else 1
 556         self.vapi.ip_table_add_del(
 557             is_add=1, table={"table_id": vrf_id, "is_ip6": is_ipv6}
 558         )
 559         for sw_if_index, ip_prefix in if_idx_ip_prefixes:
 560             self.vapi.sw_interface_set_table(
 561                 sw_if_index=sw_if_index, is_ipv6=is_ipv6, vrf_id=vrf_id
 562             )
 563             self.vapi.sw_interface_add_del_address(
 564                 sw_if_index=sw_if_index, is_add=1, prefix=ip_interface(ip_prefix)
 565             )
 566
 567     def set_interfaces_mtu(self, mtu, ip_version, **kwargs):
 568         """Set MTUs on VPP and Linux interfaces.
 569
 570         arguments --
 571         mtu -- mtu value
 572         ip_version - 4 or 6
 573         kwargs['vpp_interfaces'] -- list of vpp interface if indexes
 574         kwargs['linux_interfaces'] -- list of tuples (namespace, interface_names)
 575         return True if mtu is set, else False
 576         """
 577         vpp_interfaces = kwargs.get("vpp_interfaces")
 578         linux_interfaces = kwargs.get("linux_interfaces")
 579         # IPv6 on Linux requires an MTU value >=1280
 580         if (ip_version == 6 and mtu >= 1280) or ip_version == 4:
 581             for sw_if_idx in vpp_interfaces:
 582                 self.vapi.sw_interface_set_mtu(
 583                     sw_if_index=sw_if_idx, mtu=[mtu, 0, 0, 0]
 584                 )
 585             for namespace, interface_name in linux_interfaces:
 586                 set_interface_mtu(
 587                     namespace=namespace,
 588                     interface=interface_name,
 589                     mtu=mtu,
 590                     logger=self.logger,
 591                 )
 592             return True
 593         else:
 594             return False
 595
 596
 597 generate_vpp_interface_tests()
 598
 599 if __name__ == "__main__":
 600     unittest.main(testRunner=VppTestRunner)