testing: add support for using custom interfaces in pytest framework.

MFC after:	2 weeks
This commit is contained in:
Alexander V. Chernikov
2023-05-15 10:44:20 +00:00
parent 06db6a9dce
commit f3065e767d
2 changed files with 23 additions and 7 deletions
+15 -6
View File
@@ -151,7 +151,7 @@ def has_tentative(self) -> bool:
class IfaceFactory(object):
INTERFACES_FNAME = "created_ifaces.lst"
AUTODELETE_TYPES = ("epair", "lo", "tap", "tun")
AUTODELETE_TYPES = ("epair", "gif", "gre", "lo", "tap", "tun")
def __init__(self):
self.file_name = self.INTERFACES_FNAME
@@ -386,8 +386,9 @@ def setup_topology(self, topo: Dict, topology_id: str):
vnet_factory = VnetFactory(topology_id)
for obj_name, obj_data in topo.items():
if obj_name.startswith("if"):
epair_ifaces = iface_factory.create_iface(obj_name, "epair")
smap = SingleInterfaceMap(epair_ifaces, [])
iface_type = obj_data.get("type", "epair")
ifaces = iface_factory.create_iface(obj_name, iface_type)
smap = SingleInterfaceMap(ifaces, [])
iface_map[obj_name] = smap
for obj_name, obj_data in topo.items():
if obj_name.startswith("vnet"):
@@ -494,17 +495,25 @@ def curvnet(self):
class SingleVnetTestTemplate(VnetTestTemplate):
IPV6_PREFIXES: List[str] = []
IPV4_PREFIXES: List[str] = []
IFTYPE = "epair"
def setup_method(self, method):
def _setup_default_topology(self):
topology = copy.deepcopy(
{
"vnet1": {"ifaces": ["if1"]},
"if1": {"prefixes4": [], "prefixes6": []},
"if1": {"type": self.IFTYPE, "prefixes4": [], "prefixes6": []},
}
)
for prefix in self.IPV6_PREFIXES:
topology["if1"]["prefixes6"].append((prefix,))
for prefix in self.IPV4_PREFIXES:
topology["if1"]["prefixes4"].append((prefix,))
self.TOPOLOGY = topology
return topology
def setup_method(self, method):
if not getattr(self, "TOPOLOGY", None):
self.TOPOLOGY = self._setup_default_topology()
else:
names = self.TOPOLOGY.keys()
assert len([n for n in names if n.startswith("vnet")]) == 1
super().setup_method(method)
+8 -1
View File
@@ -302,7 +302,7 @@ def get_genl_family_id(self, family_name):
hdr = Nlmsghdr(
nlmsg_type=NlConst.GENL_ID_CTRL,
nlmsg_flags=NlmBaseFlags.NLM_F_REQUEST.value,
nlmsg_seq = self.helper.get_seq(),
nlmsg_seq=self.helper.get_seq(),
)
ghdr = GenlMsgHdr(cmd=GenlCtrlMsgType.CTRL_CMD_GETFAMILY.value)
nla = NlAttrStr(GenlCtrlAttrType.CTRL_ATTR_FAMILY_NAME, family_name)
@@ -342,6 +342,13 @@ def read_message(self) -> bytes:
self._data = self._data[hdr.nlmsg_len:]
return self.parse_message(raw_msg)
def get_reply(self, tx_msg):
self.write_message(tx_msg)
while True:
rx_msg = self.read_message()
if tx_msg.nl_hdr.nlmsg_seq == rx_msg.nl_hdr.nlmsg_seq:
return rx_msg
class NetlinkMultipartIterator(object):
def __init__(self, obj, seq_number: int, msg_type):