tests/examples/test_examples.py

   1 import pytest
   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
   7
   8 import errno
   9 import socket
  10 import subprocess
  11 import json
  12
  13 from typing import List
  14
  15
  16 # Test classes should be inherited
  17 # from the BaseTest
  18
  19
  20 class 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
  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"]
  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
  96 class 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
 141 class 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