mirror of
https://github.com/opnsense/src.git
synced 2026-02-22 01:11:30 -05:00
cfc9cf9baf
Notable amount of tests related to the packet IO require two VNET jails
for proper testing and avoiding side effects for the host system.
Additionally, it is often required to run actions in the jails seme-sequentially
- waiting for the listener initialisation can be an example of such
dependency.
This change extends pytest vnet framework to allow defining multi-vnet
multi-epair topologies in declarative style, without any need to bother
about jail or repair names. All jail creation/teardown, interface
creation/teardown and address assignments are handled automatically.
Example:
TOPOLOGY = {
"vnet1": {"ifaces": ["if1", "if2", "if3"]},
"vnet2": {"ifaces": ["if1", "if2", "if3"]},
"if1": {"prefixes6": [("2001:db8:a::1/64", "2001:db8:a::2/64")]},
"if2": {"prefixes6": [("2001:db8:b::1/64", "2001:db8:b::2/64")]},
"if3": {"prefixes6": [("2001:db8:c::1/64", "2001:db8:c::2/64")]},
}
def vnet2_handler(self, vnet, obj_map, pipe):
ss = VerboseSocketServer("::", self.DEFAULT_PORT)
pipe.send("READY")
def test_output6_base(self):
self.wait_object(second_vnet.pipe)
The definitions above will create 2 vnets ("jail_test_output6_base",
"jail_test_output6_base_2"), 3 epairs, attached to both first and
second jails, set up the IP addresses for each epair, spawn another
process for vnet2_handler and pass control to vnet2_handler and
test_output6_base. Both processes can pass objects between each
other using pre-created pipes.
Differential Revision: https://reviews.freebsd.org/D35708
73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
#!/usr/local/bin/python3
|
|
import json
|
|
import os
|
|
import socket
|
|
import time
|
|
from ctypes import cdll
|
|
from ctypes import get_errno
|
|
from ctypes.util import find_library
|
|
from typing import List
|
|
from typing import Optional
|
|
|
|
|
|
class ToolsHelper(object):
|
|
NETSTAT_PATH = "/usr/bin/netstat"
|
|
IFCONFIG_PATH = "/sbin/ifconfig"
|
|
|
|
@classmethod
|
|
def get_output(cls, cmd: str, verbose=False) -> str:
|
|
if verbose:
|
|
print("run: '{}'".format(cmd))
|
|
return os.popen(cmd).read()
|
|
|
|
@classmethod
|
|
def print_output(cls, cmd: str, verbose=True):
|
|
if verbose:
|
|
print("======= {} =====".format(cmd))
|
|
print(cls.get_output(cmd))
|
|
if verbose:
|
|
print()
|
|
|
|
@classmethod
|
|
def print_net_debug(cls):
|
|
cls.print_output("ifconfig")
|
|
cls.print_output("netstat -rnW")
|
|
|
|
@classmethod
|
|
def set_sysctl(cls, oid, val):
|
|
cls.get_output("sysctl {}={}".format(oid, val))
|
|
|
|
@classmethod
|
|
def get_routes(cls, family: str, fibnum: int = 0):
|
|
family_key = {"inet": "-4", "inet6": "-6"}.get(family)
|
|
out = cls.get_output(
|
|
"{} {} -rn -F {} --libxo json".format(cls.NETSTAT_PATH, family_key, fibnum)
|
|
)
|
|
js = json.loads(out)
|
|
js = js["statistics"]["route-information"]["route-table"]["rt-family"]
|
|
if js:
|
|
return js[0]["rt-entry"]
|
|
else:
|
|
return []
|
|
|
|
@classmethod
|
|
def get_linklocals(cls):
|
|
ret = {}
|
|
ifname = None
|
|
ips = []
|
|
for line in cls.get_output(cls.IFCONFIG_PATH).splitlines():
|
|
if line[0].isalnum():
|
|
if ifname:
|
|
ret[ifname] = ips
|
|
ips = []
|
|
ifname = line.split(":")[0]
|
|
else:
|
|
words = line.split()
|
|
if words[0] == "inet6" and words[1].startswith("fe80"):
|
|
# inet6 fe80::1%lo0 prefixlen 64 scopeid 0x2
|
|
ip = words[1].split("%")[0]
|
|
scopeid = int(words[words.index("scopeid") + 1], 16)
|
|
ips.append((ip, scopeid))
|
|
if ifname:
|
|
ret[ifname] = ips
|
|
return ret
|