test/vpp_ip.py

   1 """
   2   IP Types
   3
   4 """
   5
   6 from ipaddress import ip_address
   7 from socket import AF_INET, AF_INET6
   8 from vpp_papi import VppEnum
   9
  10
  11 class DpoProto:
  12     DPO_PROTO_IP4 = 0
  13     DPO_PROTO_IP6 = 1
  14     DPO_PROTO_MPLS = 2
  15     DPO_PROTO_ETHERNET = 3
  16     DPO_PROTO_BIER = 4
  17     DPO_PROTO_NSH = 5
  18
  19
  20 INVALID_INDEX = 0xffffffff
  21
  22
  23 class VppIpAddressUnion():
  24     def __init__(self, addr):
  25         self.addr = addr
  26         self.ip_addr = ip_address(unicode(self.addr))
  27
  28     def encode(self):
  29         if self.version is 6:
  30             return {'ip6': self.ip_addr.packed}
  31         else:
  32             return {'ip4': self.ip_addr.packed}
  33
  34     @property
  35     def version(self):
  36         return self.ip_addr.version
  37
  38     @property
  39     def address(self):
  40         return self.addr
  41
  42     @property
  43     def length(self):
  44         if self.version is 6:
  45             return 128
  46         else:
  47             return 32
  48
  49     @property
  50     def bytes(self):
  51         return self.ip_addr.packed
  52
  53     def __eq__(self, other):
  54         if isinstance(other, self.__class__):
  55             return self.ip_addr == other.ip_addr
  56         elif hasattr(other, "ip4") and hasattr(other, "ip6"):
  57             # vl_api_address_union_t
  58             if 4 is self.version:
  59                 return self.ip_addr.packed == other.ip4
  60             else:
  61                 return self.ip_addr.packed == other.ip6
  62         else:
  63             raise Exception("Comparing VppIpAddresUnions:%s"
  64                             " with unknown type: %s" %
  65                             (self, other))
  66
  67         return False
  68
  69
  70 class VppIpAddress():
  71     def __init__(self, addr):
  72         self.addr = VppIpAddressUnion(addr)
  73
  74     def encode(self):
  75         if self.addr.version is 6:
  76             return {
  77                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
  78                 'un': self.addr.encode()
  79             }
  80         else:
  81             return {
  82                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
  83                 'un': self.addr.encode()
  84             }
  85
  86     def __eq__(self, other):
  87         if isinstance(other, self.__class__):
  88             return self.addr == other.addr
  89         elif hasattr(other, "af") and hasattr(other, "un"):
  90             # a vp_api_address_t
  91             if 4 is self.version:
  92                 return other.af == \
  93                     VppEnum.vl_api_address_family_t.ADDRESS_IP4 and \
  94                     other.un == self.addr
  95             else:
  96                 return other.af == \
  97                     VppEnum.vl_api_address_family_t.ADDRESS_IP6 and \
  98                     other.un == self.addr
  99         else:
 100             raise Exception("Comparing VppIpAddress:%s with unknown type: %s" %
 101                             (self, other))
 102         return False
 103
 104     def __ne__(self, other):
 105         return not (self == other)
 106
 107     def __str__(self):
 108         return self.address
 109
 110     @property
 111     def bytes(self):
 112         return self.addr.bytes
 113
 114     @property
 115     def bytes(self):
 116         return self.addr.bytes
 117
 118     @property
 119     def address(self):
 120         return self.addr.address
 121
 122     @property
 123     def length(self):
 124         return self.addr.length
 125
 126     @property
 127     def version(self):
 128         return self.addr.version
 129
 130     @property
 131     def is_ip6(self):
 132         return (self.version == 6)
 133
 134     @property
 135     def af(self):
 136         if self.version == 6:
 137             return AF_INET6
 138         else:
 139             return AF_INET
 140
 141     @property
 142     def dpo_proto(self):
 143         if self.version is 6:
 144             return DpoProto.DPO_PROTO_IP6
 145         else:
 146             return DpoProto.DPO_PROTO_IP4
 147
 148
 149 class VppIpPrefix():
 150     def __init__(self, addr, len):
 151         self.addr = VppIpAddress(addr)
 152         self.len = len
 153
 154     def __eq__(self, other):
 155         if self.addr == other.addr and self.len == other.len:
 156             return True
 157         return False
 158
 159     def encode(self):
 160         return {'address': self.addr.encode(),
 161                 'address_length': self.len}
 162
 163     @property
 164     def address(self):
 165         return self.addr.address
 166
 167     @property
 168     def bytes(self):
 169         return self.addr.bytes
 170
 171     @property
 172     def length(self):
 173         return self.len
 174
 175     @property
 176     def is_ip6(self):
 177         return self.addr.is_ip6
 178
 179     def __str__(self):
 180         return "%s/%d" % (self.address, self.length)
 181
 182     def __eq__(self, other):
 183         if isinstance(other, self.__class__):
 184             return (self.len == other.len and self.addr == other.addr)
 185         elif hasattr(other, "address") and hasattr(other, "address_length"):
 186             # vl_api_prefix_t
 187             return self.len == other.address_length and \
 188                 self.addr == other.address
 189         else:
 190             raise Exception("Comparing VppIpPrefix:%s with unknown type: %s" %
 191                             (self, other))
 192         return False
 193
 194
 195 class VppIp6Prefix():
 196     def __init__(self, prefix, prefixlen):
 197         self.ip_prefix = ip_address(unicode(prefix))
 198         self.prefixlen = prefixlen
 199
 200     def encode(self):
 201         return {'prefix': self.ip_prefix.packed,
 202                 'len': self.prefixlen}
 203
 204
 205 class VppIp4Prefix(VppIp6Prefix):
 206     pass
 207
 208
 209 class VppIpMPrefix():
 210     def __init__(self, saddr, gaddr, len):
 211         self.saddr = saddr
 212         self.gaddr = gaddr
 213         self.len = len
 214         self.ip_saddr = ip_address(unicode(self.saddr))
 215         self.ip_gaddr = ip_address(unicode(self.gaddr))
 216
 217     def encode(self):
 218
 219         if 6 is self.ip_saddr.version:
 220             prefix = {
 221                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
 222                 'grp_address': {'ip6': self.ip_gaddr.packed},
 223                 'src_address': {'ip6': self.ip_saddr.packed},
 224                 'grp_address_length': self.len,
 225             }
 226         else:
 227             prefix = {
 228                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
 229                 'grp_address': {'ip4': self.ip_gaddr.packed},
 230                 'src_address': {'ip4': self.ip_saddr.packed},
 231                 'grp_address_length': self.len,
 232             }
 233         return prefix