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         if 6 == self.version:
  97             prefix = {
  98                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
  99                 'grp_address': {
 100                     'ip6': self.gaddr
 101                 },
 102                 'src_address': {
 103                     'ip6': self.saddr
 104                 },
 105                 'grp_address_length': self.glen,
 106             }
 107         else:
 108             prefix = {
 109                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
 110                 'grp_address': {
 111                     'ip4': self.gaddr
 112                 },
 113                 'src_address': {
 114                     'ip4':  self.saddr
 115                 },
 116                 'grp_address_length': self.glen,
 117             }
 118         return prefix
 119
 120     @property
 121     def length(self):
 122         return self.glen
 123
 124     @property
 125     def version(self):
 126         return ip_address(self.gaddr).version
 127
 128     def __str__(self):
 129         return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
 130
 131     def __eq__(self, other):
 132         if isinstance(other, self.__class__):
 133             return (self.glen == other.glen and
 134                     self.saddr == other.gaddr and
 135                     self.saddr == other.saddr)
 136         elif (hasattr(other, "grp_address_length") and
 137               hasattr(other, "grp_address") and
 138               hasattr(other, "src_address")):
 139             # vl_api_mprefix_t
 140             if 4 == self.version:
 141                 return (self.glen == other.grp_address_length and
 142                         self.gaddr == str(other.grp_address.ip4) and
 143                         self.saddr == str(other.src_address.ip4))
 144             else:
 145                 return (self.glen == other.grp_address_length and
 146                         self.gaddr == str(other.grp_address.ip6) and
 147                         self.saddr == str(other.src_address.ip6))
 148             raise Exception("Comparing VppIpMPrefix:%s with unknown type: %s" %
 149                             (self, other))
 150         return False