+++ /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