test/vpp_ip.py

   1 """
   2   IP Types
   3
   4 """
   5 import logging
   6
   7 from ipaddress import ip_address
   8 from socket import AF_INET, AF_INET6
   9 from vpp_papi import VppEnum
  10 from vpp_object import VppObject
  11 try:
  12     text_type = unicode
  13 except NameError:
  14     text_type = str
  15
  16 _log = logging.getLogger(__name__)
  17
  18
  19 class DpoProto:
  20     DPO_PROTO_IP4 = 0
  21     DPO_PROTO_IP6 = 1
  22     DPO_PROTO_MPLS = 2
  23     DPO_PROTO_ETHERNET = 3
  24     DPO_PROTO_BIER = 4
  25     DPO_PROTO_NSH = 5
  26
  27
  28 INVALID_INDEX = 0xffffffff
  29
  30
  31 def get_dpo_proto(addr):
  32     if ip_address(addr).version == 6:
  33         return DpoProto.DPO_PROTO_IP6
  34     else:
  35         return DpoProto.DPO_PROTO_IP4
  36
  37
  38 class VppIpAddressUnion():
  39     def __init__(self, addr):
  40         self.addr = addr
  41         self.ip_addr = ip_address(text_type(self.addr))
  42
  43     def encode(self):
  44         if self.version == 6:
  45             return {'ip6': self.ip_addr}
  46         else:
  47             return {'ip4': self.ip_addr}
  48
  49     @property
  50     def version(self):
  51         return self.ip_addr.version
  52
  53     @property
  54     def address(self):
  55         return self.addr
  56
  57     @property
  58     def length(self):
  59         return self.ip_addr.max_prefixlen
  60
  61     @property
  62     def bytes(self):
  63         return self.ip_addr.packed
  64
  65     def __eq__(self, other):
  66         if isinstance(other, self.__class__):
  67             return self.ip_addr == other.ip_addr
  68         elif hasattr(other, "ip4") and hasattr(other, "ip6"):
  69             # vl_api_address_union_t
  70             if 4 == self.version:
  71                 return self.ip_addr == other.ip4
  72             else:
  73                 return self.ip_addr == other.ip6
  74         else:
  75             raise Exception("Comparing VppIpAddressUnions:%s"
  76                             " with incomparable type: %s",
  77                             self, other)
  78
  79     def __ne__(self, other):
  80         return not (self == other)
  81
  82     def __str__(self):
  83         return str(self.ip_addr)
  84
  85
  86 class VppIpMPrefix():
  87     def __init__(self, saddr, gaddr, glen):
  88         self.saddr = saddr
  89         self.gaddr = gaddr
  90         self.glen = glen
  91         if ip_address(self.saddr).version != \
  92            ip_address(self.gaddr).version:
  93             raise ValueError('Source and group addresses must be of the '
  94                              'same address family.')
  95
  96     def encode(self):
  97         return {
  98             'af': ip_address(self.gaddr).vapi_af,
  99             'grp_address': {
 100                 ip_address(self.gaddr).vapi_af_name: self.gaddr
 101             },
 102             'src_address': {
 103                 ip_address(self.saddr).vapi_af_name: self.saddr
 104             },
 105             'grp_address_length': self.glen,
 106         }
 107
 108     @property
 109     def length(self):
 110         return self.glen
 111
 112     @property
 113     def version(self):
 114         return ip_address(self.gaddr).version
 115
 116     def __str__(self):
 117         return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
 118
 119     def __eq__(self, other):
 120         if isinstance(other, self.__class__):
 121             return (self.glen == other.glen and
 122                     self.saddr == other.gaddr and
 123                     self.saddr == other.saddr)
 124         elif (hasattr(other, "grp_address_length") and
 125               hasattr(other, "grp_address") and
 126               hasattr(other, "src_address")):
 127             # vl_api_mprefix_t
 128             if 4 == self.version:
 129                 return (self.glen == other.grp_address_length and
 130                         self.gaddr == str(other.grp_address.ip4) and
 131                         self.saddr == str(other.src_address.ip4))
 132             else:
 133                 return (self.glen == other.grp_address_length and
 134                         self.gaddr == str(other.grp_address.ip6) and
 135                         self.saddr == str(other.src_address.ip6))
 136         return NotImplemented
 137
 138
 139 class VppIpPuntPolicer(VppObject):
 140     def __init__(self, test, policer_index, is_ip6=False):
 141         self._test = test
 142         self._policer_index = policer_index
 143         self._is_ip6 = is_ip6
 144
 145     def add_vpp_config(self):
 146         self._test.vapi.ip_punt_police(policer_index=self._policer_index,
 147                                        is_ip6=self._is_ip6, is_add=True)
 148
 149     def remove_vpp_config(self):
 150         self._test.vapi.ip_punt_police(policer_index=self._policer_index,
 151                                        is_ip6=self._is_ip6, is_add=False)
 152
 153     def query_vpp_config(self):
 154         NotImplemented
 155
 156
 157 class VppIpPuntRedirect(VppObject):
 158     def __init__(self, test, rx_index, tx_index, nh_addr):
 159         self._test = test
 160         self._rx_index = rx_index
 161         self._tx_index = tx_index
 162         self._nh_addr = ip_address(nh_addr)
 163
 164     def encode(self):
 165         return {"rx_sw_if_index": self._rx_index,
 166                 "tx_sw_if_index": self._tx_index, "nh": self._nh_addr}
 167
 168     def add_vpp_config(self):
 169         self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=True)
 170         self._test.registry.register(self, self._test.logger)
 171
 172     def remove_vpp_config(self):
 173         self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=False)
 174
 175     def get_vpp_config(self):
 176         is_ipv6 = True if self._nh_addr.version == 6 else False
 177         return self._test.vapi.ip_punt_redirect_dump(
 178             sw_if_index=self._rx_index, is_ipv6=is_ipv6)
 179
 180     def query_vpp_config(self):
 181         if self.get_vpp_config():
 182             return True
 183         return False
 184
 185
 186 class VppIpPathMtu(VppObject):
 187     def __init__(self, test, nh, pmtu, table_id=0):
 188         self._test = test
 189         self.nh = nh
 190         self.pmtu = pmtu
 191         self.table_id = table_id
 192
 193     def add_vpp_config(self):
 194         self._test.vapi.ip_path_mtu_update(pmtu={'nh': self.nh,
 195                                                  'table_id': self.table_id,
 196                                                  'path_mtu': self.pmtu})
 197         self._test.registry.register(self, self._test.logger)
 198         return self
 199
 200     def modify(self, pmtu):
 201         self.pmtu = pmtu
 202         self._test.vapi.ip_path_mtu_update(pmtu={'nh': self.nh,
 203                                                  'table_id': self.table_id,
 204                                                  'path_mtu': self.pmtu})
 205         return self
 206
 207     def remove_vpp_config(self):
 208         self._test.vapi.ip_path_mtu_update(pmtu={'nh': self.nh,
 209                                                  'table_id': self.table_id,
 210                                                  'path_mtu': 0})
 211
 212     def query_vpp_config(self):
 213         ds = list(self._test.vapi.vpp.details_iter(
 214             self._test.vapi.ip_path_mtu_get))
 215
 216         for d in ds:
 217             if self.nh == str(d.pmtu.nh) \
 218                and self.table_id == d.pmtu.table_id \
 219                and self.pmtu == d.pmtu.path_mtu:
 220                 return True
 221         return False
 222
 223     def object_id(self):
 224         return ("ip-path-mtu-%d-%s-%d" % (self.table_id,
 225                                           self.nh,
 226                                           self.pmtu))
 227
 228     def __str__(self):
 229         return self.object_id()