Lint with mypy on CI

.gitlab-ci.yml

@@ -12,6 +12,19 @@ lint:
   variables:
     DIR: vmupdate
 
+mypy:
+  stage: checks
+  tags:
+    - docker
+  before_script:
+    - sudo dnf install -y python3-mypy python3-pip
+    - sudo python3 -m pip install lxml-stubs types-docutils
+  script:
+    - mypy --install-types --non-interactive --ignore-missing-imports --junit-xml mypy.xml vmupdate
+  artifacts:
+    reports:
+      junit: mypy.xml
+
 checks:tests:
   stage: checks
   variables:

vmupdate/agent/source/args.py

@@ -72,7 +72,11 @@ class AgentArgs:
             "help": argparse.SUPPRESS,
         },
     }
-    ALL_OPTIONS = {**OPTIONS, **EXCLUSIVE_OPTIONS_1, **EXCLUSIVE_OPTIONS_2}
+    ALL_OPTIONS: dict[tuple, dict[str, str]] = {
+        **OPTIONS,
+        **EXCLUSIVE_OPTIONS_1,
+        **EXCLUSIVE_OPTIONS_2,
+    }
 
     @staticmethod
     def add_arguments(parser):

vmupdate/agent/source/common/package_manager.py

@@ -182,6 +182,7 @@ def install_requirements(
                         break
                 else:
                     to_upgrade[pkg] = version
+        assert isinstance(self.package_manager, str)
         if to_install:
             cmd = [self.package_manager, "-q", "-y", "install", *to_install]
             result += self.run_cmd(cmd)
@@ -308,6 +309,7 @@ def upgrade_internal(self, remove_obsolete: bool) -> ProcessResult:
         """
         Just run upgrade via CLI.
         """
+        assert isinstance(self.package_manager, str)
         cmd = [self.package_manager, *self.get_action(remove_obsolete)]
 
         return self.run_cmd(cmd)

vmupdate/agent/source/common/process_result.py

@@ -66,15 +66,15 @@ def from_untrusted_out_err(
         if untrusted_out is None:
             untrusted_out_bytes = b""
         elif isinstance(untrusted_out, str):
-            untrusted_out_bytes: bytes = untrusted_out.encode()
+            untrusted_out_bytes = untrusted_out.encode()
         else:
             untrusted_out_bytes = untrusted_out
         out = ProcessResult.sanitize_output(untrusted_out_bytes)
 
         if untrusted_err is None:
             untrusted_err_bytes = b""
         elif isinstance(untrusted_err, str):
-            untrusted_err_bytes: bytes = untrusted_err.encode()
+            untrusted_err_bytes = untrusted_err.encode()
         else:
             untrusted_err_bytes = untrusted_err
         err = ProcessResult.sanitize_output(untrusted_err_bytes)

vmupdate/agent/source/common/progress_reporter.py

@@ -23,6 +23,7 @@
 import os
 import sys
 from typing import Callable, Optional
+from logging import Logger
 
 
 class Progress:
@@ -32,13 +33,13 @@ def __init__(
         log,
     ):
         self.weight = weight
-        self._callback = None
-        self._start_percent = None
-        self._stop_percent = None
-        self._last_percent = None
-        self._stdout = None
-        self._stderr = None
-        self.log = log
+        self._callback: Optional[Callable[[float], None]] = None
+        self._start_percent: Optional[float] = None
+        self._stop_percent: Optional[float] = None
+        self._last_percent: Optional[float] = None
+        self._stdout: Optional[io.TextIOWrapper] = None
+        self._stderr: Optional[io.TextIOWrapper] = None
+        self.log: Logger = log
 
     def init(
         self,

vmupdate/agent/source/dnf/dnf5_api.py

@@ -164,10 +164,10 @@ class FetchProgress(DownloadCallbacks, Progress):
     def __init__(self, weight: int, log):
         DownloadCallbacks.__init__(self)
         Progress.__init__(self, weight, log)
-        self.bytes_to_fetch = 0
-        self.bytes_fetched = 0
-        self.package_bytes = {}
-        self.package_names = {}
+        self.bytes_to_fetch = 0.0
+        self.bytes_fetched = 0.0
+        self.package_bytes: dict[int, float] = {}
+        self.package_names: dict[int, str] = {}
         self.count = 0
         self.fetching_notified = False
 
@@ -267,9 +267,9 @@ class UpgradeProgress(TransactionCallbacks, Progress):
     def __init__(self, weight: int, log):
         TransactionCallbacks.__init__(self)
         Progress.__init__(self, weight, log)
-        self.pgks = None
-        self.pgks_done = None
-        self.processed_packages = set()
+        self.pgks: int | None = None
+        self.pgks_done: int | None = None
+        self.processed_packages: set[str] = set()
 
     def install_progress(
         self, item: libdnf5.base.TransactionPackage, amount: int, total: int
@@ -287,6 +287,8 @@ def install_progress(
             print(f"Installing {package}", flush=True)
             self.processed_packages.add(package)
         pkg_progress = amount / total
+        assert isinstance(self.pgks_done, int)
+        assert isinstance(self.pgks, int)
         percent = (self.pgks_done + pkg_progress) / self.pgks * 100
         self.notify_callback(percent)
 
@@ -314,6 +316,8 @@ def uninstall_progress(
             print(f"Uninstalling {package}", flush=True)
             self.processed_packages.add(package)
         pkg_progress = amount / total
+        assert isinstance(self.pgks_done, int)
+        assert isinstance(self.pgks, int)
         percent = (self.pgks_done + pkg_progress) / self.pgks * 100
         self.notify_callback(percent)
 

vmupdate/agent/source/dnf/dnf_api.py

@@ -194,7 +194,7 @@ def __init__(self, weight: int, log, refresh: bool = False):
         self.bytes_to_fetch = None
         self.bytes_fetched = 0
         self.action = "refresh" if refresh else "fetch"
-        self.package_bytes = {}
+        self.package_bytes: dict[int, int] = {}
 
     def end(self, payload, status, msg):
         """Communicate the information that `payload` has finished downloading.

vmupdate/agent/source/plugins/__init__.py

@@ -2,10 +2,10 @@
 import os.path
 import glob
 
-modules = sorted(glob.glob(os.path.join(os.path.dirname(__file__), "*.py")))
+modules_str = sorted(glob.glob(os.path.join(os.path.dirname(__file__), "*.py")))
 __all__ = [
     os.path.basename(f)[:-3]
-    for f in modules
+    for f in modules_str
     if os.path.isfile(f) and not f.endswith("__init__.py")
 ]
 modules = [

vmupdate/agent/source/utils.py

@@ -22,9 +22,10 @@
 import re
 import ast
 from typing import Optional, Dict, Any
+from logging import Logger
 
 
-def get_os_data(logger: Optional = None) -> Dict[str, Any]:
+def get_os_data(logger: Optional[Logger] = None) -> Dict[str, Any]:
     """
     Return dictionary with info about the operating system
 
@@ -72,7 +73,7 @@ def get_os_data(logger: Optional = None) -> Dict[str, Any]:
     return data
 
 
-def _load_os_release(*os_release_files, logger: Optional):
+def _load_os_release(*os_release_files, logger: Optional[Logger]):
     """
     Load os-release as dictionary.
 

vmupdate/update_manager.py

@@ -29,7 +29,7 @@
 import logging
 import multiprocessing
 from os.path import join
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Callable
 
 from tqdm import tqdm
 
@@ -215,6 +215,7 @@ def update(self, progress):
 
     def set_description(self, desc: str):
         self.desc = desc
+        assert SimpleTerminalBar.PARENT_MULTI_BAR is not None
         SimpleTerminalBar.PARENT_MULTI_BAR.print()
 
     def close(self):
@@ -231,7 +232,9 @@ class MultipleUpdateMultipleProgressBar:
     Show update info for each qube in the terminal.
     """
 
-    def __init__(self, dummy, output, max_concurrency, printer: Optional):
+    def __init__(
+        self, dummy, output, max_concurrency, printer: Optional[Callable]
+    ):
         self.dummy = dummy
 
         self.manager = multiprocessing.Manager()
@@ -247,9 +250,9 @@ def __init__(self, dummy, output, max_concurrency, printer: Optional):
         # set SIGINT handler to graceful termination
         signal.signal(signal.SIGINT, self.signal_handler_during_feeding)
 
-        self.progresses = {}
-        self.progress_bars = {}
-        self.statuses = {}
+        self.progresses: dict[str, int | float] = {}
+        self.progress_bars: dict[str, SimpleTerminalBar | tqdm] = {}
+        self.statuses: dict[str, FinalStatus] = {}
         self.output_class = output
         self.print = printer
 
@@ -267,7 +270,7 @@ def add_bar(self, qname: str):
             desc=f"{qname} ({Status.PENDING.value})",
         )
 
-    def feeding(self):
+    def feeding(self) -> None:
         """
         Consume info from queues and update progress bars.
 
@@ -279,7 +282,7 @@ def feeding(self):
         left_to_finish = len(self.progresses)
         while left_to_finish:
             try:
-                feed: Optional[StatusInfo, str] = self.status_notifier.get(
+                feed: Optional[StatusInfo | str] = self.status_notifier.get(
                     block=True
                 )
                 if feed is None:
@@ -420,17 +423,18 @@ def _run_agent(
         self, agent_args, status_notifier, termination
     ) -> ProcessResult:
         self.log.info("Running update agent for %s", self.qube.name)
-        dest_dir = None
-        src_dir = None
+        dest_dir: Optional[str] = None
+        src_dir: Optional[str] = None
         cleanup = False
         if self.qube.klass == "AdminVM":
             if self.dom0:
-                entrypoint = ["sudo", "qubes-dom0-update", "-y"]
+                entrypoint_cmd = ["sudo", "qubes-dom0-update", "-y"]
                 if agent_args.just_print_progress or self.show_progress:
-                    entrypoint.append("--just-print-progress")
+                    entrypoint_cmd.append("--just-print-progress")
                 if agent_args.quiet:
                     # silent is equivalent to quiet for dom0-update
-                    entrypoint.append("--silent")
+                    entrypoint_cmd.append("--silent")
+                entrypoint = " ".join(entrypoint_cmd)
             else:
                 this_dir = os.path.dirname(os.path.realpath(__file__))
                 entrypoint = join(this_dir, UpdateAgentManager.ENTRYPOINT)
@@ -473,7 +477,9 @@ def _transfer_agent(self, qconn, src_dir) -> ProcessResult:
                 return result
         return result
 
-    def _run_entrypoint(self, qconn, entrypoint, agent_args) -> ProcessResult:
+    def _run_entrypoint(
+        self, qconn: QubeConnection, entrypoint: str, agent_args
+    ) -> ProcessResult:
         result = ProcessResult()
         self.log.info(
             "The agent is starting the task in qube: %s", self.qube.name