Changeset View
Standalone View
tests/examples/test_examples.py
- This file was added.
import pytest | |||||||||||
from atf_python.utils import BaseTest | |||||||||||
from atf_python.sys.net.tools import ToolsHelper | |||||||||||
from atf_python.sys.net.vnet import SingleVnetTestTemplate | |||||||||||
from atf_python.sys.net.vnet import VnetTestTemplate | |||||||||||
from atf_python.sys.net.vnet import VnetInstance | |||||||||||
import errno | |||||||||||
import socket | |||||||||||
import subprocess | |||||||||||
import json | |||||||||||
from typing import List | |||||||||||
# Test classes should be inherited | |||||||||||
# from the BaseTest | |||||||||||
class TestExampleSimplest(BaseTest): | |||||||||||
@pytest.mark.skip(reason="comment me to run the test") | |||||||||||
def test_one(self): | |||||||||||
assert ToolsHelper.get_output("uname -s").strip() == "FreeBSD" | |||||||||||
class TestExampleSimple(BaseTest): | |||||||||||
# List of required kernel modules (kldstat -v) | |||||||||||
# that needs to be present for the tests to run | |||||||||||
REQUIRED_MODULES = ["null"] | |||||||||||
@pytest.mark.skip(reason="comment me to run the test") | |||||||||||
def test_one(self): | |||||||||||
"""Optional test description | |||||||||||
This and the following lines are not propagated | |||||||||||
to the ATF test description. | |||||||||||
""" | |||||||||||
pass | |||||||||||
@pytest.mark.skip(reason="comment me to run the test") | |||||||||||
# List of all requirements supported by an atf runner | |||||||||||
# See atf-test-case(4) for the detailed description | |||||||||||
@pytest.mark.require_user("root") | |||||||||||
jlduran_gmail.comUnsubmitted Done Inline Actions
jlduran_gmail.com: | |||||||||||
@pytest.mark.require_arch(["amd64", "i386"]) | |||||||||||
@pytest.mark.require_files(["/path/file1", "/path/file2"]) | |||||||||||
@pytest.mark.require_machine(["amd64", "i386"]) | |||||||||||
@pytest.mark.require_memory("200M") | |||||||||||
@pytest.mark.require_progs(["prog1", "prog2"]) | |||||||||||
@pytest.mark.timeout(300) | |||||||||||
def test_two(self): | |||||||||||
pass | |||||||||||
@pytest.mark.skip(reason="comment me to run the test") | |||||||||||
@pytest.mark.require_user("unprivileged") | |||||||||||
def test_syscall_failure(self): | |||||||||||
Done Inline Actions
jlduran_gmail.com: | |||||||||||
Not Done Inline ActionsMay I ask, why do you keep this line commented out? jlduran_gmail.com: May I ask, why do you keep this line commented out?
CirrusCI has a bad habit (and grudgingly I)… | |||||||||||
Done Inline ActionsSure, two reasons.
Guess it's time to do to just that - as it turns out I need this one for netlink testing. So, I'd say it's worth keeping as is untils at least one of two issues get resoled. Thoughts? melifaro: Sure, two reasons.
`unprivileged` is a "special" user that kyua translates to the value of… | |||||||||||
Not Done Inline ActionsYes, I see. Thanks for the detailed explanation. I wasn't aware of all those details. jlduran_gmail.com: Yes, I see. Thanks for the detailed explanation. I wasn't aware of all those details.
So, for… | |||||||||||
Done Inline ActionsI looked at the code to refresh that part. If @pytest.mark.require_user("unprivileged") is set,
2.1 pass require.user="root" Does it sound reasonable? melifaro: I looked at the code to refresh that part.
So kyua runner passes the parameters in -vKEY=VALUE… | |||||||||||
Not Done Inline ActionsNo. If the "unprivileged_user" isn't set in /usr/local/etc/kyua.conf, then Kyua should skip the test. So I still think you should uncomment that line. asomers: No. If the "unprivileged_user" isn't set in /usr/local/etc/kyua.conf, then Kyua should skip… | |||||||||||
Done Inline ActionsRe uncommenting - will do. Let me come up with the unprivileged-user fix first. melifaro: Re uncommenting - will do. Let me come up with the `unprivileged-user` fix first. | |||||||||||
Done Inline ActionsUncommented. I've published D37923 to address the dropping-privileges issue. melifaro: Uncommented. I've published D37923 to address the dropping-privileges issue. | |||||||||||
s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) | |||||||||||
with pytest.raises(OSError) as exc_info: | |||||||||||
s.bind(("::1", 42)) | |||||||||||
assert exc_info.value.errno == errno.EACCES | |||||||||||
@pytest.mark.skip(reason="comment me to run the test") | |||||||||||
@pytest.mark.parametrize( | |||||||||||
"family_tuple", | |||||||||||
[ | |||||||||||
pytest.param([socket.AF_INET, None], id="AF_INET"), | |||||||||||
pytest.param([socket.AF_INET6, None], id="AF_INET6"), | |||||||||||
pytest.param([39, errno.EAFNOSUPPORT], id="FAMILY_39"), | |||||||||||
], | |||||||||||
Done Inline Actions
jlduran_gmail.com: | |||||||||||
) | |||||||||||
def test_parametrize(self, family_tuple): | |||||||||||
family, error = family_tuple | |||||||||||
try: | |||||||||||
s = socket.socket(family, socket.SOCK_STREAM) | |||||||||||
s.close() | |||||||||||
except OSError as e: | |||||||||||
if error is None or error != e.errno: | |||||||||||
raise | |||||||||||
# @pytest.mark.skip(reason="comment me to run the test") | |||||||||||
def test_with_cleanup(self): | |||||||||||
print("TEST BODY") | |||||||||||
def cleanup_test_with_cleanup(self, test_id): | |||||||||||
print("CLEANUP HANDLER") | |||||||||||
class TestVnetSimple(SingleVnetTestTemplate): | |||||||||||
""" | |||||||||||
SingleVnetTestTemplate creates a topology with a single | |||||||||||
vnet and a single epair between this vnet and the host system. | |||||||||||
Additionally, lo0 interface is created inside the vnet. | |||||||||||
Both vnets and interfaces are aliased as vnetX and ifY. | |||||||||||
They can be accessed via maps: | |||||||||||
vnet: VnetInstance = self.vnet_map["vnet1"] | |||||||||||
iface: VnetInterface = vnet.iface_alias_map["if1"] | |||||||||||
All prefixes from IPV4_PREFIXES and IPV6_PREFIXES are | |||||||||||
assigned to the single epair interface inside the jail. | |||||||||||
Done Inline Actions
There has been a preference lately for RFC 5737 addresses, to avoid a possible clash with a private IP currently in use. jlduran_gmail.com: There has been a preference lately for RFC 5737 addresses, to avoid a possible clash with a… | |||||||||||
Done Inline ActionsYep, I forgot about the IPv4 docprefixes. Thank you! melifaro: Yep, I forgot about the IPv4 docprefixes. Thank you! | |||||||||||
One can rely on the fact that there are no IPv6 prefixes | |||||||||||
in the tentative state when the test method is called. | |||||||||||
""" | |||||||||||
IPV6_PREFIXES: List[str] = ["2001:db8::1/64"] | |||||||||||
IPV4_PREFIXES: List[str] = ["192.0.2.1/24"] | |||||||||||
def setup_method(self, method): | |||||||||||
""" | |||||||||||
Optional pre-setup for all of the tests inside the class | |||||||||||
""" | |||||||||||
# Code to run before vnet setup | |||||||||||
# | |||||||||||
super().setup_method(method) | |||||||||||
# | |||||||||||
# Code to run after vnet setup | |||||||||||
Done Inline Actions
jlduran_gmail.com: | |||||||||||
# Executed inside the vnet | |||||||||||
@pytest.mark.skip(reason="comment me to run the test") | |||||||||||
@pytest.mark.require_user("root") | |||||||||||
def test_ping(self): | |||||||||||
assert subprocess.run("ping -c1 192.0.2.1".split()).returncode == 0 | |||||||||||
assert subprocess.run("ping -c1 2001:db8::1".split()).returncode == 0 | |||||||||||
@pytest.mark.skip(reason="comment me to run the test") | |||||||||||
def test_topology(self): | |||||||||||
vnet = self.vnet_map["vnet1"] | |||||||||||
iface = vnet.iface_alias_map["if1"] | |||||||||||
print("Iface {} inside vnet {}".format(iface.name, vnet.name)) | |||||||||||
class TestVnetDual1(VnetTestTemplate): | |||||||||||
""" | |||||||||||
VnetTestTemplate creates topology described in the self.TOPOLOGY | |||||||||||
Each vnet (except vnet1) can have a handler function, named | |||||||||||
vnetX_handler. This function will be run in a separate process | |||||||||||
inside vnetX jail. The framework automatically creates a pipe | |||||||||||
to allow communication between the main test and the vnet handler. | |||||||||||
This topology contains 2 VNETs connected with 2 epairs: | |||||||||||
[ VNET1 ] [ VNET2 ] | |||||||||||
if1(epair) 2001:db8:a::1/64 <-> 2001:db8:a::2/64 if1(epair) | |||||||||||
if2(epair) 2001:db8:b::1/64 <-> 2001:db8:b::2/64 if2(epair) | |||||||||||
lo0 lo0 | |||||||||||
""" | |||||||||||
TOPOLOGY = { | |||||||||||
"vnet1": {"ifaces": ["if1", "if2"]}, | |||||||||||
"vnet2": {"ifaces": ["if1", "if2"]}, | |||||||||||
"if1": {"prefixes6": [("2001:db8:a::1/64", "2001:db8:a::2/64")]}, | |||||||||||
"if2": {"prefixes6": [("2001:db8:b::1/64", "2001:db8:b::2/64")]}, | |||||||||||
} | |||||||||||
def _get_iface_stat(self, os_ifname: str): | |||||||||||
out = ToolsHelper.get_output( | |||||||||||
"{} -I {} --libxo json".format(ToolsHelper.NETSTAT_PATH, os_ifname) | |||||||||||
) | |||||||||||
js = json.loads(out) | |||||||||||
return js["statistics"]["interface"][0] | |||||||||||
def vnet2_handler(self, vnet: VnetInstance): | |||||||||||
""" | |||||||||||
Test handler that runs in the vnet2 as a separate process. | |||||||||||
This handler receives an interface name, fetches received/sent packets | |||||||||||
and returns this data back to the parent process. | |||||||||||
""" | |||||||||||
while True: | |||||||||||
# receives 'ifX' with an infinite timeout | |||||||||||
iface_alias = self.wait_object(vnet.pipe, None) | |||||||||||
# Translates topology interface name to the actual OS-assigned name | |||||||||||
os_ifname = vnet.iface_alias_map[iface_alias].name | |||||||||||
self.send_object(vnet.pipe, self._get_iface_stat(os_ifname)) | |||||||||||
@pytest.mark.skip(reason="comment me to run the test") | |||||||||||
@pytest.mark.require_user("root") | |||||||||||
Done Inline Actions
jlduran_gmail.com: | |||||||||||
def test_ifstat(self): | |||||||||||
"""Checks that RX interface packets are properly accounted for""" | |||||||||||
second_vnet = self.vnet_map["vnet2"] | |||||||||||
pipe = second_vnet.pipe | |||||||||||
# Ping neighbor IP on if1 and verify that the counter was incremented | |||||||||||
self.send_object(pipe, "if1") | |||||||||||
old_stat = self.wait_object(pipe) | |||||||||||
Done Inline Actions
jlduran_gmail.com: | |||||||||||
assert subprocess.run("ping -c5 2001:db8:a::2".split()).returncode == 0 | |||||||||||
self.send_object(pipe, "if1") | |||||||||||
new_stat = self.wait_object(pipe) | |||||||||||
assert new_stat["received-packets"] - old_stat["received-packets"] >= 5 | |||||||||||
# Ping neighbor IP on if2 and verify that the counter was incremented | |||||||||||
self.send_object(pipe, "if2") | |||||||||||
old_stat = self.wait_object(pipe) | |||||||||||
assert subprocess.run("ping -c5 2001:db8:b::2".split()).returncode == 0 | |||||||||||
self.send_object(pipe, "if2") | |||||||||||
new_stat = self.wait_object(pipe) | |||||||||||
assert new_stat["received-packets"] - old_stat["received-packets"] >= 5 |