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 try:
  11     text_type = unicode
  12 except NameError:
  13     text_type = str
  14
  15 _log = logging.getLogger(__name__)
  16
  17
  18 class DpoProto:
  19     DPO_PROTO_IP4 = 0
  20     DPO_PROTO_IP6 = 1
  21     DPO_PROTO_MPLS = 2
  22     DPO_PROTO_ETHERNET = 3
  23     DPO_PROTO_BIER = 4
  24     DPO_PROTO_NSH = 5
  25
  26
  27 INVALID_INDEX = 0xffffffff
  28
  29
  30 def get_dpo_proto(addr):
  31     if ip_address(addr).version == 6:
  32         return DpoProto.DPO_PROTO_IP6
  33     else:
  34         return DpoProto.DPO_PROTO_IP4
  35
  36
  37 class VppIpAddressUnion():
  38     def __init__(self, addr):
  39         self.addr = addr
  40         self.ip_addr = ip_address(text_type(self.addr))
  41
  42     def encode(self):
  43         if self.version == 6:
  44             return {'ip6': self.ip_addr}
  45         else:
  46             return {'ip4': self.ip_addr}
  47
  48     @property
  49     def version(self):
  50         return self.ip_addr.version
  51
  52     @property
  53     def address(self):
  54         return self.addr
  55
  56     @property
  57     def length(self):
  58         return self.ip_addr.max_prefixlen
  59
  60     @property
  61     def bytes(self):
  62         return self.ip_addr.packed
  63
  64     def __eq__(self, other):
  65         if isinstance(other, self.__class__):
  66             return self.ip_addr == other.ip_addr
  67         elif hasattr(other, "ip4") and hasattr(other, "ip6"):
  68             # vl_api_address_union_t
  69             if 4 == self.version:
  70                 return self.ip_addr == other.ip4
  71             else:
  72                 return self.ip_addr == other.ip6
  73         else:
  74             raise Exception("Comparing VppIpAddressUnions:%s"
  75                             " with incomparable type: %s",
  76                             self, other)
  77
  78     def __ne__(self, other):
  79         return not (self == other)
  80
  81     def __str__(self):
  82         return str(self.ip_addr)
  83
  84
  85 class VppIpMPrefix():
  86     def __init__(self, saddr, gaddr, glen):
  87         self.saddr = saddr
  88         self.gaddr = gaddr
  89         self.glen = glen
  90         if ip_address(self.saddr).version != \
  91            ip_address(self.gaddr).version:
  92             raise ValueError('Source and group addresses must be of the '
  93                              'same address family.')
  94
  95     def encode(self):
  96         return {
  97             'af': ip_address(self.gaddr).vapi_af,
  98             'grp_address': {
  99                 ip_address(self.gaddr).vapi_af_name: self.gaddr
 100             },
 101             'src_address': {
 102                 ip_address(self.saddr).vapi_af_name: self.saddr
 103             },
 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 (self.glen == other.glen and
 121                     self.saddr == other.gaddr and
 122                     self.saddr == other.saddr)
 123         elif (hasattr(other, "grp_address_length") and
 124               hasattr(other, "grp_address") and
 125               hasattr(other, "src_address")):
 126             # vl_api_mprefix_t
 127             if 4 == self.version:
 128                 return (self.glen == other.grp_address_length and
 129                         self.gaddr == str(other.grp_address.ip4) and
 130                         self.saddr == str(other.src_address.ip4))
 131             else:
 132                 return (self.glen == other.grp_address_length and
 133                         self.gaddr == str(other.grp_address.ip6) and
 134                         self.saddr == str(other.src_address.ip6))
 135         return NotImplemented