Run the code checkers on the whole repository

glehmann · glehmann · commit 97c6161bc6b4 · 2026-02-17T22:38:30.000+01:00
Signed-off-by: Gaëtan Lehmann &lt;gaetan.lehmann@vates.tech&gt;
diff --git a/.github/workflows/code-checkers.yml b/.github/workflows/code-checkers.yml
@@ -14,7 +14,9 @@ jobs:
       - uses: ./.github/actions/uv-setup
       - name: Create a dummy data.py
         run: cp data.py-dist data.py
-      - run: mypy --install-types --non-interactive lib/ conftest.py pkgfixtures.py tests/
+      - name: Create a dummy vm_data.py
+        run: cp vm_data.py-dist vm_data.py
+      - run: mypy --install-types --non-interactive .
 
   pyright:
     runs-on: ubuntu-latest
@@ -25,7 +27,9 @@ jobs:
       - uses: ./.github/actions/uv-setup/
       - name: Create a dummy data.py
         run: cp data.py-dist data.py
-      - run: pyright lib/ conftest.py pkgfixtures.py # tests/
+      - name: Create a dummy vm_data.py
+        run: cp vm_data.py-dist vm_data.py
+      - run: pyright .
 
   ruff:
     runs-on: ubuntu-latest
@@ -38,7 +42,7 @@ jobs:
       - uses: ./.github/actions/uv-setup/
       - name: Create a dummy data.py
         run: cp data.py-dist data.py
-      - run: ruff check lib/ tests/
+      - run: ruff check
 
   flake8:
     runs-on: ubuntu-latest
diff --git a/conftest.py b/conftest.py
@@ -13,9 +13,9 @@
 import lib.config as global_config
 from lib import pxe
 from lib.common import (
-    callable_marker,
     DiskDevName,
     HostAddress,
+    callable_marker,
     is_uuid,
     prefix_object_name,
     setup_formatted_and_mounted_disk,
@@ -24,8 +24,8 @@
     vm_image,
     wait_for,
 )
-from lib.netutil import is_ipv6
 from lib.host import Host
+from lib.netutil import is_ipv6
 from lib.pool import Pool
 from lib.sr import SR
 from lib.vm import VM, vm_cache_key_from_def
@@ -321,7 +321,7 @@ def xfail_on_xcpng_8_3(host, request):
 @pytest.fixture(scope='session')
 def host_no_ipv6(host):
     if is_ipv6(host.hostname_or_ip):
-        pytest.skip(f"This test requires an IPv4 XCP-ng")
+        pytest.skip("This test requires an IPv4 XCP-ng")
 
 @pytest.fixture(scope="session")
 def shared_sr(host):
@@ -442,9 +442,7 @@ def vm_ref(request):
             logging.info(">> No VM specified on CLI, and no default found in test definition. Using global default.")
             ref = 'mini-linux-x86_64-bios'
 
-    if is_uuid(ref):
-        return ref
-    elif ref.startswith('http'):
+    if is_uuid(ref) or ref.startswith('http'):
         return ref
     else:
         return vm_image(ref)
diff --git a/data.py-dist b/data.py-dist
@@ -17,7 +17,7 @@ HOST_DEFAULT_PASSWORD = ""
 
 def hash_password(password):
     """Hash password for /etc/password."""
-    salt = crypt.mksalt(crypt.METHOD_SHA512)
+    salt = crypt.mksalt(crypt.METHOD_SHA512)  # type: ignore
     return crypt.crypt(password, salt)
 
 HOST_DEFAULT_PASSWORD_HASH = hash_password(HOST_DEFAULT_PASSWORD)
diff --git a/lib/commands.py b/lib/commands.py
@@ -86,6 +86,7 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
         opts.append(f'-o "ControlPath ~/.ssh/control-{os.getpid()}:%h:%p:%r"')
         opts.append('-o "ControlPersist 10m"')
         opts.append('-o "ServerAliveInterval 10s"')
