2 from atf_python.utils import BaseTest
3 from atf_python.sys.net.tools import ToolsHelper
4 from atf_python.sys.net.vnet import SingleVnetTestTemplate
5 from atf_python.sys.net.vnet import VnetTestTemplate
6 from atf_python.sys.net.vnet import VnetInstance
13 from typing import List
16 # Test classes should be inherited
20 class TestExampleSimplest(BaseTest):
21 @pytest.mark.skip(reason="comment me to run the test")
23 assert ToolsHelper.get_output("uname -s").strip() == "FreeBSD"
26 class TestExampleSimple(BaseTest):
27 # List of required kernel modules (kldstat -v)
28 # that needs to be present for the tests to run
29 REQUIRED_MODULES = ["null"]
31 @pytest.mark.skip(reason="comment me to run the test")
33 """Optional test description
34 This and the following lines are not propagated
35 to the ATF test description.
39 @pytest.mark.skip(reason="comment me to run the test")
40 # List of all requirements supported by an atf runner
41 # See atf-test-case(4) for the detailed description
42 @pytest.mark.require_user("root")
43 @pytest.mark.require_arch(["amd64", "i386"])
44 @pytest.mark.require_files(["/path/file1", "/path/file2"])
45 @pytest.mark.require_machine(["amd64", "i386"])
46 @pytest.mark.require_memory("200M")
47 @pytest.mark.require_progs(["prog1", "prog2"])
48 @pytest.mark.timeout(300)
52 @pytest.mark.skip(reason="comment me to run the test")
53 def test_get_properties(self, request):
54 """Shows fetching of test src dir and ATF-set variables"""
56 print("SRC_DIR={}".format(request.fspath.dirname))
58 for k, v in self.atf_vars.items():
59 print(" {}: {}".format(k, v))
62 @pytest.mark.skip(reason="comment me to run the test")
63 @pytest.mark.require_user("unprivileged")
64 def test_syscall_failure(self):
65 s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
66 with pytest.raises(OSError) as exc_info:
68 assert exc_info.value.errno == errno.EACCES
70 @pytest.mark.skip(reason="comment me to run the test")
71 @pytest.mark.parametrize(
74 pytest.param([socket.AF_INET, None], id="AF_INET"),
75 pytest.param([socket.AF_INET6, None], id="AF_INET6"),
76 pytest.param([39, errno.EAFNOSUPPORT], id="FAMILY_39"),
79 def test_parametrize(self, family_tuple):
80 family, error = family_tuple
82 s = socket.socket(family, socket.SOCK_STREAM)
85 if error is None or error != e.errno:
88 # @pytest.mark.skip(reason="comment me to run the test")
89 def test_with_cleanup(self):
92 def cleanup_test_with_cleanup(self, test_id):
93 print("CLEANUP HANDLER")
96 class TestVnetSimple(SingleVnetTestTemplate):
98 SingleVnetTestTemplate creates a topology with a single
99 vnet and a single epair between this vnet and the host system.
100 Additionally, lo0 interface is created inside the vnet.
102 Both vnets and interfaces are aliased as vnetX and ifY.
103 They can be accessed via maps:
104 vnet: VnetInstance = self.vnet_map["vnet1"]
105 iface: VnetInterface = vnet.iface_alias_map["if1"]
107 All prefixes from IPV4_PREFIXES and IPV6_PREFIXES are
108 assigned to the single epair interface inside the jail.
110 One can rely on the fact that there are no IPv6 prefixes
111 in the tentative state when the test method is called.
114 IPV6_PREFIXES: List[str] = ["2001:db8::1/64"]
115 IPV4_PREFIXES: List[str] = ["192.0.2.1/24"]
117 def setup_method(self, method):
119 Optional pre-setup for all of the tests inside the class
121 # Code to run before vnet setup
123 super().setup_method(method)
125 # Code to run after vnet setup
126 # Executed inside the vnet
128 @pytest.mark.skip(reason="comment me to run the test")
129 @pytest.mark.require_user("root")
131 assert subprocess.run("ping -c1 192.0.2.1".split()).returncode == 0
132 assert subprocess.run("ping -c1 2001:db8::1".split()).returncode == 0
134 @pytest.mark.skip(reason="comment me to run the test")
135 def test_topology(self):
136 vnet = self.vnet_map["vnet1"]
137 iface = vnet.iface_alias_map["if1"]
138 print("Iface {} inside vnet {}".format(iface.name, vnet.name))
141 class TestVnetDual1(VnetTestTemplate):
143 VnetTestTemplate creates topology described in the self.TOPOLOGY
145 Each vnet (except vnet1) can have a handler function, named
146 vnetX_handler. This function will be run in a separate process
147 inside vnetX jail. The framework automatically creates a pipe
148 to allow communication between the main test and the vnet handler.
150 This topology contains 2 VNETs connected with 2 epairs:
153 if1(epair) 2001:db8:a::1/64 <-> 2001:db8:a::2/64 if1(epair)
154 if2(epair) 2001:db8:b::1/64 <-> 2001:db8:b::2/64 if2(epair)
160 "vnet1": {"ifaces": ["if1", "if2"]},
161 "vnet2": {"ifaces": ["if1", "if2"]},
162 "if1": {"prefixes6": [("2001:db8:a::1/64", "2001:db8:a::2/64")]},
163 "if2": {"prefixes6": [("2001:db8:b::1/64", "2001:db8:b::2/64")]},
166 def _get_iface_stat(self, os_ifname: str):
167 out = ToolsHelper.get_output(
168 "{} -I {} --libxo json".format(ToolsHelper.NETSTAT_PATH, os_ifname)
171 return js["statistics"]["interface"][0]
173 def vnet2_handler(self, vnet: VnetInstance):
175 Test handler that runs in the vnet2 as a separate process.
177 This handler receives an interface name, fetches received/sent packets
178 and returns this data back to the parent process.
181 # receives 'ifX' with an infinite timeout
182 iface_alias = self.wait_object(vnet.pipe, None)
183 # Translates topology interface name to the actual OS-assigned name
184 os_ifname = vnet.iface_alias_map[iface_alias].name
185 self.send_object(vnet.pipe, self._get_iface_stat(os_ifname))
187 @pytest.mark.skip(reason="comment me to run the test")
188 @pytest.mark.require_user("root")
189 def test_ifstat(self):
190 """Checks that RX interface packets are properly accounted for"""
191 second_vnet = self.vnet_map["vnet2"]
192 pipe = second_vnet.pipe
194 # Ping neighbor IP on if1 and verify that the counter was incremented
195 self.send_object(pipe, "if1")
196 old_stat = self.wait_object(pipe)
197 assert subprocess.run("ping -c5 2001:db8:a::2".split()).returncode == 0
198 self.send_object(pipe, "if1")
199 new_stat = self.wait_object(pipe)
200 assert new_stat["received-packets"] - old_stat["received-packets"] >= 5
202 # Ping neighbor IP on if2 and verify that the counter was incremented
203 self.send_object(pipe, "if2")
204 old_stat = self.wait_object(pipe)
205 assert subprocess.run("ping -c5 2001:db8:b::2".split()).returncode == 0
206 self.send_object(pipe, "if2")
207 new_stat = self.wait_object(pipe)
208 assert new_stat["received-packets"] - old_stat["received-packets"] >= 5