The branch main has been updated by melifaro:
URL:
https://cgit.FreeBSD.org/src/commit/?id=584ad4126c76721b7f5da39d4ca4818ded7d64db
commit 584ad4126c76721b7f5da39d4ca4818ded7d64db
Author: Alexander V. Chernikov <melif...@freebsd.org>
AuthorDate: 2023-05-16 15:00:45 +0000
Commit: Alexander V. Chernikov <melif...@freebsd.org>
CommitDate: 2023-05-16 15:00:45 +0000
testing: improve vnet support in pytest
* Allow vnet object to be directly referenced
(self.vnet1 vs self.vnet_map["vnet1"])
* Allow iface object to be directly reference
(vnet.bridge vs vnet.iface_alias_map["bridge"])
* Allow arbitrary interface alias names insted of ifX
* Add wait_objects_any() method for waiting object from
multiple vnets
* Add wait() method for indefinite sleep on vnet handlers
MFC after: 2 weeks
---
tests/atf_python/sys/net/vnet.py | 46 +++++++++++++++++++++++++++++++---------
1 file changed, 36 insertions(+), 10 deletions(-)
diff --git a/tests/atf_python/sys/net/vnet.py b/tests/atf_python/sys/net/vnet.py
index a2884b0d7ce2..8942e6839b35 100644
--- a/tests/atf_python/sys/net/vnet.py
+++ b/tests/atf_python/sys/net/vnet.py
@@ -1,11 +1,12 @@
#!/usr/local/bin/python3
import copy
import ipaddress
-import re
import os
+import re
import socket
import sys
import time
+from multiprocessing import connection
from multiprocessing import Pipe
from multiprocessing import Process
from typing import Dict
@@ -218,6 +219,8 @@ class VnetInstance(object):
iface.set_jailed(True)
self.iface_alias_map[iface.alias] = iface
self.iface_map[iface.name] = iface
+ # Allow reference to interfce aliases as attributes
+ setattr(self, iface.alias, iface)
self.need_dad = False # Disable duplicate address detection by default
self.attached = False
self.pipe = None
@@ -364,7 +367,8 @@ class VnetTestTemplate(BaseTest):
if prefixes6:
iface.enable_ipv6()
for prefix in prefixes6 + prefixes4:
- iface.setup_addr(prefix[idx])
+ if prefix[idx]:
+ iface.setup_addr(prefix[idx])
for iface in ipv6_ifaces:
while iface.has_tentative():
time.sleep(0.1)
@@ -378,18 +382,28 @@ class VnetTestTemplate(BaseTest):
self.drop_privileges()
handler(vnet)
+ def _get_topo_ifmap(self, topo: Dict):
+ iface_factory = IfaceFactory()
+ iface_map: Dict[str, SingleInterfaceMap] = {}
+ iface_aliases = set()
+ for obj_name, obj_data in topo.items():
+ if obj_name.startswith("vnet"):
+ for iface_alias in obj_data["ifaces"]:
+ iface_aliases.add(iface_alias)
+ for iface_alias in iface_aliases:
+ print("Creating {}".format(iface_alias))
+ iface_data = topo[iface_alias]
+ iface_type = iface_data.get("type", "epair")
+ ifaces = iface_factory.create_iface(iface_alias, iface_type)
+ smap = SingleInterfaceMap(ifaces, [])
+ iface_map[iface_alias] = smap
+ return iface_map
+
def setup_topology(self, topo: Dict, topology_id: str):
"""Creates jails & interfaces for the provided topology"""
- iface_map: Dict[str, SingleInterfaceMap] = {}
vnet_map = {}
- iface_factory = IfaceFactory()
vnet_factory = VnetFactory(topology_id)
- for obj_name, obj_data in topo.items():
- if obj_name.startswith("if"):
- iface_type = obj_data.get("type", "epair")
- ifaces = iface_factory.create_iface(obj_name, iface_type)
- smap = SingleInterfaceMap(ifaces, [])
- iface_map[obj_name] = smap
+ iface_map = self._get_topo_ifmap(topo)
for obj_name, obj_data in topo.items():
if obj_name.startswith("vnet"):
vnet_ifaces = []
@@ -401,6 +415,8 @@ class VnetTestTemplate(BaseTest):
vnet_ifaces.append(iface_map[iface_alias].ifaces[idx])
vnet = vnet_factory.create_vnet(obj_name, vnet_ifaces)
vnet_map[obj_name] = vnet
+ # Allow reference to VNETs as attributes
+ setattr(self, obj_name, vnet)
# Debug output
print("============= TEST TOPOLOGY =============")
for vnet_alias, vnet in vnet_map.items():
@@ -484,9 +500,19 @@ class VnetTestTemplate(BaseTest):
return pipe.recv()
raise TimeoutError
+ def wait_objects_any(self, pipe_list, timeout=5):
+ objects = connection.wait(pipe_list, timeout)
+ if objects:
+ return objects[0].recv()
+ raise TimeoutError
+
def send_object(self, pipe, obj):
pipe.send(obj)
+ def wait(self):
+ while True:
+ time.sleep(1)
+
@property
def curvnet(self):
pass
