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