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