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