--- /dev/null
+import socket
+from ipaddress import ip_network
+
+from vpp_object import VppObject
+
+
+class VppLispLocatorSet(VppObject):
+ """Represents LISP locator set in VPP"""
+
+ def __init__(self, test, ls_name):
+ self._test = test
+ self._ls_name = ls_name
+
+ @property
+ def test(self):
+ return self._test
+
+ @property
+ def ls_name(self):
+ return self._ls_name
+
+ def add_vpp_config(self):
+ self.test.vapi.lisp_add_del_locator_set(locator_set_name=self._ls_name)
+ self._test.registry.register(self, self.test.logger)
+
+ def get_lisp_locator_sets_dump_entry(self):
+ result = self.test.vapi.lisp_locator_set_dump()
+ for ls in result:
+ if ls.ls_name.strip("\x00") == self._ls_name:
+ return ls
+ return None
+
+ def query_vpp_config(self):
+ return self.get_lisp_locator_sets_dump_entry() is not None
+
+ def remove_vpp_config(self):
+ self.test.vapi.lisp_add_del_locator_set(
+ locator_set_name=self._ls_name, is_add=0
+ )
+
+ def object_id(self):
+ return "lisp-locator-set-%s" % self._ls_name
+
+
+class VppLispLocator(VppObject):
+ """Represents LISP locator in VPP"""
+
+ def __init__(self, test, sw_if_index, ls_name, priority=1, weight=1):
+ self._test = test
+ self._sw_if_index = sw_if_index
+ self._ls_name = ls_name
+ self._priority = priority
+ self._weight = weight
+
+ @property
+ def test(self):
+ """Test which created this locator"""
+ return self._test
+
+ @property
+ def ls_name(self):
+ """Locator set name"""
+ return self._ls_name
+
+ @property
+ def sw_if_index(self):
+ return self._sw_if_index
+
+ @property
+ def priority(self):
+ return self._priority
+
+ @property
+ def weight(self):
+ return self._weight
+
+ def add_vpp_config(self):
+ self.test.vapi.lisp_add_del_locator(
+ locator_set_name=self._ls_name,
+ sw_if_index=self._sw_if_index,
+ priority=self._priority,
+ weight=self._weight,
+ )
+ self._test.registry.register(self, self.test.logger)
+
+ def get_lisp_locator_dump_entry(self):
+ locators = self.test.vapi.lisp_locator_dump(
+ is_index_set=0, ls_name=self._ls_name
+ )
+ for locator in locators:
+ if locator.sw_if_index == self._sw_if_index:
+ return locator
+ return None
+
+ def query_vpp_config(self):
+ locator = self.get_lisp_locator_dump_entry()
+ return locator is not None
+
+ def remove_vpp_config(self):
+ self.test.vapi.lisp_add_del_locator(
+ locator_set_name=self._ls_name,
+ sw_if_index=self._sw_if_index,
+ priority=self._priority,
+ weight=self._weight,
+ is_add=0,
+ )
+ self._test.registry.register(self, self.test.logger)
+
+ def object_id(self):
+ return "lisp-locator-%s-%d" % (self._ls_name, self._sw_if_index)
+
+
+class LispEIDType:
+ PREFIX = 0
+ MAC = 1
+ NSH = 2
+
+
+class LispKeyIdType:
+ NONE = 0
+ SHA1 = 1
+ SHA256 = 2
+
+
+class LispEID:
+ """Lisp endpoint identifier"""
+
+ def __init__(self, eid):
+ self.eid = eid
+ self._type = -1
+
+ # find out whether EID is ip prefix, or MAC
+ try:
+ self.prefix = ip_network(self.eid)
+ self._type = LispEIDType.PREFIX
+ return
+ except ValueError:
+ if self.eid.count(":") == 5: # MAC address
+ self.mac = self.eid
+ self._type = LispEIDType.MAC
+ return
+ raise Exception("Unsupported EID format {!s}!".format(eid))
+
+ @property
+ def eid_type(self):
+ return self._type
+
+ @property
+ def address(self):
+ if self.eid_type == LispEIDType.PREFIX:
+ return self.prefix
+ elif self.eid_type == LispEIDType.MAC:
+ return self.mac
+ elif self.eid_type == LispEIDType.NSH:
+ return Exception("Unimplemented")
+
+ @property
+ def packed(self):
+ if self.eid_type == LispEIDType.PREFIX:
+ return {"type": self._type, "address": {"prefix": self.prefix}}
+ elif self.eid_type == LispEIDType.MAC:
+ return {"type": self._type, "address": {"mac": self.mac}}
+ elif self.eid_type == LispEIDType.NSH:
+ return Exception("Unimplemented")
+
+
+class LispKey:
+ """Lisp Key"""
+
+ def __init__(self, key_type, key):
+ self._key_type = key_type
+ self._key = key
+
+ @property
+ def packed(self):
+ return {"id": self._key_type, "key": self._key}
+
+
+class VppLispMapping(VppObject):
+ """Represents common features for remote and local LISP mapping in VPP"""
+
+ def __init__(self, test, eid, vni=0, priority=1, weight=1):
+ self._eid = LispEID(eid)
+ self._test = test
+ self._priority = priority
+ self._weight = weight
+ self._vni = vni
+
+ @property
+ def test(self):
+ return self._test
+
+ @property
+ def vni(self):
+ return self._vni
+
+ @property
+ def eid(self):
+ return self._eid
+
+ @property
+ def priority(self):
+ return self._priority
+
+ @property
+ def weight(self):
+ return self._weight
+
+ def get_lisp_mapping_dump_entry(self):
+ return self.test.vapi.lisp_eid_table_dump(
+ eid_set=1, vni=self._vni, eid=self._eid.packed
+ )
+
+ def query_vpp_config(self):
+ mapping = self.get_lisp_mapping_dump_entry()
+ return mapping
+
+ def object_id(self):
+ return "lisp-mapping-[%s]-%s-%s-%s" % (
+ self.vni,
+ self.eid.address,
+ self.priority,
+ self.weight,
+ )
+
+
+class VppLocalMapping(VppLispMapping):
+ """LISP Local mapping"""
+
+ def __init__(
+ self,
+ test,
+ eid,
+ ls_name,
+ vni=0,
+ priority=1,
+ weight=1,
+ key_id=LispKeyIdType.NONE,
+ key="",
+ ):
+ super(VppLocalMapping, self).__init__(test, eid, vni, priority, weight)
+ self._ls_name = ls_name
+ self._key = LispKey(key_id, key)
+
+ @property
+ def ls_name(self):
+ return self._ls_name
+
+ @property
+ def key_id(self):
+ return self._key_id
+
+ @property
+ def key(self):
+ return self._key
+
+ def add_vpp_config(self):
+ self.test.vapi.lisp_add_del_local_eid(
+ locator_set_name=self._ls_name,
+ eid=self._eid.packed,
+ vni=self._vni,
+ key=self._key.packed,
+ )
+ self._test.registry.register(self, self.test.logger)
+
+ def remove_vpp_config(self):
+ self.test.vapi.lisp_add_del_local_eid(
+ locator_set_name=self._ls_name,
+ eid=self._eid.packed,
+ vni=self._vni,
+ is_add=0,
+ )
+
+ def object_id(self):
+ return "lisp-eid-local-mapping-%s[%d]" % (self._eid.address, self._vni)
+
+
+class LispRemoteLocator:
+ def __init__(self, addr, priority=1, weight=1):
+ self.addr = addr
+ self.priority = priority
+ self.weight = weight
+
+ @property
+ def packed(self):
+ return {
+ "priority": self.priority,
+ "weight": self.weight,
+ "ip_address": self.addr,
+ }
+
+
+class VppRemoteMapping(VppLispMapping):
+ def __init__(self, test, eid, rlocs=None, vni=0, priority=1, weight=1):
+ super(VppRemoteMapping, self).__init__(test, eid, vni, priority, weight)
+ self._rlocs = rlocs
+
+ @property
+ def rlocs(self):
+ rlocs = []
+ for rloc in self._rlocs:
+ rlocs.append(rloc.packed)
+ return rlocs
+
+ def add_vpp_config(self):
+ self.test.vapi.lisp_add_del_remote_mapping(
+ rlocs=self.rlocs,
+ deid=self._eid.packed,
+ vni=self._vni,
+ rloc_num=len(self._rlocs),
+ )
+ self._test.registry.register(self, self.test.logger)
+
+ def remove_vpp_config(self):
+ self.test.vapi.lisp_add_del_remote_mapping(
+ deid=self._eid.packed, vni=self._vni, is_add=0, rloc_num=0
+ )
+
+ def object_id(self):
+ return "lisp-eid-remote-mapping-%s[%d]" % (self._eid.address, self._vni)
+
+
+class VppLispAdjacency(VppObject):
+ """Represents LISP adjacency in VPP"""
+
+ def __init__(self, test, leid, reid, vni=0):
+ self._leid = LispEID(leid)
+ self._reid = LispEID(reid)
+ if self._leid.eid_type != self._reid.eid_type:
+ raise Exception("remote and local EID are different types!")
+ self._vni = vni
+ self._test = test
+
+ @property
+ def test(self):
+ return self._test
+
+ @property
+ def leid(self):
+ return self._leid
+
+ @property
+ def reid(self):
+ return self._reid
+
+ @property
+ def vni(self):
+ return self._vni
+
+ def add_vpp_config(self):
+ self.test.vapi.lisp_add_del_adjacency(
+ leid=self._leid.packed, reid=self._reid.packed, vni=self._vni
+ )
+ self._test.registry.register(self, self.test.logger)
+
+ @staticmethod
+ def eid_equal(eid, eid_api):
+ if eid.eid_type != eid_api.type:
+ return False
+
+ if eid_api.type == LispEIDType.PREFIX:
+ if eid.address.prefixlen != eid_api.address.prefix.prefixlen:
+ return False
+
+ if eid.address != eid_api.address:
+ return False
+
+ return True
+
+ def query_vpp_config(self):
+ res = self.test.vapi.lisp_adjacencies_get(vni=self._vni)
+ for adj in res.adjacencies:
+ if self.eid_equal(self._leid, adj.leid) and self.eid_equal(
+ self._reid, adj.reid
+ ):
+ return True
+ return False
+
+ def remove_vpp_config(self):
+ self.test.vapi.lisp_add_del_adjacency(
+ leid=self._leid.packed, reid=self._reid.packed, vni=self._vni, is_add=0
+ )
+
+ def object_id(self):
+ return "lisp-adjacency-%s-%s[%d]" % (self._leid, self._reid, self._vni)
Code Review / vpp.git / blobdiff