test/vpp_interface.py

   1 import socket
   2 import abc
   3 import reprlib
   4
   5 from util import Host, mk_ll_addr
   6 from vpp_papi import VppEnum
   7 from ipaddress import IPv4Network
   8
   9 try:
  10     text_type = unicode
  11 except NameError:
  12     text_type = str
  13
  14
  15 class VppInterface(metaclass=abc.ABCMeta):
  16     """Generic VPP interface."""
  17
  18     @property
  19     def sw_if_index(self):
  20         """Interface index assigned by VPP."""
  21         return self._sw_if_index
  22
  23     @property
  24     def remote_mac(self):
  25         """MAC-address of the remote interface "connected" to this interface"""
  26         return self._remote_hosts[0].mac
  27
  28     @property
  29     def local_mac(self):
  30         """MAC-address of the VPP interface."""
  31         return str(self._local_mac)
  32
  33     @property
  34     def local_addr(self):
  35         return self._local_addr
  36
  37     @property
  38     def remote_addr(self):
  39         return self._remote_addr
  40
  41     @property
  42     def local_ip4(self):
  43         """Local IPv4 address on VPP interface (string)."""
  44         return self._local_ip4
  45
  46     @local_ip4.setter
  47     def local_ip4(self, value):
  48         self._local_ip4 = value
  49
  50     @property
  51     def local_ip4_prefix_len(self):
  52         """Local IPv4 prefix length"""
  53         return self._local_ip4_len
  54
  55     @local_ip4_prefix_len.setter
  56     def local_ip4_prefix_len(self, value):
  57         self._local_ip4_len = value
  58
  59     @property
  60     def local_ip4_prefix(self):
  61         """Local IPv4 prefix"""
  62         return "%s/%d" % (self._local_ip4, self._local_ip4_len)
  63
  64     @property
  65     def remote_ip4(self):
  66         """IPv4 address of remote peer "connected" to this interface."""
  67         return self._remote_hosts[0].ip4
  68
  69     @property
  70     def local_ip6(self):
  71         """Local IPv6 address on VPP interface (string)."""
  72         return self._local_ip6
  73
  74     @local_ip6.setter
  75     def local_ip6(self, value):
  76         self._local_ip6 = value
  77
  78     @property
  79     def local_ip6_prefix_len(self):
  80         """Local IPv6 prefix length"""
  81         return self._local_ip6_len
  82
  83     @local_ip6_prefix_len.setter
  84     def local_ip6_prefix_len(self, value):
  85         self._local_ip6_len = value
  86
  87     @property
  88     def local_ip6_prefix(self):
  89         """Local IPv4 prefix"""
  90         return "%s/%d" % (self._local_ip6, self._local_ip6_len)
  91
  92     @property
  93     def remote_ip6(self):
  94         """IPv6 address of remote peer "connected" to this interface."""
  95         return self._remote_hosts[0].ip6
  96
  97     @property
  98     def local_ip6_ll(self):
  99         """Local IPv6 link-local address on VPP interface (string)."""
 100         if not self._local_ip6_ll:
 101             self._local_ip6_ll = str(
 102                 self.test.vapi.sw_interface_ip6_get_link_local_address(
 103                     self.sw_if_index
 104                 ).ip
 105             )
 106         return self._local_ip6_ll
 107
 108     @property
 109     def remote_ip6_ll(self):
 110         """Link-local IPv6 address of remote peer
 111         "connected" to this interface."""
 112         return self._remote_ip6_ll
 113
 114     @property
 115     def name(self):
 116         """Name of the interface."""
 117         return self._name
 118
 119     @property
 120     def dump(self):
 121         """RAW result of sw_interface_dump for this interface."""
 122         return self._dump
 123
 124     @property
 125     def test(self):
 126         """Test case creating this interface."""
 127         return self._test
 128
 129     @property
 130     def remote_hosts(self):
 131         """Remote hosts list"""
 132         return self._remote_hosts
 133
 134     @remote_hosts.setter
 135     def remote_hosts(self, value):
 136         """
 137         :param list value: List of remote hosts.
 138         """
 139         self._remote_hosts = value
 140         self._hosts_by_mac = {}
 141         self._hosts_by_ip4 = {}
 142         self._hosts_by_ip6 = {}
 143         for host in self._remote_hosts:
 144             self._hosts_by_mac[host.mac] = host
 145             self._hosts_by_ip4[host.ip4] = host
 146             self._hosts_by_ip6[host.ip6] = host
 147
 148     def host_by_mac(self, mac):
 149         """
 150         :param mac: MAC address to find host by.
 151         :return: Host object assigned to interface.
 152         """
 153         return self._hosts_by_mac[mac]
 154
 155     def host_by_ip4(self, ip):
 156         """
 157         :param ip: IPv4 address to find host by.
 158         :return: Host object assigned to interface.
 159         """
 160         return self._hosts_by_ip4[ip]
 161
 162     def host_by_ip6(self, ip):
 163         """
 164         :param ip: IPv6 address to find host by.
 165         :return: Host object assigned to interface.
 166         """
 167         return self._hosts_by_ip6[ip]
 168
 169     def generate_remote_hosts(self, count=1):
 170         """Generate and add remote hosts for the interface.
 171
 172         :param int count: Number of generated remote hosts.
 173         """
 174         self._remote_hosts = []
 175         self._hosts_by_mac = {}
 176         self._hosts_by_ip4 = {}
 177         self._hosts_by_ip6 = {}
 178         for i in range(2, count + 2):  # 0: network address, 1: local vpp address
 179             mac = "02:%02x:00:00:ff:%02x" % (self.sw_if_index, i)
 180             ip4 = "172.16.%u.%u" % (self.sw_if_index, i)
 181             ip6 = "fd01:%x::%x" % (self.sw_if_index, i)
 182             ip6_ll = mk_ll_addr(mac)
 183             host = Host(mac, ip4, ip6, ip6_ll)
 184             self._remote_hosts.append(host)
 185             self._hosts_by_mac[mac] = host
 186             self._hosts_by_ip4[ip4] = host
 187             self._hosts_by_ip6[ip6] = host
 188
 189     @abc.abstractmethod
 190     def __init__(self, test):
 191         self._test = test
 192
 193         self._remote_hosts = []
 194         self._hosts_by_mac = {}
 195         self._hosts_by_ip4 = {}
 196         self._hosts_by_ip6 = {}
 197
 198     def set_mac(self, mac):
 199         self._local_mac = str(mac)
 200         self.test.vapi.sw_interface_set_mac_address(self.sw_if_index, mac.packed)
 201         return self
 202
 203     def set_sw_if_index(self, sw_if_index):
 204         if sw_if_index > 255:
 205             raise RuntimeError("Don't support sw_if_index values greater than 255.")
 206         self._sw_if_index = sw_if_index
 207
 208         self.generate_remote_hosts()
 209
 210         self._local_ip4 = "172.16.%u.1" % self.sw_if_index
 211         self._local_ip4_len = 24
 212         self._local_ip4_subnet = "172.16.%u.0" % self.sw_if_index
 213         self._local_ip4_bcast = "172.16.%u.255" % self.sw_if_index
 214         self.has_ip4_config = False
 215         self.ip4_table_id = 0
 216
 217         self._local_ip6 = "fd01:%x::1" % self.sw_if_index
 218         self._local_ip6_len = 64
 219         self.has_ip6_config = False
 220         self.ip6_table_id = 0
 221
 222         self._local_addr = {
 223             socket.AF_INET: self.local_ip4,
 224             socket.AF_INET6: self.local_ip6,
 225         }
 226         self._remote_addr = {
 227             socket.AF_INET: self.remote_ip4,
 228             socket.AF_INET6: self.remote_ip6,
 229         }
 230
 231         r = self.test.vapi.sw_interface_dump(sw_if_index=self.sw_if_index)
 232         for intf in r:
 233             if intf.sw_if_index == self.sw_if_index:
 234                 self._name = intf.interface_name
 235                 self._local_mac = intf.l2_address
 236                 self._dump = intf
 237                 break
 238         else:
 239             raise Exception(
 240                 "Could not find interface with sw_if_index %d "
 241                 "in interface dump %s" % (self.sw_if_index, reprlib.repr(r))
 242             )
 243         self._remote_ip6_ll = mk_ll_addr(self.remote_mac)
 244         self._local_ip6_ll = None
 245
 246     def config_ip4(self):
 247         """Configure IPv4 address on the VPP interface."""
 248         self.test.vapi.sw_interface_add_del_address(
 249             sw_if_index=self.sw_if_index, prefix=self.local_ip4_prefix
 250         )
 251         self.has_ip4_config = True
 252         return self
 253
 254     def unconfig_ip4(self):
 255         """Remove IPv4 address on the VPP interface."""
 256         try:
 257             if self.has_ip4_config:
 258                 self.test.vapi.sw_interface_add_del_address(
 259                     sw_if_index=self.sw_if_index, prefix=self.local_ip4_prefix, is_add=0
 260                 )
 261         except AttributeError:
 262             self.has_ip4_config = False
 263         self.has_ip4_config = False
 264         return self
 265
 266     def configure_ipv4_neighbors(self):
 267         """For every remote host assign neighbor's MAC to IPv4 addresses.
 268
 269         :param vrf_id: The FIB table / VRF ID. (Default value = 0)
 270         """
 271         for host in self._remote_hosts:
 272             self.test.vapi.ip_neighbor_add_del(self.sw_if_index, host.mac, host.ip4)
 273         return self
 274
 275     def config_ip6(self):
 276         """Configure IPv6 address on the VPP interface."""
 277         self.test.vapi.sw_interface_add_del_address(
 278             sw_if_index=self.sw_if_index, prefix=self.local_ip6_prefix
 279         )
 280         self.has_ip6_config = True
 281         return self
 282
 283     def unconfig_ip6(self):
 284         """Remove IPv6 address on the VPP interface."""
 285         try:
 286             if self.has_ip6_config:
 287                 self.test.vapi.sw_interface_add_del_address(
 288                     sw_if_index=self.sw_if_index, prefix=self.local_ip6_prefix, is_add=0
 289                 )
 290         except AttributeError:
 291             self.has_ip6_config = False
 292         self.has_ip6_config = False
 293         return self
 294
 295     def configure_ipv6_neighbors(self):
 296         """For every remote host assign neighbor's MAC to IPv6 addresses.
 297
 298         :param vrf_id: The FIB table / VRF ID. (Default value = 0)
 299         """
 300         for host in self._remote_hosts:
 301             self.test.vapi.ip_neighbor_add_del(self.sw_if_index, host.mac, host.ip6)
 302
 303     def unconfig(self):
 304         """Unconfigure IPv6 and IPv4 address on the VPP interface."""
 305         self.unconfig_ip4()
 306         self.unconfig_ip6()
 307         return self
 308
 309     def set_table_ip4(self, table_id):
 310         """Set the interface in a IPv4 Table.
 311
 312         .. note:: Must be called before configuring IP4 addresses.
 313         """
 314         self.ip4_table_id = table_id
 315         self.test.vapi.sw_interface_set_table(self.sw_if_index, 0, self.ip4_table_id)
 316         return self
 317
 318     def set_table_ip6(self, table_id):
 319         """Set the interface in a IPv6 Table.
 320
 321         .. note:: Must be called before configuring IP6 addresses.
 322         """
 323         self.ip6_table_id = table_id
 324         self.test.vapi.sw_interface_set_table(self.sw_if_index, 1, self.ip6_table_id)
 325         return self
 326
 327     def disable_ipv6_ra(self):
 328         """Configure IPv6 RA suppress on the VPP interface."""
 329         self.test.vapi.sw_interface_ip6nd_ra_config(
 330             sw_if_index=self.sw_if_index, suppress=1
 331         )
 332         return self
 333
 334     def ip6_ra_config(self, no=0, suppress=0, send_unicast=0):
 335         """Configure IPv6 RA suppress on the VPP interface."""
 336         self.test.vapi.sw_interface_ip6nd_ra_config(
 337             sw_if_index=self.sw_if_index,
 338             is_no=no,
 339             suppress=suppress,
 340             send_unicast=send_unicast,
 341         )
 342         return self
 343
 344     def ip6_ra_prefix(
 345         self, prefix, is_no=0, off_link=0, no_autoconfig=0, use_default=0
 346     ):
 347         """Configure IPv6 RA suppress on the VPP interface.
 348
 349         prefix can be a string in the format of '<address>/<length_in_bits>'
 350         or ipaddress.ipnetwork object (if strict.)"""
 351
 352         self.test.vapi.sw_interface_ip6nd_ra_prefix(
 353             sw_if_index=self.sw_if_index,
 354             prefix=prefix,
 355             use_default=use_default,
 356             off_link=off_link,
 357             no_autoconfig=no_autoconfig,
 358             is_no=is_no,
 359         )
 360         return self
 361
 362     def admin_up(self):
 363         """Put interface ADMIN-UP."""
 364         self.test.vapi.sw_interface_set_flags(
 365             self.sw_if_index,
 366             flags=VppEnum.vl_api_if_status_flags_t.IF_STATUS_API_FLAG_ADMIN_UP,
 367         )
 368         return self
 369
 370     def admin_down(self):
 371         """Put interface ADMIN-down."""
 372         self.test.vapi.sw_interface_set_flags(self.sw_if_index, flags=0)
 373         return self
 374
 375     def link_up(self):
 376         """Put interface link-state-UP."""
 377         self.test.vapi.cli("test interface link-state %s up" % self.name)
 378
 379     def link_down(self):
 380         """Put interface link-state-down."""
 381         self.test.vapi.cli("test interface link-state %s down" % self.name)
 382
 383     def ip6_enable(self):
 384         """IPv6 Enable interface"""
 385         self.test.vapi.sw_interface_ip6_enable_disable(self.sw_if_index, enable=1)
 386         return self
 387
 388     def ip6_disable(self):
 389         """Put interface ADMIN-DOWN."""
 390         self.test.vapi.sw_interface_ip6_enable_disable(self.sw_if_index, enable=0)
 391         return self
 392
 393     def add_sub_if(self, sub_if):
 394         """Register a sub-interface with this interface.
 395
 396         :param sub_if: sub-interface
 397         """
 398         if not hasattr(self, "sub_if"):
 399             self.sub_if = sub_if
 400         else:
 401             if isinstance(self.sub_if, list):
 402                 self.sub_if.append(sub_if)
 403             else:
 404                 self.sub_if = sub_if
 405         return self
 406
 407     def enable_mpls(self):
 408         """Enable MPLS on the VPP interface."""
 409         self.test.vapi.sw_interface_set_mpls_enable(self.sw_if_index)
 410         return self
 411
 412     def disable_mpls(self):
 413         """Enable MPLS on the VPP interface."""
 414         self.test.vapi.sw_interface_set_mpls_enable(self.sw_if_index, 0)
 415         return self
 416
 417     def is_ip4_entry_in_fib_dump(self, dump):
 418         for i in dump:
 419             n = IPv4Network(
 420                 text_type("%s/%d" % (self.local_ip4, self.local_ip4_prefix_len))
 421             )
 422             if i.route.prefix == n and i.route.table_id == self.ip4_table_id:
 423                 return True
 424         return False
 425
 426     def set_unnumbered(self, ip_sw_if_index):
 427         """Set the interface to unnumbered via ip_sw_if_index"""
 428         self.test.vapi.sw_interface_set_unnumbered(ip_sw_if_index, self.sw_if_index)
 429         return self
 430
 431     def unset_unnumbered(self, ip_sw_if_index):
 432         """Unset the interface to unnumbered via ip_sw_if_index"""
 433         self.test.vapi.sw_interface_set_unnumbered(
 434             ip_sw_if_index, self.sw_if_index, is_add=0
 435         )
 436         return self
 437
 438     def set_proxy_arp(self, enable=1):
 439         """Set the interface to enable/disable Proxy ARP"""
 440         self.test.vapi.proxy_arp_intfc_enable_disable(
 441             sw_if_index=self.sw_if_index, enable=enable
 442         )
 443         return self
 444
 445     def query_vpp_config(self):
 446         dump = self.test.vapi.sw_interface_dump(sw_if_index=self.sw_if_index)
 447         return self.is_interface_config_in_dump(dump)
 448
 449     def get_interface_config_from_dump(self, dump):
 450         for i in dump:
 451             if (
 452                 i.interface_name.rstrip(" \t\r\n\0") == self.name
 453                 and i.sw_if_index == self.sw_if_index
 454             ):
 455                 return i
 456         else:
 457             return None
 458
 459     def is_interface_config_in_dump(self, dump):
 460         return self.get_interface_config_from_dump(dump) is not None
 461
 462     def assert_interface_state(self, admin_up_down, link_up_down, expect_event=False):
 463         if expect_event:
 464             event = self.test.vapi.wait_for_event(timeout=1, name="sw_interface_event")
 465             self.test.assert_equal(event.sw_if_index, self.sw_if_index, "sw_if_index")
 466             self.test.assert_equal((event.flags & 1), admin_up_down, "admin state")
 467             self.test.assert_equal((event.flags & 2), link_up_down, "link state")
 468         dump = self.test.vapi.sw_interface_dump()
 469         if_state = self.get_interface_config_from_dump(dump)
 470         self.test.assert_equal((if_state.flags & 1), admin_up_down, "admin state")
 471         self.test.assert_equal((if_state.flags & 2), link_up_down, "link state")
 472
 473     def __str__(self):
 474         return self.name
 475
 476     def get_rx_stats(self):
 477         return self.test.statistics["/if/rx"][:, self.sw_if_index].sum_packets()
 478
 479     def get_tx_stats(self):
 480         return self.test.statistics["/if/tx"][:, self.sw_if_index].sum_packets()
 481
 482     def set_l3_mtu(self, mtu):
 483         self.test.vapi.sw_interface_set_mtu(self.sw_if_index, [mtu, 0, 0, 0])
 484         return self
 485
 486     def set_ip4_mtu(self, mtu):
 487         self.test.vapi.sw_interface_set_mtu(self.sw_if_index, [0, mtu, 0, 0])
 488         return self
 489
 490     def set_ip6_mtu(self, mtu):
 491         self.test.vapi.sw_interface_set_mtu(self.sw_if_index, [0, 0, mtu, 0])
 492         return self
 493
 494     def set_mpls_mtu(self, mtu):
 495         self.test.vapi.sw_interface_set_mtu(self.sw_if_index, [0, 0, 0, mtu])
 496         return self