+        opts.append('-o "LogLevel ERROR"')
     else:
         opts.append('-o "ControlMaster no"')
 
diff --git a/pkgfixtures.py b/pkgfixtures.py
@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import pytest
-from typing import TYPE_CHECKING, Generator
 
 import logging
 
 from lib.common import setup_formatted_and_mounted_disk, teardown_formatted_and_mounted_disk
 
+from typing import TYPE_CHECKING, Generator
+
 if TYPE_CHECKING:
     from lib.common import DiskDevName
     from lib.host import Host
diff --git a/pyproject.toml b/pyproject.toml
@@ -86,6 +86,10 @@ pydocstyle.convention = "pep257"
   "F401",  # F401 unused-import
   "F811",  # F811 redefined-while-unused
 ]
+"conftest.py" = [
+  "F401",  # F401 unused-import
+  "F811",  # F811 redefined-while-unused
+]
 
 [tool.ruff.lint.isort.sections]
 testing = ["pytest*"]
diff --git a/scripts/install_xcpng.py b/scripts/install_xcpng.py
@@ -17,7 +17,7 @@
 # flake8: noqa: E402
 sys.path.append(f"{os.path.abspath(os.path.dirname(__file__))}/..")
 from lib import pxe
-from lib.commands import SSHCommandFailed, scp, ssh
+from lib.commands import SSHCommandFailed, ssh
 from lib.common import is_uuid, wait_for
 from lib.host import host_data
 from lib.pool import Pool
@@ -62,7 +62,7 @@ def generate_answerfile(directory, installer, hostname_or_ip, target_hostname, a
 </installation>
         """)
         elif action == 'restore':
-            answerfile.write(f"""<?xml version="1.0"?>
+            answerfile.write("""<?xml version="1.0"?>
 <restore>
 </restore>
         """)
@@ -160,7 +160,7 @@ def main():
     try:
         pool = Pool(args.host) # will fail if host is not XCP-ng or XAPI doesn't respond yet
     except Exception as e:
-        raise Exception(f"Host `{args.host}` isn't ready or isn't an XCP-ng host")
+        raise Exception(f"Host `{args.host}` isn't ready or isn't an XCP-ng host") from e
 
     host = pool.master
     assert host.is_enabled()
@@ -178,6 +178,7 @@ def main():
     vm = VM(args.vm_uuid, host)
     vif = vm.vifs()[0]
     mac_address = vif.param_get('MAC')
+    assert mac_address is not None
     with tempfile.TemporaryDirectory(suffix=mac_address) as tmp_local_path:
         logging.info('Generate files: answerfile.xml and boot.conf')
         hdd = 'nvme0n1' if vm.is_uefi else 'sda'
@@ -201,11 +202,12 @@ def main():
             "Waiting for the installation process to complete and the VM to reboot and be up", 3600, 10
         )
         vm_ip_address = get_new_host_ip(mac_address)
+        assert vm_ip_address is not None
         logging.info('The IP address of the installed XCP-ng is: ' + vm_ip_address)
         wait_for(lambda: is_new_host_ready(vm_ip_address), "Waiting for XAPI to be ready", 600, 10)
         pool2 = Pool(vm_ip_address)
         host2 = pool2.master
-        host2.inventory = host2._get_xensource_inventory()
+        host2.inventory = host2._get_xensource_inventory()  # noqa: SLF001
         check_mac_address(host2, mac_address)
         logging.info(f'Target host is started and enabled in version: {host2.xcp_version}')
         if args.action == 'restore' and host2.xcp_version >= version.parse(xcp_version):
diff --git a/scripts/xcpng-fs-diff.py b/scripts/xcpng-fs-diff.py
@@ -56,11 +56,7 @@ class DataType(StrEnum):
     PACKAGE = auto()
 
 def ignore_file(filename, ignored_files):
-    for i in ignored_files:
-        if fnmatch(filename, i):
-            return True
-
-    return False
+    return any(fnmatch(filename, i) for i in ignored_files)
 
 def ssh_cmd(host, cmd):
     args = ["ssh", f"root@{host}", cmd]
