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