test/vpp_ip.py

   1 """
   2   IP Types
   3
   4 """
   5
   6 import logging
   7
   8 from ipaddress import ip_address
   9 from vpp_object import VppObject
  10
  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(
  76                 "Comparing VppIpAddressUnions:%s with incomparable type: %s",
  77                 self,
  78                 other,
  79             )
  80
  81     def __ne__(self, other):
  82         return not (self == other)
  83
  84     def __str__(self):
  85         return str(self.ip_addr)
  86
  87
  88 class VppIpMPrefix:
  89     def __init__(self, saddr, gaddr, glen):
  90         self.saddr = saddr
  91         self.gaddr = gaddr
  92         self.glen = glen
  93         if ip_address(self.saddr).version != ip_address(self.gaddr).version:
  94             raise ValueError(
  95                 "Source and group addresses must be of the same address family."
  96             )
  97
  98     def encode(self):
  99         return {
 100             "af": ip_address(self.gaddr).vapi_af,
 101             "grp_address": {ip_address(self.gaddr).vapi_af_name: self.gaddr},
 102             "src_address": {ip_address(self.saddr).vapi_af_name: self.saddr},
 103             "grp_address_length": self.glen,
 104         }
 105
 106     @property
 107     def length(self):
 108         return self.glen
 109
 110     @property
 111     def version(self):
 112         return ip_address(self.gaddr).version
 113
 114     def __str__(self):
 115         return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
 116
 117     def __eq__(self, other):
 118         if isinstance(other, self.__class__):
 119             return (
 120                 self.glen == other.glen
 121                 and self.saddr == other.gaddr
 122                 and self.saddr == other.saddr
 123             )
 124         elif (
 125             hasattr(other, "grp_address_length")
 126             and hasattr(other, "grp_address")
 127             and hasattr(other, "src_address")
 128         ):
 129             # vl_api_mprefix_t
 130             if 4 == self.version:
 131                 return (
 132                     self.glen == other.grp_address_length
 133                     and self.gaddr == str(other.grp_address.ip4)
 134                     and self.saddr == str(other.src_address.ip4)
 135                 )
 136             else:
 137                 return (
 138                     self.glen == other.grp_address_length
 139                     and self.gaddr == str(other.grp_address.ip6)
 140                     and self.saddr == str(other.src_address.ip6)
 141                 )
 142         return NotImplemented
 143
 144
 145 class VppIpPuntPolicer(VppObject):
 146     def __init__(self, test, policer_index, is_ip6=False):
 147         self._test = test
 148         self._policer_index = policer_index
 149         self._is_ip6 = is_ip6
 150
 151     def add_vpp_config(self):
 152         self._test.vapi.ip_punt_police(
 153             policer_index=self._policer_index, is_ip6=self._is_ip6, is_add=True
 154         )
 155
 156     def remove_vpp_config(self):
 157         self._test.vapi.ip_punt_police(
 158             policer_index=self._policer_index, is_ip6=self._is_ip6, is_add=False
 159         )
 160
 161     def query_vpp_config(self):
 162         NotImplemented
 163
 164
 165 class VppIpPuntRedirect(VppObject):
 166     def __init__(self, test, rx_index, tx_index, nh_addr):
 167         self._test = test
 168         self._rx_index = rx_index
 169         self._tx_index = tx_index
 170         self._nh_addr = ip_address(nh_addr)
 171
 172     def encode(self):
 173         return {
 174             "rx_sw_if_index": self._rx_index,
 175             "tx_sw_if_index": self._tx_index,
 176             "nh": self._nh_addr,
 177         }
 178
 179     def add_vpp_config(self):
 180         self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=True)
 181         self._test.registry.register(self, self._test.logger)
 182         return self
 183
 184     def remove_vpp_config(self):
 185         self._test.vapi.ip_punt_redirect(punt=self.encode(), is_add=False)
 186
 187     def get_vpp_config(self):
 188         is_ipv6 = True if self._nh_addr.version == 6 else False
 189         return self._test.vapi.ip_punt_redirect_dump(
 190             sw_if_index=self._rx_index, is_ipv6=is_ipv6
 191         )
 192
 193     def query_vpp_config(self):
 194         if self.get_vpp_config():
 195             return True
 196         return False
 197
 198
 199 class VppIpPathMtu(VppObject):
 200     def __init__(self, test, nh, pmtu, table_id=0):
 201         self._test = test
 202         self.nh = nh
 203         self.pmtu = pmtu
 204         self.table_id = table_id
 205
 206     def add_vpp_config(self):
 207         self._test.vapi.ip_path_mtu_update(
 208             pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": self.pmtu}
 209         )
 210         self._test.registry.register(self, self._test.logger)
 211         return self
 212
 213     def modify(self, pmtu):
 214         self.pmtu = pmtu
 215         self._test.vapi.ip_path_mtu_update(
 216             pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": self.pmtu}
 217         )
 218         return self
 219
 220     def remove_vpp_config(self):
 221         self._test.vapi.ip_path_mtu_update(
 222             pmtu={"nh": self.nh, "table_id": self.table_id, "path_mtu": 0}
 223         )
 224
 225     def query_vpp_config(self):
 226         ds = list(self._test.vapi.vpp.details_iter(self._test.vapi.ip_path_mtu_get))
 227
 228         for d in ds:
 229             if (
 230                 self.nh == str(d.pmtu.nh)
 231                 and self.table_id == d.pmtu.table_id
 232                 and self.pmtu == d.pmtu.path_mtu
 233             ):
 234                 return True
 235         return False
 236
 237     def object_id(self):
 238         return "ip-path-mtu-%d-%s-%d" % (self.table_id, self.nh, self.pmtu)
 239
 240     def __str__(self):
 241         return self.object_id()