lint python code with ruff

.github/workflows/jobs-check.yml

@@ -55,3 +55,21 @@ jobs:
       run: cp data.py-dist data.py
     - name: Check with pyright
       run: pyright lib/ # tests/
+
+  ruff:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: 3.8
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements/base.txt
+        pip install ruff
+    - name: Create a dummy data.py
+      run: cp data.py-dist data.py
+    - name: Check with ruff
+      run: ruff check lib/ tests/

lib/basevm.py

@@ -2,7 +2,6 @@
 
 from typing import Any, Literal, Optional, overload, TYPE_CHECKING
 
-import lib.commands as commands
 if TYPE_CHECKING:
     import lib.host
 
@@ -87,10 +86,7 @@ def all_vdis_on_host(self, host):
         return True
 
     def all_vdis_on_sr(self, sr) -> bool:
-        for vdi_uuid in self.vdi_uuids():
-            if self.host.pool.get_vdi_sr_uuid(vdi_uuid) != sr.uuid:
-                return False
-        return True
+        return all(self.host.pool.get_vdi_sr_uuid(vdi_uuid) == sr.uuid for vdi_uuid in self.vdi_uuids())
 
     def get_sr(self):
         # in this method we assume the SR of the first VDI is the VM SR

lib/commands.py

@@ -74,10 +74,7 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
         opts.append('-o "LogLevel ERROR"')
         opts.append('-o "UserKnownHostsFile /dev/null"')
 
-    if isinstance(cmd, str):
-        command = cmd
-    else:
-        command = " ".join(cmd)
+    command = cmd if isinstance(cmd, str) else ' '.join(cmd)
 
     ssh_cmd = f"ssh root@{hostname_or_ip} {' '.join(opts)} {shlex.quote(command)}"
 

lib/host.py

@@ -3,7 +3,6 @@
 import logging
 import os
 import shlex
-import subprocess
 import tempfile
 import uuid
 

lib/pif.py

@@ -1,4 +1,3 @@
-import lib.commands as commands
 
 from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
 

lib/pool.py

@@ -1,13 +1,14 @@
 import logging
 import traceback
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Optional
 
 from packaging import version
 
 import lib.commands as commands
 from lib.common import _param_get, _param_set, safe_split, wait_for, wait_for_not
 from lib.host import Host
 from lib.sr import SR
+import contextlib
 
 
 class Pool:
@@ -63,10 +64,8 @@ def exec_on_hosts_on_error_rollback(self, func, rollback_func, host_list=[]):
 
                     logging.info("Attempting to run the rollback function on host(s) "
                                  f"{', '.join([str(h) for h in rollback_hosts])}...")
-                    try:
+                    with contextlib.suppress(Exception):
                         self.exec_on_hosts_on_error_continue(rollback_func, rollback_hosts)
-                    except Exception:
-                        pass
                 raise e
 
     def exec_on_hosts_on_error_continue(self, func, host_list=[]):
@@ -236,7 +235,7 @@ def install_custom_uefi_certs(self, auths):
             assert 'db' in auths_dict
 
             logging.info('Installing auths to pool: %s' % list(auths_dict.keys()))
-            for key in auths_dict.keys():
+            for key in auths_dict:
                 value = host.ssh([f'md5sum {auths_dict[key]} | cut -d " " -f 1'])
                 logging.debug('Key: %s, value: %s' % (key, value))
             params = [auths_dict['PK'], auths_dict['KEK'], auths_dict['db']]

lib/vif.py

@@ -1,4 +1,3 @@
-import lib.commands as commands
 
 from lib.common import _param_add, _param_clear, _param_get, _param_remove, _param_set
 

lib/vm.py

@@ -2,9 +2,8 @@
 
 import logging
 import os
-import subprocess
 import tempfile
-from typing import Dict, List, Literal, Optional, overload, TYPE_CHECKING, Union
+from typing import List, Literal, Optional, overload, TYPE_CHECKING, Union
 
 import lib.commands as commands
 import lib.efi as efi

pyproject.toml

@@ -1,2 +1,18 @@
 [tool.pyright]
 typeCheckingMode = "standard"
+
+[tool.ruff]
+preview = true
+line-length = 120
+exclude = ["data.py", "vm_data.py", ".git"]
+
+[tool.ruff.format]
+quote-style = "preserve"
+
+[tool.ruff.lint]
+select = ["F", "SLF", "SIM"]
+
+[tool.ruff.lint.extend-per-file-ignores]
+# pytest requires some import and function arguments to match, but
+# the linter doesn't know that
+"tests/**/*.py" = ["F401", "F811"]

tests/fs_diff/test_fs_diff.py

@@ -9,7 +9,7 @@
 def test_fs_diff(hosts):
     assert len(hosts) == 2, "This test requires exactly 2 hosts"
 
