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 @pytest.mark.require_user("unprivileged") 54 def test_syscall_failure(self): 55 s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) 56 with pytest.raises(OSError) as exc_info: 57 s.bind(("::1", 42)) 58 assert exc_info.value.errno == errno.EACCES 59 60 @pytest.mark.skip(reason="comment me to run the test") 61 @pytest.mark.parametrize( 62 "family_tuple", 63 [ 64 pytest.param([socket.AF_INET, None], id="AF_INET"), 65 pytest.param([socket.AF_INET6, None], id="AF_INET6"), 66 pytest.param([39, errno.EAFNOSUPPORT], id="FAMILY_39"), 67 ], 68 ) 69 def test_parametrize(self, family_tuple): 70 family, error = family_tuple 71 try: 72 s = socket.socket(family, socket.SOCK_STREAM) 73 s.close() 74 except OSError as e: 75 if error is None or error != e.errno: 76 raise 77 78 # @pytest.mark.skip(reason="comment me to run the test") 79 def test_with_cleanup(self): 80 print("TEST BODY") 81 82 def cleanup_test_with_cleanup(self, test_id): 83 print("CLEANUP HANDLER") 84 85 86class TestVnetSimple(SingleVnetTestTemplate): 87 """ 88 SingleVnetTestTemplate creates a topology with a single 89 vnet and a single epair between this vnet and the host system. 90 Additionally, lo0 interface is created inside the vnet. 91 92 Both vnets and interfaces are aliased as vnetX and ifY. 93 They can be accessed via maps: 94 vnet: VnetInstance = self.vnet_map["vnet1"] 95 iface: VnetInterface = vnet.iface_alias_map["if1"] 96 97 All prefixes from IPV4_PREFIXES and IPV6_PREFIXES are 98 assigned to the single epair interface inside the jail. 99 100 One can rely on the fact that there are no IPv6 prefixes 101 in the tentative state when the test method is called. 102 """ 103 104 IPV6_PREFIXES: List[str] = ["2001:db8::1/64"] 105 IPV4_PREFIXES: List[str] = ["192.0.2.1/24"] 106 107 def setup_method(self, method): 108 """ 109 Optional pre-setup for all of the tests inside the class 110 """ 111 # Code to run before vnet setup 112 # 113 super().setup_method(method) 114 # 115 # Code to run after vnet setup 116 # Executed inside the vnet 117 118 @pytest.mark.skip(reason="comment me to run the test") 119 @pytest.mark.require_user("root") 120 def test_ping(self): 121 assert subprocess.run("ping -c1 192.0.2.1".split()).returncode == 0 122 assert subprocess.run("ping -c1 2001:db8::1".split()).returncode == 0 123 124 @pytest.mark.skip(reason="comment me to run the test") 125 def test_topology(self): 126 vnet = self.vnet_map["vnet1"] 127 iface = vnet.iface_alias_map["if1"] 128 print("Iface {} inside vnet {}".format(iface.name, vnet.name)) 129 130 131class TestVnetDual1(VnetTestTemplate): 132 """ 133 VnetTestTemplate creates topology described in the self.TOPOLOGY 134 135 Each vnet (except vnet1) can have a handler function, named 136 vnetX_handler. This function will be run in a separate process 137 inside vnetX jail. The framework automatically creates a pipe 138 to allow communication between the main test and the vnet handler. 139 140 This topology contains 2 VNETs connected with 2 epairs: 141 142 [ VNET1 ] [ VNET2 ] 143 if1(epair) 2001:db8:a::1/64 <-> 2001:db8:a::2/64 if1(epair) 144 if2(epair) 2001:db8:b::1/64 <-> 2001:db8:b::2/64 if2(epair) 145 lo0 lo0 146 147 """ 148 149 TOPOLOGY = { 150 "vnet1": {"ifaces": ["if1", "if2"]}, 151 "vnet2": {"ifaces": ["if1", "if2"]}, 152 "if1": {"prefixes6": [("2001:db8:a::1/64", "2001:db8:a::2/64")]}, 153 "if2": {"prefixes6": [("2001:db8:b::1/64", "2001:db8:b::2/64")]}, 154 } 155 156 def _get_iface_stat(self, os_ifname: str): 157 out = ToolsHelper.get_output( 158 "{} -I {} --libxo json".format(ToolsHelper.NETSTAT_PATH, os_ifname) 159 ) 160 js = json.loads(out) 161 return js["statistics"]["interface"][0] 162 163 def vnet2_handler(self, vnet: VnetInstance): 164 """ 165 Test handler that runs in the vnet2 as a separate process. 166 167 This handler receives an interface name, fetches received/sent packets 168 and returns this data back to the parent process. 169 """ 170 while True: 171 # receives 'ifX' with an infinite timeout 172 iface_alias = self.wait_object(vnet.pipe, None) 173 # Translates topology interface name to the actual OS-assigned name 174 os_ifname = vnet.iface_alias_map[iface_alias].name 175 self.send_object(vnet.pipe, self._get_iface_stat(os_ifname)) 176 177 @pytest.mark.skip(reason="comment me to run the test") 178 @pytest.mark.require_user("root") 179 def test_ifstat(self): 180 """Checks that RX interface packets are properly accounted for""" 181 second_vnet = self.vnet_map["vnet2"] 182 pipe = second_vnet.pipe 183 184 # Ping neighbor IP on if1 and verify that the counter was incremented 185 self.send_object(pipe, "if1") 186 old_stat = self.wait_object(pipe) 187 assert subprocess.run("ping -c5 2001:db8:a::2".split()).returncode == 0 188 self.send_object(pipe, "if1") 189 new_stat = self.wait_object(pipe) 190 assert new_stat["received-packets"] - old_stat["received-packets"] >= 5 191 192 # Ping neighbor IP on if2 and verify that the counter was incremented 193 self.send_object(pipe, "if2") 194 old_stat = self.wait_object(pipe) 195 assert subprocess.run("ping -c5 2001:db8:b::2".split()).returncode == 0 196 self.send_object(pipe, "if2") 197 new_stat = self.wait_object(pipe) 198 assert new_stat["received-packets"] - old_stat["received-packets"] >= 5 199