Merge pull request #402 from xcp-ng/gln/xapi-network-typehints

Author: glehmann

tests/network/conftest.py

@@ -1,7 +1,9 @@
 import pytest
 
+from lib.host import Host
+
 @pytest.fixture(scope='package')
-def host_no_sdn_controller(host):
+def host_no_sdn_controller(host: Host) -> None:
     """ An XCP-ng with no SDN controller. """
     if host.xe('sdn-controller-list', minimal=True):
         pytest.skip("This test requires an XCP-ng with no SDN controller")

tests/network/test_management_disable_address_type.py

@@ -1,7 +1,9 @@
+from lib.host import Host
+
 # Requirements:
 # - one XCP-ng host (--hosts) (>= 8.3 for IPv6 test)
 
-def test_management_disable_address_type(host):
+def test_management_disable_address_type(host: Host) -> None:
     management_pif = host.management_pif()
     type = management_pif.param_get("primary-address-type").lower()
 

tests/network/test_vif_allowed_ip.py

@@ -9,13 +9,13 @@
 # - one XCP-ng host (--hosts) >= 8.2 (>= 8.3 for the CIDR tests) with no SDN controller configured
 # - a VM (--vm)
 
-def ip_responsive(ip):
+def ip_responsive(ip: str) -> bool:
     return not os.system(f"ping -c 3 -W 10 {ip} > /dev/null 2>&1")
 
 @pytest.mark.small_vm
 @pytest.mark.usefixtures("host_no_sdn_controller")
 class TestAllowedIP:
-    def test_unallowed_ip(self, running_vm: VM):
+    def test_unallowed_ip(self, running_vm: VM) -> None:
         vm = running_vm
         vif = vm.vifs()[0]
         ip = vm.ip
@@ -32,7 +32,7 @@ def test_unallowed_ip(self, running_vm: VM):
         vif.param_clear(f"ipv{ip_family}-allowed")
         vif.param_set("locking-mode", "unlocked")
 
-    def test_allowed_ip(self, running_vm):
+    def test_allowed_ip(self, running_vm: VM) -> None:
         vm = running_vm
         vif = vm.vifs()[0]
         ip = vm.ip
@@ -55,7 +55,7 @@ def test_allowed_ip(self, running_vm):
 @pytest.mark.small_vm
 @pytest.mark.usefixtures("host_at_least_8_3", "host_no_sdn_controller")
 class TestAllowedCIDR:
-    def test_unallowed_cidr(self, running_vm: VM):
+    def test_unallowed_cidr(self, running_vm: VM) -> None:
         vm = running_vm
         vif = vm.vifs()[0]
         ip = vm.ip
@@ -73,7 +73,7 @@ def test_unallowed_cidr(self, running_vm: VM):
         vif.param_clear(f"ipv{ip_family}-allowed")
         vif.param_set("locking-mode", "unlocked")
 
-    def test_allowed_cidr(self, running_vm: VM):
+    def test_allowed_cidr(self, running_vm: VM) -> None:
         vm = running_vm
         vif = vm.vifs()[0]
         ip = vm.ip

tests/network/test_vif_management.py

@@ -20,7 +20,7 @@ def count_interfaces(vm: VM) -> int:
 
 @pytest.mark.small_vm
 class TestVIFManagement:
-    def test_vif_management(self, running_unix_vm: VM):
+    def test_vif_management(self, running_unix_vm: VM) -> None:
         vm = running_unix_vm
         host = vm.host
         network_uuid = host.management_network()

tests/xapi/tls_verification/test_tls_verification.py

@@ -6,6 +6,8 @@
 from lib.common import strtobool
 from lib.host import Host
 
+from typing import Generator
+
 # Requirements:
 # From --hosts parameter:
 # - A XCP-ng >= 8.3 pool with at least two hosts
@@ -21,18 +23,19 @@
 XAPI_POOL_PEM_FILEPATH = f'/etc/xensource/{XAPI_POOL_PEM_FILENAME}'
 
 @pytest.fixture(scope="module")
