system/freebsd-src
system/freebsd-src
1
0
Fork 0
mirror of https://github.com/freebsd/freebsd-src synced 2024-07-23 19:28:36 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

testing: improve python vnet wrapper.

Browse source 
* 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
This commit is contained in:
Alexander V. Chernikov 2022-12-29 19:07:34 +00:00
parent 7063b9974f
commit f63825ff21
3 changed files with 82 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

92
tests/atf_python/sys/net/vnet.py
View file

@ -12,7 +12,8 @@
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 @@ def has_tentative(self) -> bool:
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 @@ def attach(self):
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 @@ def _wait_interfaces(vnet_name: str, ifaces: List[str]) -> List[str]:
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 @@ def create_vnet(self, vnet_alias: str, ifaces: List[VnetInterface]):
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 @@ def cleanup(self):
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 @@ def _setup_vnet(self, vnet: VnetInstance, obj_map: Dict, pipe):
"""
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 @@ def _setup_vnet(self, vnet: VnetInstance, obj_map: Dict, pipe):
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 @@ def _setup_vnet(self, vnet: VnetInstance, obj_map: Dict, pipe):
# 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 @@ def setup_topology(self, topo: Dict, test_name: str):
)
)
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 @@ def setup_method(self, method):
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

12
tests/atf_python/utils.py
View file

@ -42,5 +42,15 @@ def _check_modules(self):
"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

30
tests/sys/netinet6/test_ip6_output.py
View file

@ -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 @@ def test_output6_pktinfo(self, params):
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 @@ def test_output6_linklocal(self):
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 @@ def test_output6_nhop_linklocal(self):
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 @@ def vnet2_handler(self, vnet, obj_map, pipe):
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 @@ def test_output6_linklocal_scope(self, params):
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")