git: 176e0427b208 - stable/13 - testing: add python test examples
- Go to: [ bottom of page ] [ top of archives ] [ this month ]
Date: Mon, 23 Jan 2023 22:12:16 UTC
The branch stable/13 has been updated by melifaro:
URL: https://cgit.FreeBSD.org/src/commit/?id=176e0427b208a326ef8256c53c79edeea43e13dc
commit 176e0427b208a326ef8256c53c79edeea43e13dc
Author: Alexander V. Chernikov <melifaro@FreeBSD.org>
AuthorDate: 2023-01-01 14:35:41 +0000
Commit: Alexander V. Chernikov <melifaro@FreeBSD.org>
CommitDate: 2023-01-23 22:09:04 +0000
testing: add python test examples
Simplify the adoption of python tests by proving some examples,
utilising commonly-used patterns.
Differential Revision: https://reviews.freebsd.org/D37902
Reviewed by: asomers
MFC after: 2 weeks
(cherry picked from commit 8161b823d77f9d89ffabd47444a83d693f74c515)
---
etc/mtree/BSD.tests.dist | 2 +
tests/Makefile | 1 +
tests/examples/Makefile | 10 ++
tests/examples/test_examples.py | 198 ++++++++++++++++++++++++++++++++++++++++
4 files changed, 211 insertions(+)
diff --git a/etc/mtree/BSD.tests.dist b/etc/mtree/BSD.tests.dist
index 0dabe96110d8..136877db00d6 100644
--- a/etc/mtree/BSD.tests.dist
+++ b/etc/mtree/BSD.tests.dist
@@ -263,6 +263,8 @@
rc.d
..
..
+ examples
+ ..
games
..
gnu
diff --git a/tests/Makefile b/tests/Makefile
index b406b8dc6c17..47fc9488f772 100644
--- a/tests/Makefile
+++ b/tests/Makefile
@@ -9,6 +9,7 @@ ${PACKAGE}FILES+= README __init__.py conftest.py
KYUAFILE= yes
SUBDIR+= etc
+SUBDIR+= examples
SUBDIR+= sys
SUBDIR+= atf_python
diff --git a/tests/examples/Makefile b/tests/examples/Makefile
new file mode 100644
index 000000000000..7a5d84a98dfe
--- /dev/null
+++ b/tests/examples/Makefile
@@ -0,0 +1,10 @@
+# $FreeBSD$
+
+PACKAGE= tests
+
+TESTSDIR= ${TESTSBASE}/examples
+
+ATF_TESTS_PYTEST += test_examples.py
+
+.include <bsd.test.mk>
+
diff --git a/tests/examples/test_examples.py b/tests/examples/test_examples.py
new file mode 100644
index 000000000000..13fdcc420f0e
--- /dev/null
+++ b/tests/examples/test_examples.py
@@ -0,0 +1,198 @@
+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")
+ @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):
+ 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"),
+ ],
+ )
+ 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.
+
+ 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
+ # 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")
+ 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)
+ 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