-    assert (hosts[0].xcp_version == hosts[1].xcp_version), f"Host versions must be the same"
+    assert (hosts[0].xcp_version == hosts[1].xcp_version), "Host versions must be the same"
 
     fsdiff = os.path.realpath(f"{os.path.dirname(__file__)}/../../scripts/xcpng-fs-diff.py")
 

tests/guest_tools/win/test_xenclean.py

@@ -53,6 +53,6 @@ def test_xenclean_with_other_tools(self, vm_install_other_drivers: Tuple[VM, Dic
         if param.get("vendor_device"):
             pytest.skip("Skipping XenClean with vendor device present")
             return
-        logging.info(f"XenClean with other tools")
+        logging.info("XenClean with other tools")
         run_xenclean(vm, guest_tools_iso)
         assert vm.are_windows_tools_uninstalled()

tests/network/conftest.py

@@ -4,4 +4,4 @@
 def host_no_sdn_controller(host):
     """ An XCP-ng with no SDN controller. """
     if host.xe('sdn-controller-list', minimal=True):
-        pytest.skip(f"This test requires an XCP-ng with no SDN controller")
+        pytest.skip("This test requires an XCP-ng with no SDN controller")

tests/storage/__init__.py

@@ -1 +1,7 @@
-from .storage import *
+from .storage import (
+    try_to_create_sr_with_missing_device,
+    cold_migration_then_come_back,
+    live_storage_migration_then_come_back,
+    live_storage_migration_then_come_back,
+    vdi_is_open,
+)

tests/storage/glusterfs/conftest.py

@@ -1,3 +1,4 @@
+import contextlib
 import logging
 import pytest
 
@@ -101,11 +102,9 @@ def teardown_for_host(h):
             if h.hostname_or_ip != h2.hostname_or_ip:
                 h.ssh(['gluster', '--mode=script', 'peer', 'detach', h2.hostname_or_ip])
 
-        try:
+        with contextlib.suppress(Exception):
             # Volume might already be stopped if failure happened on delete
             h.ssh(['gluster', '--mode=script', 'volume', 'stop', 'vol0'])
-        except Exception as e:
-            pass
 
         h.ssh(['gluster', '--mode=script', 'volume', 'delete', 'vol0'])
 

tests/storage/iso/conftest.py

@@ -26,10 +26,7 @@ def copy_tools_iso_to_iso_sr(host, sr, location=None):
     # copy the ISO file to the right location
     iso_path = host.ssh(['find', '/opt/xensource/packages/iso/', '-name', '"*.iso"'])
     iso_new_name = sr.uuid + "_test.iso"
-    if location is not None:
-        iso_new_path = f"{location}/{iso_new_name}"
-    else:
-        iso_new_path = f"/run/sr-mount/{sr.uuid}/{iso_new_name}"
+    iso_new_path = f'{location}/{iso_new_name}' if location is not None else f'/run/sr-mount/{sr.uuid}/{iso_new_name}'
     host.ssh(['cp', '-f', iso_path, iso_new_path])
     sr.scan()
     return iso_new_path

tests/storage/linstor/test_linstor_sr.py

@@ -6,6 +6,7 @@
 from lib.commands import SSHCommandFailed
 from lib.common import wait_for, vm_image
 from tests.storage import vdi_is_open
+import contextlib
 
 # Requirements:
 # - two or more XCP-ng hosts >= 8.2 with additional unused disk(s) for the SR
@@ -28,10 +29,8 @@ def test_create_sr_without_linstor(self, host, lvm_disks, provisioning_type, sto
                 'redundancy': '1',
                 'provisioning': provisioning_type
             }, shared=True)
-            try:
+            with contextlib.suppress(Exception):
                 sr.destroy()
-            except Exception:
-                pass
             assert False, "SR creation should not have succeeded!"
         except SSHCommandFailed as e:
             logging.info("SR creation failed, as expected: {}".format(e))

tests/storage/storage.py

@@ -4,7 +4,7 @@
 
 def try_to_create_sr_with_missing_device(sr_type, label, host):
     try:
-        sr = host.sr_create(sr_type, label, {}, verify=True)
+        host.sr_create(sr_type, label, {}, verify=True)
     except SSHCommandFailed as e:
         assert e.stdout == (
             'Error code: SR_BACKEND_FAILURE_90\nError parameters: , '

tests/xen/test_xtf.py

@@ -48,7 +48,7 @@ def test_all(self, host, xtf_runner):
                              "Checking whether they belong to the allowed list...")
                 for skipped_test in skipped_tests:
                     if skipped_test not in self._common_skips:
-                        logging.error(f"... At least one doesn't")
+                        logging.error("... At least one doesn't")
                         raise
                 logging.info("... They do")
             else: