Add CI coverage for vmupdate formatting and pylint checks.

.gitlab-ci.yml

@@ -1,9 +1,17 @@
 include:
+- file: /common.yml
+  project: QubesOS/qubes-continuous-integration
 - file: /r4.3/gitlab-base.yml
   project: QubesOS/qubes-continuous-integration
 - file: /r4.3/gitlab-host.yml
   project: QubesOS/qubes-continuous-integration
 
+lint:
+  extends: .lint
+  stage: checks
+  variables:
+    DIR: vmupdate
+
 checks:tests:
   stage: checks
   variables:

.pylintrc

@@ -0,0 +1,28 @@
+[MASTER]
+init-hook='import sys; sys.path.append("vmupdate/agent")'
+
+[MESSAGES CONTROL]
+disable=
+    broad-exception-caught,
+    duplicate-code,
+    import-error,
+    invalid-name,
+    line-too-long,
+    missing-class-docstring,
+    missing-function-docstring,
+    missing-module-docstring,
+    possibly-used-before-assignment,
+    too-few-public-methods,
+    too-many-arguments,
+    too-many-branches,
+    too-many-instance-attributes,
+    too-many-locals,
+    too-many-positional-arguments,
+    too-many-return-statements,
+    too-many-statements,
+    unspecified-encoding,
+    consider-using-with,
+    f-string-without-interpolation,
+    implicit-str-concat,
+    unused-argument,
+    unused-import

setup.py

@@ -16,18 +16,17 @@
 # with this program; if not, see <http://www.gnu.org/licenses/>.
 import setuptools
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     setuptools.setup(
-        name='qubes-vmupdate',
-        version=open('version').read().strip(),
-        author='Invisible Things Lab',
-        author_email='qubes-devel@googlegroups.com',
-        description='Qubes VM updater',
-        license='GPL2+',
-        url='https://www.qubes-os.org/',
+        name="qubes-vmupdate",
+        version=open("version").read().strip(),
+        author="Invisible Things Lab",
+        author_email="qubes-devel@googlegroups.com",
+        description="Qubes VM updater",
+        license="GPL2+",
+        url="https://www.qubes-os.org/",
         packages=setuptools.find_packages(include=("vmupdate", "vmupdate*")),
         entry_points={
-            'console_scripts':
-                'qubes-vm-update = vmupdate.vmupdate:main',
+            "console_scripts": "qubes-vm-update = vmupdate.vmupdate:main",
         },
-    )
+    )

vmupdate/agent/source/common/package_manager.py

@@ -19,6 +19,7 @@
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
 # USA.
 """package manager for VMs"""
+
 import io
 import logging
 import subprocess

vmupdate/agent/source/plugins/whonix17_key.py

@@ -176,11 +176,14 @@
 # sub:e:4096:1:10FDAC53119B3FD6:1389913402:1769172143:::
 # sub:e:4096:1:CB8D50BB77BB3C48:1389913659:1769172143:::
 
+
 def whonix17_key(os_data, log, **kwargs):
     """
     Update kicksecure signing key - https://www.kicksecure.com/wiki/EXPKEYSIG
     """
-    if os_data.get("codename", "") != "bookworm" or not os.path.exists(KEYFILE_PATH):
+    if os_data.get("codename", "") != "bookworm" or not os.path.exists(
+        KEYFILE_PATH
+    ):
         return
 
     # check if key is expire

vmupdate/qube_connection.py

@@ -95,11 +95,12 @@ def __exit__(self, exc_type, exc_val, exc_tb):
         if self.qube.is_running() and not self._initially_running:
             if self._has_assigned_pci_devices(self.qube):
                 self.logger.info(
-                    'Waiting for full shutdown %s (PCI devices assigned)',
-                    self.qube.name)
+                    "Waiting for full shutdown %s (PCI devices assigned)",
+                    self.qube.name,
+                )
                 shutdown_domains([self.qube], self.logger)
             else:
-                self.logger.info('Shutdown %s', self.qube.name)
+                self.logger.info("Shutdown %s", self.qube.name)
                 self.qube.shutdown()
 
         self.__connected = False
@@ -108,7 +109,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     def _has_assigned_pci_devices(vm) -> bool:
         """Return True when VM has assigned PCI devices."""
         try:
-            return any(vm.devices['pci'].get_assigned_devices())
+            return any(vm.devices["pci"].get_assigned_devices())
         except qubesadmin.exc.QubesDaemonAccessError:
             return False
 

vmupdate/tests/test_qube_connection.py

