1import pytest 2from atf_python.utils import BaseTest 3from atf_python.sys.net.tools import ToolsHelper 4from atf_python.sys.net.vnet import SingleVnetTestTemplate 5from atf_python.sys.net.vnet import VnetTestTemplate 6from atf_python.sys.net.vnet import VnetInstance 7 8import errno 9import socket 10import subprocess 11import json 12 13from typing import List 14 15 16# Test classes should be inherited 17# from the BaseTest 18 19 20class TestExampleSimplest(BaseTest): 21 @pytest.mark.skip(reason="comment me to run the test") 22 def test_one(self): 23 assert ToolsHelper.get_output("uname -s").strip() == "FreeBSD" 24 25 26class 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"] 30 31 @pytest.mark.skip(reason="comment me to run the test") 32 def test_one(self): 33 """Optional test description 34 This and the following lines are not propagated 35 to the ATF test description. 36 """ 37 pass 38 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) 49 def test_two(self): 50 pass 51 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""" 55 print() 56 print("SRC_DIR={}".format(request.fspath.dirname)) 57 print("ATF VARS:") 58 for k, v in self.atf_vars.items(): 59 print(" {}: {}".format(k, v)) 60 print() 61 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: 67 s.bind(("::1", 42)) 68 assert exc_info.value.errno == errno.EACCES 69 70 @pytest.mark.skip(reason="comment me to run the test") 71 @pytest.mark.parametrize( 72 "family_tuple", 73 [ 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"), 77 ], 78 ) 79 def test_parametrize(self, family_tuple): 80 family, error = family_tuple 81 try: 82 s = socket.socket(family, socket.SOCK_STREAM) 83 s.close() 84 except OSError as e: 85 if error is None or error != e.errno: 86 raise 87 88 # @pytest.mark.skip(reason="comment me to run the test") 89 def test_with_cleanup(self): 90 print("TEST BODY") 91 92 def cleanup_test_with_cleanup(self, test_id): 93 print("CLEANUP HANDLER") 94 95 96class TestVnetSimple(SingleVnetTestTemplate): 97 """ 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. 101 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"] 106 107 All prefixes from IPV4_PREFIXES and IPV6_PREFIXES are 108 assigned to the single epair interface inside the jail. 109 110 One can rely on the fact that there are no IPv6 prefixes 111 in the tentative state when the test method is called. 112 """ 113 114 IPV6_PREFIXES: List[str] = ["2001:db8::1/64"] 115 IPV4_PREFIXES: List[str] = ["192.0.2.1/24"] 116 117 def setup_method(self, method): 118 """ 119 Optional pre-setup for all of the tests inside the class 120 """ 121 # Code to run before vnet setup 122 # 123 super().setup_method(method) 124 # 125 # Code to run after vnet setup 126 # Executed inside the vnet 127 128 @pytest.mark.skip(reason="comment me to run the test") 129 @pytest.mark.require_user("root") 130 def test_ping(self): 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 133 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)) 139 140 141class TestVnetDual1(VnetTestTemplate): 142 """ 143 VnetTestTemplate creates topology described in the self.TOPOLOGY 144 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. 149 150 This topology contains 2 VNETs connected with 2 epairs: 151 152 [ VNET1 ] [ VNET2 ] 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) 155 lo0 lo0 156 157 """ 158 159 TOPOLOGY = { 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")]}, 164 } 165 166 def _get_iface_stat(self, os_ifname: str): 167 out = ToolsHelper.get_output( 168 "{} -I {} --libxo json".format(ToolsHelper.NETSTAT_PATH, os_ifname) 169 ) 170 js = json.loads(out) 171 return js["statistics"]["interface"][0] 172 173 def vnet2_handler(self, vnet: VnetInstance): 174 """ 175 Test handler that runs in the vnet2 as a separate process. 176 177 This handler receives an interface name, fetches received/sent packets 178 and returns this data back to the parent process. 179 """ 180 while True: 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)) 186 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 193 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 201 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 209