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