-def host_with_tls_verification_enabled(hostA1):
+def host_with_tls_verification_enabled(hostA1: Host) -> Generator[Host, None, None]:
     for h in hostA1.pool.hosts:
         logging.info(f"Check that TLS verification is enabled on host {h}")
         assert strtobool(h.param_get("tls-verification-enabled")), f"TLS verification must be enabled on host {h}"
         logging.info(f"Check that the host certificate exists on host {h}")
         cert_uuid = hostA1.xe('certificate-list', {'host': h.uuid, 'type': 'host_internal'}, minimal=True)
         assert len(cert_uuid) > 0, f"A host_internal certificate must exist on host {h}"
+    yield hostA1
 
 
 @pytest.mark.usefixtures("host_at_least_8_3", "host_with_tls_verification_enabled")
 class TestTLSVerification:
-    def _test_tls_verification(self, hostA1: Host, with_toolstack_restart=False):
+    def _test_tls_verification(self, hostA1: Host, with_toolstack_restart: bool = False) -> None:
         for h in hostA1.pool.hosts[1:]:
             logging.info(f"Establish a connexion from host {hostA1} to host {h} by running 'xe host-dmesg'")
             hostA1.xe('host-dmesg', {'host': h.uuid})
@@ -44,10 +47,10 @@ def _test_tls_verification(self, hostA1: Host, with_toolstack_restart=False):
             logging.info(f"Test connexion from host {h} to host {hostA1} by running 'xe host-dmesg'")
             h.xe('host-dmesg', {'host': hostA1.uuid})
 
-    def test_tls_verification(self, hostA1):
+    def test_tls_verification(self, hostA1: Host) -> None:
         self._test_tls_verification(hostA1)
 
-    def test_refresh_certificate(self, hostA1):
+    def test_refresh_certificate(self, hostA1: Host) -> None:
         logging.info("Refresh the xapi:pool certificate on every pool member")
         for h in hostA1.pool.hosts:
             old_checksum = h.ssh(f'md5sum {XAPI_POOL_PEM_FILEPATH}').split()[0]
@@ -58,7 +61,7 @@ def test_refresh_certificate(self, hostA1):
         self._test_tls_verification(hostA1, with_toolstack_restart=True)
 
     @pytest.fixture(scope="function")
-    def hostA2_with_saved_cert(self, hostA2: Host):
+    def hostA2_with_saved_cert(self, hostA2: Host) -> Generator[Host, None, None]:
         tmp_dir = hostA2.ssh('mktemp -d')
         logging.info(f"Save {XAPI_POOL_PEM_FILEPATH} on {hostA2}")
         hostA2.ssh(f'cp {XAPI_POOL_PEM_FILEPATH} {tmp_dir}')
@@ -68,7 +71,7 @@ def hostA2_with_saved_cert(self, hostA2: Host):
         hostA2.ssh(f'rm -r {tmp_dir}')
         hostA2.ssh('systemctl reload-or-restart stunnel@xapi')
 
-    def test_break_cert(self, hostA1: Host, hostA2_with_saved_cert: Host):
+    def test_break_cert(self, hostA1: Host, hostA2_with_saved_cert: Host) -> None:
         hostA2 = hostA2_with_saved_cert
         logging.info(f"Replace the certificate on host {hostA2}")
         hostA2.ssh(f'rm {XAPI_POOL_PEM_FILEPATH}')
@@ -88,7 +91,7 @@ def test_break_cert(self, hostA1: Host, hostA2_with_saved_cert: Host):
 The server may be switched off or there may be network connectivity problems." in excinfo.value.stdout
         )
 