@@ -155,9 +151,9 @@ def sftp_get(host, remote_file, local_file):
     return res
 
 def remote_diff(host_ref, host_test, filename):
+    file_ref = None
+    file_test = None
     try:
-        file_ref = None
-        file_test = None
 
         # check remote files are text files
         cmd = f"file -b {shlex.quote(filename)}"
diff --git a/scripts/xva_bridge.py b/scripts/xva_bridge.py
@@ -8,8 +8,8 @@
 import os
 from xml.dom import minidom
 
-import libarchive
-import libarchive.ffi
+import libarchive  # type: ignore
+import libarchive.ffi  # type: ignore
 
 class XvaHeaderMember:
     def __init__(self, member: minidom.Element):
@@ -124,10 +124,10 @@ def set_bridge(self, bridge: str):
                         setattr(new_entry, attr, getattr(entry, attr))
 
                     # ArchiveEntry doesn't expose block copying, so write the entry manually via the FFI interface
-                    libarchive.ffi.write_header(output_file._pointer, new_entry._entry_p)
+                    libarchive.ffi.write_header(output_file._pointer, new_entry._entry_p)  # type: ignore # noqa: SLF001
                     for block in entry.get_blocks():
-                        libarchive.ffi.write_data(output_file._pointer, block, len(block))
-                    libarchive.ffi.write_finish_entry(output_file._pointer)
+                        libarchive.ffi.write_data(output_file._pointer, block, len(block)) # type: ignore # noqa: SLF001
+                    libarchive.ffi.write_finish_entry(output_file._pointer)  # type: ignore  # noqa: SLF001
 
             if args.in_place:
                 backup_path = args.backup_path
