Merge remote-tracking branch 'origin/pr/215'

* origin/pr/215:
  ci: add vmupdate black and pylint checks
  style: format vmupdate with black

Pull request description:

This extends the existing GitLab test job so vmupdate changes are checked with black and pylint after the unit tests run.

The patch also formats the current vmupdate files so the new black check starts from a clean baseline, and adds a focused pylint configuration for the checks that are actionable in this tree.

Mypy is intentionally not enabled in this PR because current upstream still has existing typing errors that need a separate cleanup before it can be a blocking CI check.

fixes: #213 (comment)

Author: marmarek

.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.