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