8800f732f8
- Make framework.py classes a subset of asfframework.py classes - Remove all packet related code from asfframework.py - Add test class and test case set up debug output to log - Repatriate packet tests from asf to test directory - Remove non-packet related code from framework.py and inherit them from asfframework.py classes - Clean up unused import variables - Re-enable BFD tests on Ubuntu 22.04 and fix intermittent test failures in echo_looped_back testcases (where # control packets verified but not guaranteed to be received during test) - Re-enable Wireguard tests on Ubuntu 22.04 and fix intermittent test failures in handshake ratelimiting testcases and event testcase - Run Wiregard testcase suites solo - Improve debug output in log.txt - Increase VCL/LDP post sleep timeout to allow iperf server to finish cleanly. - Fix pcap history files to be sorted by suite and testcase and ensure order/timestamp is correct based on creation in the testcase. - Decode pcap files for each suite and testcase for all errors or if configured via comandline option / env var - Improve vpp corefile detection to allow complete corefile generation - Disable vm vpp interfaces testcases on debian11 - Clean up failed unittest dir when retrying failed testcases and unify testname directory and failed linknames into framwork functions Type: test Change-Id: I0764f79ea5bb639d278bf635ed2408d4d5220e1e Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
386 lines
9.8 KiB
Python
386 lines
9.8 KiB
Python
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)
|