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