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