-    def test_toolstack_restart(self, hostA1: Host, hostA2: Host):
+    def test_toolstack_restart(self, hostA1: Host, hostA2: Host) -> None:
         """
         Same test as the previous one, but we don't break the cert.
 

tests/xapi_plugins/plugin_hyperthreading/test_hyperthreading.py

@@ -1,6 +1,8 @@
+from lib.host import Host
+
 # Requirements:
 # From --hosts parameter:
 # - host(A1): first XCP-ng host > 8.2.
 
-def test_get_hyperthreading(host):
+def test_get_hyperthreading(host: Host) -> None:
     host.call_plugin('hyperthreading.py', 'get_hyperthreading')

tests/xapi_plugins/plugin_lsblk/test_lsblk.py

@@ -1,6 +1,8 @@
+from lib.host import Host
+
 # Requirements:
 # From --hosts parameter:
 # - host(A1): first XCP-ng host > 8.2.
 
-def test_list_block_devices(host):
+def test_list_block_devices(host: Host) -> None:
     host.call_plugin('lsblk.py', 'list_block_devices')

tests/xapi_plugins/plugin_netdata/test_netdata.py

@@ -1,6 +1,9 @@
 import pytest
 
 from lib.common import strtobool
+from lib.host import Host
+
+from typing import Generator
 
 # Requirements:
 # From --hosts parameter:
@@ -9,15 +12,15 @@
 # - access to XCP-ng RPM repository from hostA1
 
 @pytest.fixture(scope='module')
-def host_without_netdata(host):
+def host_without_netdata(host: Host) -> Generator[Host, None, None]:
     assert not strtobool(host.call_plugin('netdata.py', 'is_netdata_installed'))
     yield host
 
 class TestInstall:
-    def test_is_netdata_installed(self, host):
+    def test_is_netdata_installed(self, host: Host) -> None:
         host.call_plugin('netdata.py', 'is_netdata_installed')
 
-    def test_install_netdata(self, host_without_netdata):
+    def test_install_netdata(self, host_without_netdata: Host) -> None:
         host = host_without_netdata
         host.yum_save_state()
         host.call_plugin('netdata.py', 'install_netdata', {
@@ -29,7 +32,7 @@ def test_install_netdata(self, host_without_netdata):
         host.yum_restore_saved_state()
 
 class TestApiKey:
-    def test_get_netdata_api_key(self, host_without_netdata):
+    def test_get_netdata_api_key(self, host_without_netdata: Host) -> None:
         host = host_without_netdata
         host.yum_save_state()
         host.call_plugin('netdata.py', 'install_netdata', {

tests/xapi_plugins/plugin_raid/test_raid.py

@@ -4,12 +4,14 @@
 
 from lib.host import Host
 
+from typing import Generator
+
 # Requirements:
 # From --hosts parameter:
 # - host(A1): first XCP-ng host > 8.2.
 
 @pytest.fixture(scope='module')
-def host_with_raid(host: Host):
+def host_with_raid(host: Host) -> Generator[Host, None, None]:
     dummy_raid = False
     if not host.file_exists('/dev/md127', regular_file=False):
         logging.info("> Host has no raids, creating one for tests")
@@ -29,6 +31,6 @@ def host_with_raid(host: Host):
         host.ssh('losetup -d /dev/loop1')
         host.ssh('rm -rf raid-1 raid-0')
 
-def test_check_raid_pool(host_with_raid):
+def test_check_raid_pool(host_with_raid: Host) -> None:
     host = host_with_raid
     host.call_plugin('raid.py', 'check_raid_pool')

tests/xapi_plugins/plugin_smartctl/test_smartctl.py

@@ -2,21 +2,23 @@
 
 import json
 
+from lib.host import Host
+
 # Requirements:
 # From --hosts parameter:
 # - host(A1): first XCP-ng host >= 8.3.
 
-def _call_plugin(host, fn):
+def _call_plugin(host: Host, fn: str) -> None:
     ret = host.call_plugin("smartctl.py", fn)
     try:
         json.loads(ret)
     except ValueError:
         pytest.fail("JSON string was expected but ValueError was raised")
 
 @pytest.mark.usefixtures("host_at_least_8_3")
-def test_smartctl_information(host):
+def test_smartctl_information(host: Host) -> None:
     _call_plugin(host, "information")
 
 @pytest.mark.usefixtures("host_at_least_8_3")
-def test_smartctl_health(host):
+def test_smartctl_health(host: Host) -> None:
     _call_plugin(host, "health")