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