@@ -28,14 +28,19 @@ def test_wait_for_shutdown_when_vm_started_by_update(shutdown_domains):
     vm = Mock()
     vm.name = "hvm1"
     vm.is_running.side_effect = [False, True]
-    vm.devices = {'pci': Mock()}
-    vm.devices['pci'].get_assigned_devices.return_value = ["00_1f.2"]
+    vm.devices = {"pci": Mock()}
+    vm.devices["pci"].get_assigned_devices.return_value = ["00_1f.2"]
     status_notifier = Mock()
     logger = Mock()
 
     with QubeConnection(
-            vm, "/tmp/qubes-update", cleanup=False, logger=logger,
-            show_progress=False, status_notifier=status_notifier):
+        vm,
+        "/tmp/qubes-update",
+        cleanup=False,
+        logger=logger,
+        show_progress=False,
+        status_notifier=status_notifier,
+    ):
         pass
 
     shutdown_domains.assert_called_once_with([vm], logger)
@@ -47,14 +52,19 @@ def test_do_not_wait_for_shutdown_without_assigned_pci(shutdown_domains):
     vm = Mock()
     vm.name = "hvm2"
     vm.is_running.side_effect = [False, True]
-    vm.devices = {'pci': Mock()}
-    vm.devices['pci'].get_assigned_devices.return_value = []
+    vm.devices = {"pci": Mock()}
+    vm.devices["pci"].get_assigned_devices.return_value = []
     status_notifier = Mock()
     logger = Mock()
 
     with QubeConnection(
-            vm, "/tmp/qubes-update", cleanup=False, logger=logger,
-            show_progress=False, status_notifier=status_notifier):
+        vm,
+        "/tmp/qubes-update",
+        cleanup=False,
+        logger=logger,
+        show_progress=False,
+        status_notifier=status_notifier,
+    ):
         pass
 
     vm.shutdown.assert_called_once_with()
@@ -66,14 +76,19 @@ def test_do_not_shutdown_if_vm_was_already_running(shutdown_domains):
     vm = Mock()
     vm.name = "hvm3"
     vm.is_running.return_value = True
-    vm.devices = {'pci': Mock()}
-    vm.devices['pci'].get_assigned_devices.return_value = ["00_1f.2"]
+    vm.devices = {"pci": Mock()}
+    vm.devices["pci"].get_assigned_devices.return_value = ["00_1f.2"]
     status_notifier = Mock()
     logger = Mock()
 
     with QubeConnection(
-            vm, "/tmp/qubes-update", cleanup=False, logger=logger,
-            show_progress=False, status_notifier=status_notifier):
+        vm,
+        "/tmp/qubes-update",
+        cleanup=False,
+        logger=logger,
+        show_progress=False,
+        status_notifier=status_notifier,
+    ):
         pass
 
     vm.shutdown.assert_not_called()

vmupdate/utils.py

@@ -70,13 +70,15 @@ def is_stale(vm, expiration_period):
     """Return True if VM has not been checked for updates recently."""
     today = datetime.today()
     try:
-        if not ('qrexec' in vm.features.keys()
-                and vm.features.get('os', '') == 'Linux'):
+        if not (
+            "qrexec" in vm.features.keys()
+            and vm.features.get("os", "") == "Linux"
+        ):
             return False
 
         last_update_str = vm.features.check_with_template(
-            'last-updates-check',
-            datetime.fromtimestamp(0).strftime('%Y-%m-%d %H:%M:%S')
+            "last-updates-check",
+            datetime.fromtimestamp(0).strftime("%Y-%m-%d %H:%M:%S"),
         )
         last_update = datetime.fromisoformat(last_update_str)
         if (today - last_update).days > expiration_period:

vmupdate/vmupdate.py

@@ -14,8 +14,12 @@
 import qubesadmin.exc
 from vmupdate.agent.source.status import FinalStatus
 from vmupdate.agent.source.common.exit_codes import EXIT
-from vmupdate.utils import shutdown_domains, get_feature, get_boolean_feature, \
-    is_stale
+from vmupdate.utils import (
+    shutdown_domains,
+    get_feature,
+    get_boolean_feature,
+    is_stale,
+)
 from . import update_manager
 from .agent.source.args import AgentArgs
 
@@ -492,7 +496,6 @@ def get_derived_vm_to_apply(templates, derived_statuses):
     return to_restart, to_shutdown
 
 
-
 def restart_vms(to_restart, log):
     """
     Try to restart vms.