diff --git a/tests/install/test.py b/tests/install/test.py
@@ -85,7 +85,7 @@ class TestNested:
         .top_append(
             {"iso": {"TAG": "source", "type": "local"},
              "net": {"TAG": "source", "type": "url",
-                     "CONTENTS": ISO_IMAGES[iso_version]['net-url']},
+                     "CONTENTS": ISO_IMAGES[iso_version]['net-url']},  # type: ignore
              }[package_source],
             {"TAG": "admin-interface", "name": "eth0", "proto": "dhcp"},
             {"TAG": "primary-disk",
@@ -339,7 +339,7 @@ def test_boot_inst(self, create_vms,
         lambda system_disks_names, package_source, iso_version: AnswerFile("UPGRADE").top_append(
             {"iso": {"TAG": "source", "type": "local"},
              "net": {"TAG": "source", "type": "url",
-                     "CONTENTS": ISO_IMAGES[iso_version]['net-url']},
+                     "CONTENTS": ISO_IMAGES[iso_version]['net-url']},  # type: ignore
              }[package_source],
             {"TAG": "existing-installation",
              "CONTENTS": system_disks_names[0]},
diff --git a/tests/misc/test_file_server.py b/tests/misc/test_file_server.py
@@ -3,6 +3,7 @@
 import re
 
 import lib.commands as commands
+from lib.host import Host
 from lib.netutil import wrap_ip
 
 # These tests are meant to test an host fileserver behavior.
@@ -15,7 +16,7 @@ def _header_equal(header, name, value):
     regex = fr"{name}:\s?{re.escape(value)}"
     return re.match(regex, header) is not None
 
-def test_fileserver_redirect_https(host):
+def test_fileserver_redirect_https(host: Host):
     path = "/path/to/dir/file.txt"
     ip = wrap_ip(host.hostname_or_ip)
     res = commands.local_cmd(["curl", "-s", "-i", "http://" + ip + path])
@@ -28,18 +29,19 @@ class TestHSTS:
     HSTS_HEADER_NAME = "strict-transport-security"
     HSTS_HEADER_VALUE = "max-age=63072000"
 
-    def __get_header(host):
+    @staticmethod
+    def __get_header(host: Host):
         res = commands.local_cmd(
             ["curl", "-s", "-XGET", "-k", "-I", "https://" + wrap_ip(host.hostname_or_ip)]
         )
         return res.stdout.splitlines()
 
-    def test_fileserver_hsts_default(self, host):
+    def test_fileserver_hsts_default(self, host: Host):
         # By default HSTS header should not be set
         for line in TestHSTS.__get_header(host):
             assert not _header_equal(line, TestHSTS.HSTS_HEADER_NAME, TestHSTS.HSTS_HEADER_VALUE)
 
-    def test_fileserver_hsts(self, host_with_hsts):
+    def test_fileserver_hsts(self, host_with_hsts: Host):
         hsts_header_found = False
 
         for line in TestHSTS.__get_header(host_with_hsts):
diff --git a/tests/network/test_vif_management.py b/tests/network/test_vif_management.py
@@ -14,7 +14,9 @@ def count_interfaces(vm: VM) -> int:
     ip_link_show = vm.ssh_with_result(['ip', 'link', 'show'])
     assert ip_link_show.returncode == 0
 
-    return ip_link_show.stdout.count('link/ether')
+    stdout = ip_link_show.stdout
+    assert isinstance(stdout, str)
+    return stdout.count('link/ether')
 
 @pytest.mark.small_vm
 class TestVIFManagement:
@@ -29,10 +31,10 @@ def test_vif_management(self, running_unix_vm: VM):
         # number of interfaces in the VM
         n_interfaces = count_interfaces(vm)
 
-        try:
-            # create a new VIF in the management network
-            vif_new = vm.create_vif(n_vif, network_uuid=network_uuid)
+        # create a new VIF in the management network
+        vif_new = vm.create_vif(n_vif, network_uuid=network_uuid)
 
+        try:
             # check one more VIF
             assert len(vm.vifs()) == n_vif + 1
             assert count_interfaces(vm) == n_interfaces
diff --git a/tests/packages/bugtool/test_bugtool.py b/tests/packages/bugtool/test_bugtool.py
@@ -18,6 +18,7 @@ def verify_contains(host: Host, archive: str, files: list[str]):
 class TestsBugtool:
     # Verify a minimal bugtool invocation that only queries certain capabilities
     def test_bugtool_entries(self, host):
+        filename = ''
         try:
             filename = host.ssh(['xen-bugtool', '-y', '-s',
                                  '--entries=xenserver-logs,xenserver-databases,system-logs'])
@@ -33,6 +34,7 @@ def test_bugtool_entries(self, host):
 
     # Verify that a full xen-bugtool invocation contains the most essential files
     def test_bugtool_all(self, host):
+        filename = ''
         try:
             filename = host.ssh(['xen-bugtool', '-y', '-s'])
             verify_contains(host, filename,
diff --git a/tests/packages/netdata/test_netdata.py b/tests/packages/netdata/test_netdata.py
@@ -2,6 +2,7 @@
 
 import subprocess
 
+from lib.host import Host
 from lib.netutil import wrap_ip
 
 # This test installs netdata and netdata-ui packages, verifies the service status,
@@ -13,7 +14,8 @@
 
 @pytest.mark.usefixtures("host_with_netdata")
 class TestsNetdata:
-    def __get_headers(host, port, path=None):
+    @staticmethod
+    def __get_headers(host: Host, port, path=None):
         url = f"http://{wrap_ip(host.hostname_or_ip)}:{port}"
         if path is not None:
             url += f"/{path}"
diff --git a/tests/xen/test_hvmloader_grant-table.py b/tests/xen/test_hvmloader_grant-table.py
@@ -38,6 +38,7 @@ def are_grant_tables_inside_uncachable_mapping(mtrr_ranges_list, pci_io_address)
 
     for mtrr_range in mtrr_ranges_list:
         m = re.match(regexp, mtrr_range)
+        assert m is not None
         base_address = int(m.group(2), 16)
         size = int(m.group(3)) * 1024 * 1024
         cache_mode = m.group(5)
