git: f63825ff21a3 - main - testing: improve python vnet wrapper.
- Go to: [ bottom of page ] [ top of archives ] [ this month ]
Date: Thu, 29 Dec 2022 19:59:21 UTC
The branch main has been updated by melifaro:
URL: https://cgit.FreeBSD.org/src/commit/?id=f63825ff21a3bee2630ea8b0ed27a4583cc4242b
commit f63825ff21a3bee2630ea8b0ed27a4583cc4242b
Author: Alexander V. Chernikov <melifaro@FreeBSD.org>
AuthorDate: 2022-12-29 19:07:34 +0000
Commit: Alexander V. Chernikov <melifaro@FreeBSD.org>
CommitDate: 2022-12-29 19:59:11 +0000
testing: improve python vnet wrapper.
* Derive jail name from class name and method name, instead of just
method name. This change reduces the chances of different tests
clashing.
Old: 'jail_test_one'. New: 'pytest:TestExampleSimplest:test_one'
* Simplify vnetX_handler() method signature by skipping obj_map (unused)
and pipe. The latter can be accessed as the vnet property.
* Add `send_object()` method as a pair to the `wait_object` inside the
VnetTestTemplate class.
* Add `test_id` property to the BaseTest method. Previously it was
provided only for the VnetTestTemplate class. This change makes
the identifier easily accessible for all users.
MFC after: 2 weeks
---
tests/atf_python/sys/net/vnet.py | 92 +++++++++++++++++++++--------------
tests/atf_python/utils.py | 12 ++++-
tests/sys/netinet6/test_ip6_output.py | 30 ++++++------
3 files changed, 82 insertions(+), 52 deletions(-)
diff --git a/tests/atf_python/sys/net/vnet.py b/tests/atf_python/sys/net/vnet.py
index faae58e95b6f..aca1b53d388c 100644
--- a/tests/atf_python/sys/net/vnet.py
+++ b/tests/atf_python/sys/net/vnet.py
@@ -12,7 +12,8 @@ from typing import List
from typing import NamedTuple
from atf_python.sys.net.tools import ToolsHelper
-from atf_python.utils import libc, BaseTest
+from atf_python.utils import BaseTest
+from atf_python.utils import libc
def run_cmd(cmd: str, verbose=True) -> str:
@@ -20,11 +21,20 @@ def run_cmd(cmd: str, verbose=True) -> str:
return os.popen(cmd).read()
+def get_topology_id(test_id: str) -> str:
+ """
+ Gets a unique topology id based on the pytest test_id.
+ "test_ip6_output.py::TestIP6Output::test_output6_pktinfo[ipandif]" ->
+ "TestIP6Output:test_output6_pktinfo[ipandif]"
+ """
+ return ":".join(test_id.split("::")[-2:])
+
+
def convert_test_name(test_name: str) -> str:
"""Convert test name to a string that can be used in the file/jail names"""
ret = ""
for char in test_name:
- if char.isalnum() or char in ("_", "-"):
+ if char.isalnum() or char in ("_", "-", ":"):
ret += char
elif char in ("["):
ret += "_"
@@ -140,9 +150,7 @@ class VnetInterface(object):
class IfaceFactory(object):
INTERFACES_FNAME = "created_ifaces.lst"
- def __init__(self, test_name: str):
- self.test_name = test_name
- self.test_id = convert_test_name(test_name)
+ def __init__(self):
self.file_name = self.INTERFACES_FNAME
def _register_iface(self, iface_name: str):
@@ -213,9 +221,8 @@ class VnetInstance(object):
class VnetFactory(object):
JAILS_FNAME = "created_jails.lst"
- def __init__(self, test_name: str):
- self.test_name = test_name
- self.test_id = convert_test_name(test_name)
+ def __init__(self, topology_id: str):
+ self.topology_id = topology_id
self.file_name = self.JAILS_FNAME
self._vnets: List[str] = []
@@ -240,7 +247,7 @@ class VnetFactory(object):
return not_matched
def create_vnet(self, vnet_alias: str, ifaces: List[VnetInterface]):
- vnet_name = "jail_{}".format(self.test_id)
+ vnet_name = "pytest:{}".format(convert_test_name(self.topology_id))
if self._vnets:
# add number to distinguish jails
vnet_name = "{}_{}".format(vnet_name, len(self._vnets) + 1)
@@ -248,10 +255,13 @@ class VnetFactory(object):
cmd = "/usr/sbin/jail -i -c name={} persist vnet {}".format(
vnet_name, iface_cmds
)
- jid_str = run_cmd(cmd)
- jid = int(jid_str)
- if jid <= 0:
- raise Exception("Jail creation failed, output: {}".format(jid))
+ jid = 0
+ try:
+ jid_str = run_cmd(cmd)
+ jid = int(jid_str)
+ except ValueError as e:
+ print("Jail creation failed, output: {}".format(jid_str))
+ raise
self._register_vnet(vnet_name)
# Run expedited version of routing
@@ -268,11 +278,11 @@ class VnetFactory(object):
try:
with open(self.file_name) as f:
for line in f:
- jail_name = line.strip()
+ vnet_name = line.strip()
ToolsHelper.print_output(
- "/usr/sbin/jexec {} ifconfig -l".format(jail_name)
+ "/usr/sbin/jexec {} ifconfig -l".format(vnet_name)
)
- run_cmd("/usr/sbin/jail -r {}".format(line.strip()))
+ run_cmd("/usr/sbin/jail -r {}".format(vnet_name))
os.unlink(self.JAILS_FNAME)
except OSError:
pass
@@ -283,6 +293,12 @@ class SingleInterfaceMap(NamedTuple):
vnet_aliases: List[str]
+class ObjectsMap(NamedTuple):
+ iface_map: Dict[str, SingleInterfaceMap] # keyed by ifX
+ vnet_map: Dict[str, VnetInstance] # keyed by vnetX
+ topo_map: Dict # self.TOPOLOGY
+
+
class VnetTestTemplate(BaseTest):
TOPOLOGY = {}
@@ -297,8 +313,10 @@ class VnetTestTemplate(BaseTest):
"""
vnet.attach()
print("# setup_vnet({})".format(vnet.name))
+ if pipe is not None:
+ vnet.set_pipe(pipe)
- topo = obj_map["topo_map"]
+ topo = obj_map.topo_map
ipv6_ifaces = []
# Disable DAD
if not vnet.need_dad:
@@ -306,7 +324,7 @@ class VnetTestTemplate(BaseTest):
for iface in vnet.ifaces:
# check index of vnet within an interface
# as we have prefixes for both ends of the interface
- iface_map = obj_map["iface_map"][iface.alias]
+ iface_map = obj_map.iface_map[iface.alias]
idx = iface_map.vnet_aliases.index(vnet.alias)
prefixes6 = topo[iface.alias].get("prefixes6", [])
prefixes4 = topo[iface.alias].get("prefixes4", [])
@@ -327,14 +345,14 @@ class VnetTestTemplate(BaseTest):
# Do unbuffered stdout for children
# so the logs are present if the child hangs
sys.stdout.reconfigure(line_buffering=True)
- handler(vnet, obj_map, pipe)
+ handler(vnet)
- def setup_topology(self, topo: Dict, test_name: str):
+ 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(test_name)
- vnet_factory = VnetFactory(test_name)
+ iface_factory = IfaceFactory()
+ 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")
@@ -381,19 +399,18 @@ class VnetTestTemplate(BaseTest):
)
)
print()
- return {"iface_map": iface_map, "vnet_map": vnet_map, "topo_map": topo}
+ return ObjectsMap(iface_map, vnet_map, topo)
- def setup_method(self, method):
+ def setup_method(self, _method):
"""Sets up all the required topology and handlers for the given test"""
- # 'test_ip6_output.py::TestIP6Output::test_output6_pktinfo[ipandif] (setup)'
- test_id = os.environ.get("PYTEST_CURRENT_TEST").split(" ")[0]
- test_name = test_id.split("::")[-1]
- self.check_constraints()
+ super().setup_method(_method)
+ # TestIP6Output.test_output6_pktinfo[ipandif]
+ topology_id = get_topology_id(self.test_id)
topology = self.TOPOLOGY
# First, setup kernel objects - interfaces & vnets
- obj_map = self.setup_topology(topology, test_name)
+ obj_map = self.setup_topology(topology, topology_id)
main_vnet = None # one without subprocess handler
- for vnet_alias, vnet in obj_map["vnet_map"].items():
+ for vnet_alias, vnet in obj_map.vnet_map.items():
if self._get_vnet_handler(vnet_alias):
# Need subprocess to run
parent_pipe, child_pipe = Pipe()
@@ -417,23 +434,26 @@ class VnetTestTemplate(BaseTest):
self.vnet = main_vnet
self._setup_vnet(main_vnet, obj_map, None)
# Save state for the main handler
- self.iface_map = obj_map["iface_map"]
- self.vnet_map = obj_map["vnet_map"]
+ self.iface_map = obj_map.iface_map
+ self.vnet_map = obj_map.vnet_map
def cleanup(self, test_id: str):
# pytest test id: file::class::test_name
- test_name = test_id.split("::")[-1]
+ topology_id = get_topology_id(self.test_id)
print("==== vnet cleanup ===")
- print("# test_name: '{}'".format(test_name))
- VnetFactory(test_name).cleanup()
- IfaceFactory(test_name).cleanup()
+ print("# topology_id: '{}'".format(topology_id))
+ VnetFactory(topology_id).cleanup()
+ IfaceFactory().cleanup()
def wait_object(self, pipe, timeout=5):
if pipe.poll(timeout):
return pipe.recv()
raise TimeoutError
+ def send_object(self, pipe, obj):
+ pipe.send(obj)
+
@property
def curvnet(self):
pass
diff --git a/tests/atf_python/utils.py b/tests/atf_python/utils.py
index 12cd56c10149..17824262b1fd 100644
--- a/tests/atf_python/utils.py
+++ b/tests/atf_python/utils.py
@@ -42,5 +42,15 @@ class BaseTest(object):
"kernel module '{}' not available: {}".format(mod_name, err_str)
)
- def check_constraints(self):
+ @property
+ def test_id(self):
+ # 'test_ip6_output.py::TestIP6Output::test_output6_pktinfo[ipandif] (setup)'
+ return os.environ.get("PYTEST_CURRENT_TEST").split(" ")[0]
+
+ def setup_method(self, method):
+ """Run all pre-requisits for the test execution"""
self._check_modules()
+
+ def cleanup(self, test_id: str):
+ """Cleanup all test resources here"""
+ pass
diff --git a/tests/sys/netinet6/test_ip6_output.py b/tests/sys/netinet6/test_ip6_output.py
index 35adb6a7137a..fc821606a726 100644
--- a/tests/sys/netinet6/test_ip6_output.py
+++ b/tests/sys/netinet6/test_ip6_output.py
@@ -73,24 +73,24 @@ class BaseTestIP6Ouput(VnetTestTemplate):
}
DEFAULT_PORT = 45365
- def _vnet2_handler(self, vnet, obj_map, pipe, ip: str, os_ifname: str = None):
+ def _vnet2_handler(self, vnet, ip: str, os_ifname: str = None):
"""Generic listener that sends first received packet with metadata
back to the sender via pipw
"""
ll_data = ToolsHelper.get_linklocals()
# Start listener
ss = VerboseSocketServer(ip, self.DEFAULT_PORT, os_ifname)
- pipe.send(ll_data)
+ vnet.pipe.send(ll_data)
tx_obj = ss.recv()
tx_obj["dst_iface_alias"] = vnet.iface_map[tx_obj["dst_iface"]].alias
- pipe.send(tx_obj)
+ vnet.pipe.send(tx_obj)
class TestIP6Output(BaseTestIP6Ouput):
- def vnet2_handler(self, vnet, obj_map, pipe):
+ def vnet2_handler(self, vnet):
ip = str(vnet.iface_alias_map["if2"].first_ipv6.ip)
- self._vnet2_handler(vnet, obj_map, pipe, ip, None)
+ self._vnet2_handler(vnet, ip, None)
@pytest.mark.require_user("root")
def test_output6_base(self):
@@ -221,14 +221,14 @@ class TestIP6Output(BaseTestIP6Ouput):
class TestIP6OutputLL(BaseTestIP6Ouput):
- def vnet2_handler(self, vnet, obj_map, pipe):
+ def vnet2_handler(self, vnet):
"""Generic listener that sends first received packet with metadata
back to the sender via pipw
"""
os_ifname = vnet.iface_alias_map["if2"].name
ll_data = ToolsHelper.get_linklocals()
ll_ip, _ = ll_data[os_ifname][0]
- self._vnet2_handler(vnet, obj_map, pipe, ll_ip, os_ifname)
+ self._vnet2_handler(vnet, ll_ip, os_ifname)
@pytest.mark.require_user("root")
def test_output6_linklocal(self):
@@ -258,12 +258,12 @@ class TestIP6OutputLL(BaseTestIP6Ouput):
class TestIP6OutputNhopLL(BaseTestIP6Ouput):
- def vnet2_handler(self, vnet, obj_map, pipe):
+ def vnet2_handler(self, vnet):
"""Generic listener that sends first received packet with metadata
back to the sender via pipw
"""
ip = str(vnet.iface_alias_map["if2"].first_ipv6.ip)
- self._vnet2_handler(vnet, obj_map, pipe, ip, None)
+ self._vnet2_handler(vnet, ip, None)
@pytest.mark.require_user("root")
def test_output6_nhop_linklocal(self):
@@ -296,11 +296,11 @@ class TestIP6OutputNhopLL(BaseTestIP6Ouput):
class TestIP6OutputScope(BaseTestIP6Ouput):
- def vnet2_handler(self, vnet, obj_map, pipe):
+ def vnet2_handler(self, vnet):
"""Generic listener that sends first received packet with metadata
back to the sender via pipw
"""
- bind_ip, bind_ifp = self.wait_object(pipe)
+ bind_ip, bind_ifp = self.wait_object(vnet.pipe)
if bind_ip is None:
os_ifname = vnet.iface_alias_map[bind_ifp].name
ll_data = ToolsHelper.get_linklocals()
@@ -308,7 +308,7 @@ class TestIP6OutputScope(BaseTestIP6Ouput):
if bind_ifp is not None:
bind_ifp = vnet.iface_alias_map[bind_ifp].name
print("## BIND {}%{}".format(bind_ip, bind_ifp))
- self._vnet2_handler(vnet, obj_map, pipe, bind_ip, bind_ifp)
+ self._vnet2_handler(vnet, bind_ip, bind_ifp)
@pytest.mark.parametrize(
"params",
@@ -402,10 +402,10 @@ class TestIP6OutputScope(BaseTestIP6Ouput):
class TestIP6OutputMulticast(BaseTestIP6Ouput):
- def vnet2_handler(self, vnet, obj_map, pipe):
- group = self.wait_object(pipe)
+ def vnet2_handler(self, vnet):
+ group = self.wait_object(vnet.pipe)
os_ifname = vnet.iface_alias_map["if2"].name
- self._vnet2_handler(vnet, obj_map, pipe, group, os_ifname)
+ self._vnet2_handler(vnet, group, os_ifname)
@pytest.mark.parametrize("group_scope", ["ff02", "ff05", "ff08", "ff0e"])
@pytest.mark.require_user("root")