test/lisp.py

   1 import socket
   2 from ipaddress import ip_network
   3
   4 from vpp_object import VppObject
   5
   6
   7 class VppLispLocatorSet(VppObject):
   8     """ Represents LISP locator set in VPP """
   9
  10     def __init__(self, test, ls_name):
  11         self._test = test
  12         self._ls_name = ls_name
  13
  14     @property
  15     def test(self):
  16         return self._test
  17
  18     @property
  19     def ls_name(self):
  20         return self._ls_name
  21
  22     def add_vpp_config(self):
  23         self.test.vapi.lisp_add_del_locator_set(locator_set_name=self._ls_name)
  24         self._test.registry.register(self, self.test.logger)
  25
  26     def get_lisp_locator_sets_dump_entry(self):
  27         result = self.test.vapi.lisp_locator_set_dump()
  28         for ls in result:
  29             if ls.ls_name.strip('\x00') == self._ls_name:
  30                 return ls
  31         return None
  32
  33     def query_vpp_config(self):
  34         return self.get_lisp_locator_sets_dump_entry() is not None
  35
  36     def remove_vpp_config(self):
  37         self.test.vapi.lisp_add_del_locator_set(locator_set_name=self._ls_name,
  38                                                 is_add=0)
  39
  40     def object_id(self):
  41         return 'lisp-locator-set-%s' % self._ls_name
  42
  43
  44 class VppLispLocator(VppObject):
  45     """ Represents LISP locator in VPP """
  46
  47     def __init__(self, test, sw_if_index, ls_name, priority=1, weight=1):
  48         self._test = test
  49         self._sw_if_index = sw_if_index
  50         self._ls_name = ls_name
  51         self._priority = priority
  52         self._weight = weight
  53
  54     @property
  55     def test(self):
  56         """ Test which created this locator """
  57         return self._test
  58
  59     @property
  60     def ls_name(self):
  61         """ Locator set name """
  62         return self._ls_name
  63
  64     @property
  65     def sw_if_index(self):
  66         return self._sw_if_index
  67
  68     @property
  69     def priority(self):
  70         return self._priority
  71
  72     @property
  73     def weight(self):
  74         return self._weight
  75
  76     def add_vpp_config(self):
  77         self.test.vapi.lisp_add_del_locator(locator_set_name=self._ls_name,
  78                                             sw_if_index=self._sw_if_index,
  79                                             priority=self._priority,
  80                                             weight=self._weight)
  81         self._test.registry.register(self, self.test.logger)
  82
  83     def get_lisp_locator_dump_entry(self):
  84         locators = self.test.vapi.lisp_locator_dump(
  85                 is_index_set=0, ls_name=self._ls_name)
  86         for locator in locators:
  87             if locator.sw_if_index == self._sw_if_index:
  88                 return locator
  89         return None
  90
  91     def query_vpp_config(self):
  92         locator = self.get_lisp_locator_dump_entry()
  93         return locator is not None
  94
  95     def remove_vpp_config(self):
  96         self.test.vapi.lisp_add_del_locator(
  97                 locator_set_name=self._ls_name, sw_if_index=self._sw_if_index,
  98                 priority=self._priority, weight=self._weight, is_add=0)
  99         self._test.registry.register(self, self.test.logger)
 100
 101     def object_id(self):
 102         return 'lisp-locator-%s-%d' % (self._ls_name, self._sw_if_index)
 103
 104
 105 class LispEIDType(object):
 106     PREFIX = 0
 107     MAC = 1
 108     NSH = 2
 109
 110
 111 class LispKeyIdType(object):
 112     NONE = 0
 113     SHA1 = 1
 114     SHA256 = 2
 115
 116
 117 class LispEID(object):
 118     """ Lisp endpoint identifier """
 119     def __init__(self, eid):
 120         self.eid = eid
 121         self._type = -1
 122
 123         # find out whether EID is ip prefix, or MAC
 124         try:
 125             self.prefix = ip_network(self.eid)
 126             self._type = LispEIDType.PREFIX
 127             return
 128         except ValueError:
 129             if self.eid.count(":") == 5:  # MAC address
 130                 self.mac = self.eid
 131                 self._type = LispEIDType.MAC
 132                 return
 133         raise Exception('Unsupported EID format {!s}!'.format(eid))
 134
 135     @property
 136     def eid_type(self):
 137         return self._type
 138
 139     @property
 140     def address(self):
 141         if self.eid_type == LispEIDType.PREFIX:
 142             return self.prefix
 143         elif self.eid_type == LispEIDType.MAC:
 144             return self.mac
 145         elif self.eid_type == LispEIDType.NSH:
 146             return Exception('Unimplemented')
 147
 148     @property
 149     def packed(self):
 150         if self.eid_type == LispEIDType.PREFIX:
 151             return {"type": self._type, "address": {"prefix": self.prefix}}
 152         elif self.eid_type == LispEIDType.MAC:
 153             return {"type": self._type, "address": {"mac": self.mac}}
 154         elif self.eid_type == LispEIDType.NSH:
 155             return Exception('Unimplemented')
 156
 157
 158 class LispKey(object):
 159     """ Lisp Key """
 160     def __init__(self, key_type, key):
 161         self._key_type = key_type
 162         self._key = key
 163
 164     @property
 165     def packed(self):
 166         return {"id": self._key_type, "key": self._key}
 167
 168
 169 class VppLispMapping(VppObject):
 170     """ Represents common features for remote and local LISP mapping in VPP """
 171
 172     def __init__(self, test, eid, vni=0, priority=1, weight=1):
 173         self._eid = LispEID(eid)
 174         self._test = test
 175         self._priority = priority
 176         self._weight = weight
 177         self._vni = vni
 178
 179     @property
 180     def test(self):
 181         return self._test
 182
 183     @property
 184     def vni(self):
 185         return self._vni
 186
 187     @property
 188     def eid(self):
 189         return self._eid
 190
 191     @property
 192     def priority(self):
 193         return self._priority
 194
 195     @property
 196     def weight(self):
 197         return self._weight
 198
 199     def get_lisp_mapping_dump_entry(self):
 200         return self.test.vapi.lisp_eid_table_dump(
 201             eid_set=1, vni=self._vni, eid=self._eid.packed)
 202
 203     def query_vpp_config(self):
 204         mapping = self.get_lisp_mapping_dump_entry()
 205         return mapping
 206
 207     def object_id(self):
 208         return 'lisp-mapping-[%s]-%s-%s-%s' % (
 209             self.vni, self.eid.address, self.priority, self.weight)
 210
 211
 212 class VppLocalMapping(VppLispMapping):
 213     """ LISP Local mapping """
 214     def __init__(self, test, eid, ls_name, vni=0, priority=1, weight=1,
 215                  key_id=LispKeyIdType.NONE, key=''):
 216         super(VppLocalMapping, self).__init__(test, eid, vni, priority, weight)
 217         self._ls_name = ls_name
 218         self._key = LispKey(key_id, key)
 219
 220     @property
 221     def ls_name(self):
 222         return self._ls_name
 223
 224     @property
 225     def key_id(self):
 226         return self._key_id
 227
 228     @property
 229     def key(self):
 230         return self._key
 231
 232     def add_vpp_config(self):
 233         self.test.vapi.lisp_add_del_local_eid(
 234                 locator_set_name=self._ls_name, eid=self._eid.packed,
 235                 vni=self._vni, key=self._key.packed)
 236         self._test.registry.register(self, self.test.logger)
 237
 238     def remove_vpp_config(self):
 239         self.test.vapi.lisp_add_del_local_eid(
 240                 locator_set_name=self._ls_name, eid=self._eid.packed,
 241                 vni=self._vni, is_add=0)
 242
 243     def object_id(self):
 244         return 'lisp-eid-local-mapping-%s[%d]' % (self._eid.address, self._vni)
 245
 246
 247 class LispRemoteLocator(object):
 248     def __init__(self, addr, priority=1, weight=1):
 249         self.addr = addr
 250         self.priority = priority
 251         self.weight = weight
 252
 253     @property
 254     def packed(self):
 255         return {"priority": self.priority, "weight": self.weight,
 256                 "ip_address": self.addr}
 257
 258
 259 class VppRemoteMapping(VppLispMapping):
 260
 261     def __init__(self, test, eid, rlocs=None, vni=0, priority=1, weight=1):
 262         super(VppRemoteMapping, self).__init__(test, eid, vni, priority,
 263                                                weight)
 264         self._rlocs = rlocs
 265
 266     @property
 267     def rlocs(self):
 268         rlocs = []
 269         for rloc in self._rlocs:
 270             rlocs.append(rloc.packed)
 271         return rlocs
 272
 273     def add_vpp_config(self):
 274         self.test.vapi.lisp_add_del_remote_mapping(
 275                 rlocs=self.rlocs, deid=self._eid.packed,
 276                 vni=self._vni, rloc_num=len(self._rlocs))
 277         self._test.registry.register(self, self.test.logger)
 278
 279     def remove_vpp_config(self):
 280         self.test.vapi.lisp_add_del_remote_mapping(
 281                 deid=self._eid.packed, vni=self._vni, is_add=0, rloc_num=0)
 282
 283     def object_id(self):
 284         return 'lisp-eid-remote-mapping-%s[%d]' % (self._eid.address,
 285                                                    self._vni)
 286
 287
 288 class VppLispAdjacency(VppObject):
 289     """ Represents LISP adjacency in VPP """
 290
 291     def __init__(self, test, leid, reid, vni=0):
 292         self._leid = LispEID(leid)
 293         self._reid = LispEID(reid)
 294         if self._leid.eid_type != self._reid.eid_type:
 295             raise Exception('remote and local EID are different types!')
 296         self._vni = vni
 297         self._test = test
 298
 299     @property
 300     def test(self):
 301         return self._test
 302
 303     @property
 304     def leid(self):
 305         return self._leid
 306
 307     @property
 308     def reid(self):
 309         return self._reid
 310
 311     @property
 312     def vni(self):
 313         return self._vni
 314
 315     def add_vpp_config(self):
 316         self.test.vapi.lisp_add_del_adjacency(
 317                 leid=self._leid.packed, reid=self._reid.packed, vni=self._vni)
 318         self._test.registry.register(self, self.test.logger)
 319
 320     @staticmethod
 321     def eid_equal(eid, eid_api):
 322         if eid.eid_type != eid_api.type:
 323             return False
 324
 325         if eid_api.type == LispEIDType.PREFIX:
 326             if eid.address.prefixlen != eid_api.address.prefix.prefixlen:
 327                 return False
 328
 329         if eid.address != eid_api.address:
 330             return False
 331
 332         return True
 333
 334     def query_vpp_config(self):
 335         res = self.test.vapi.lisp_adjacencies_get(vni=self._vni)
 336         for adj in res.adjacencies:
 337             if self.eid_equal(self._leid, adj.leid) and \
 338                     self.eid_equal(self._reid, adj.reid):
 339                 return True
 340         return False
 341
 342     def remove_vpp_config(self):
 343         self.test.vapi.lisp_add_del_adjacency(
 344                 leid=self._leid.packed, reid=self._reid.packed,
 345                 vni=self._vni, is_add=0)
 346
 347     def object_id(self):
 348         return 'lisp-adjacency-%s-%s[%d]' % (self._leid, self._reid, self._vni)