test/vpp_interface.py

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