Improve AliBuild typing and linter strictness

aliBuild

@@ -28,7 +28,7 @@ def doMain(args, parser):
   # do not get fooled when parsing localized messages.
   # We set ALIBUILD_ARCHITECTURE so that it's picked up by the external
   # command which does the reporting.
-  if not "architecture" in args:
+  if "architecture" not in args:
     args.architecture = detectArch()
   ENVIRONMENT_OVERRIDES = {
     "LANG": "C",
@@ -57,7 +57,7 @@ def doMain(args, parser):
       error(e.message)
       exit(1)
 
-  if args.action == "version" or args.action == None:
+  if args.action == "version" or args.action is None:
     print("aliBuild version: {version} ({arch})".format(
       version=__version__ or "unknown", arch=args.architecture or "unknown"))
     sys.exit(0)
@@ -111,13 +111,14 @@ if __name__ == "__main__":
   else:
     os.environ["ALIBUILD_ANALYTICS_USER_UUID"] = open(expanduser("~/.config/alibuild/analytics-uuid")).read().strip()
   try:
-    profiler = "--profile" in sys.argv
-    if profiler:
+    useProfiler = "--profile" in sys.argv
+    if useProfiler:
       print("profiler started")
       import cProfile, pstats
       from io import StringIO
       pr = cProfile.Profile()
       pr.enable()
+
       def profiler():
         pr.disable()
         print("profiler stopped")

alibuild_helpers/__init__.py

@@ -1,6 +1,9 @@
 # This file is needed to package build_template.sh.
 
 # Single-source a PEP440-compliant version using setuptools_scm.
+
+from typing import Optional
+__version__: Optional[str]
 try:
     # This is an sdist or wheel, and it's properly installed.
     from alibuild_helpers._version import version as __version__

alibuild_helpers/analytics.py

@@ -104,7 +104,7 @@ def report_exception(e):
     exd = e.__class__.__name__,
     exf = "1")
 
-def enable_analytics():
+def enable_analytics() -> None:
   if exists(expanduser("~/.config/alibuild/disable-analytics")):
     unlink(expanduser("~/.config/alibuild/disable-analytics"))
   if not exists(expanduser("~/.config/alibuild/analytics-uuid")):

alibuild_helpers/build.py

@@ -35,7 +35,7 @@
 import time
 
 
-def writeAll(fn, txt):
+def writeAll(fn: str, txt: str) -> None:
   f = open(fn, "w")
   f.write(txt)
   f.close()
@@ -983,7 +983,7 @@
       fp.close()
     except:
       from pkg_resources import resource_string
-      cmd_raw = resource_string("alibuild_helpers", 'build_template.sh')
+      cmd_raw = resource_string("alibuild_helpers", 'build_template.sh').decode()
 
     if args.docker:
       cachedTarball = re.sub("^" + workDir, "/sw", spec["cachedTarball"])
@@ -1077,7 +1077,7 @@
        args.develPrefix if "develPrefix" in args and spec["is_devel_pkg"] else spec["version"])
     )
     err = execute(build_command, printer=progress)
-    progress.end("failed" if err else "done", err)
+    progress.end("failed" if err else "done", bool(err))
     report_event("BuildError" if err else "BuildSuccess", spec["package"], " ".join((
       args.architecture,
       spec["version"],

alibuild_helpers/cmd.py

@@ -5,10 +5,11 @@
 from textwrap import dedent
 from subprocess import TimeoutExpired
 from shlex import quote
+from typing import Union, Tuple, List
 
 from alibuild_helpers.log import debug, warning, dieOnError
 
-def decode_with_fallback(data):
+def decode_with_fallback(data : Union[bytes, str]) -> str:
   """Try to decode DATA as utf-8; if that doesn't work, fall back to latin-1.
 
   This combination should cover every possible byte string, as latin-1 covers
@@ -23,7 +24,7 @@
     return str(data)
 
 
-def getoutput(command, timeout=None):
+def getoutput(command:str, timeout: Union[int, None] = None) -> str:
   """Run command, check it succeeded, and return its stdout as a string."""
   proc = Popen(command, shell=isinstance(command, str), stdout=PIPE, stderr=PIPE)
   try:
@@ -37,26 +38,27 @@
   return decode_with_fallback(stdout)
 
 
-def getstatusoutput(command, timeout=None):
+def getstatusoutput(command:str, timeout: Union[int, None] = None) -> Tuple[int, str]:
   """Run command and return its return code and output (stdout and stderr)."""
   proc = Popen(command, shell=isinstance(command, str), stdout=PIPE, stderr=STDOUT)
   try:
-    merged_output, _ = proc.communicate(timeout=timeout)
+    merged_output_bytes, _ = proc.communicate(timeout=timeout)
   except TimeoutExpired:
     warning("Process %r timed out; terminated", command)
     proc.terminate()
-    merged_output, _ = proc.communicate()
-  merged_output = decode_with_fallback(merged_output)
+    merged_output_bytes, _ = proc.communicate()
+  merged_output = decode_with_fallback(merged_output_bytes)
   # Strip a single trailing newline, if one exists, to match the behaviour of
   # subprocess.getstatusoutput.
   if merged_output.endswith("\n"):
     merged_output = merged_output[:-1]
   return proc.returncode, merged_output
 
 
-def execute(command, printer=debug, timeout=None):
+def execute(command: Union[str, List[str]], printer=debug, timeout:Union[int, None] =None) -> int:
   popen = Popen(command, shell=isinstance(command, str), stdout=PIPE, stderr=STDOUT)
   start_time = time.time()
+  assert popen.stdout is not None, "Could not open stdout for command"
   for line in iter(popen.stdout.readline, b""):
     printer("%s", decode_with_fallback(line).strip("\n"))
     if timeout is not None and time.time() > start_time + timeout:

alibuild_helpers/doctor.py

@@ -7,7 +7,7 @@
 from alibuild_helpers.utilities import getPackageList, parseDefaults, readDefaults, validateDefaults
 from alibuild_helpers.cmd import getstatusoutput, DockerRunner
 
-def prunePaths(workDir):
+def prunePaths(workDir: str) -> None:
   for x in ["PATH", "LD_LIBRARY_PATH", "DYLD_LIBRARY_PATH"]:
     if not x in os.environ:
       continue
@@ -52,7 +52,7 @@ def checkRequirements(spec, cmd, homebrew_replacement, getstatusoutput_docker):
           spec.get("system_requirement_missing"))
     return (err, "")
 
-def systemInfo():
+def systemInfo() -> None:
   _,out = getstatusoutput("env")
   debug("Environment:\n%s", out)
   _,out = getstatusoutput("uname -a")

alibuild_helpers/init.py

@@ -2,13 +2,13 @@
 from alibuild_helpers.utilities import getPackageList, parseDefaults, readDefaults, validateDefaults
 from alibuild_helpers.log import debug, error, warning, banner, info
 from alibuild_helpers.log import dieOnError
-from alibuild_helpers.workarea import cleanup_git_log, updateReferenceRepoSpec
+from alibuild_helpers.workarea import updateReferenceRepoSpec
 
 from os.path import join
 import os.path as path
 import os, sys
 
-def parsePackagesDefinition(pkgname):
+def parsePackagesDefinition(pkgname: str):
   return [ dict(zip(["name","ver"], y.split("@")[0:2]))
            for y in [ x+"@" for x in list(filter(lambda y: y, pkgname.split(","))) ] ]
 
@@ -23,8 +23,10 @@ def doInit(args):
          "--dry-run / -n specified. Doing nothing.", ",".join(x["name"] for x in pkgs))
     sys.exit(0)
   try:
-    path.exists(args.develPrefix) or os.mkdir(args.develPrefix)
-    path.exists(args.referenceSources) or os.makedirs(args.referenceSources)
+    if not path.exists(args.develPrefix):
+      os.mkdir(args.develPrefix)
+    if not path.exists(args.referenceSources):
+      os.makedirs(args.referenceSources)
   except OSError as e:
     error("%s", e)
     sys.exit(1)
@@ -66,9 +68,10 @@ def doInit(args):
 
   for p in pkgs:
     spec = specs.get(p["name"])
+    dieOnError(spec is None, "cannot find recipe for package %s" % p["name"])
+    assert spec
     spec["is_devel_pkg"] = False
     spec["scm"] = Git()
-    dieOnError(spec is None, "cannot find recipe for package %s" % p["name"])
     dest = join(args.develPrefix, spec["package"])
     writeRepo = spec.get("write_repo", spec.get("source"))
     dieOnError(not writeRepo, "package %s has no source field and cannot be developed" % spec["package"])

alibuild_helpers/log.py

@@ -3,10 +3,9 @@
 import re
 import time
 import datetime
+from typing import Any
 
-debug, error, warning, info, success = (None, None, None, None, None)
-
-def dieOnError(err, msg):
+def dieOnError(err: Any, msg: str) -> None:
   if err:
     error("%s", msg)
     sys.exit(1)
@@ -60,7 +59,7 @@ def __init__(self, begin_msg=""):
     self.lasttime = 0
     self.STAGES = ".", "..", "...", "....", ".....", "....", "...", ".."
     self.begin_msg = begin_msg
-    self.percent = -1
+    self.percent: float = -1
 
   def __call__(self, txt, *args):
     if logger.level <= logging.DEBUG or not sys.stdout.isatty():
@@ -88,7 +87,7 @@ def __call__(self, txt, *args):
     self.lasttime = time.time()
     sys.stderr.flush()
 
-  def erase(self):
+  def erase(self) -> None:
     nerase = len(self.STAGES[self.count]) if self.count > -1 else 0
     if self.percent > -1:
       nerase = nerase + 7

alibuild_helpers/sync.py

@@ -12,30 +12,15 @@
 from alibuild_helpers.cmd import execute
 from alibuild_helpers.log import debug, info, error, dieOnError, ProgressPrint
 from alibuild_helpers.utilities import resolve_store_path, resolve_links_path, symlink
-
-
-def remote_from_url(read_url, write_url, architecture, work_dir, insecure=False):
-  """Parse remote store URLs and return the correct RemoteSync instance for them."""
-  if read_url.startswith("http"):
-    return HttpRemoteSync(read_url, architecture, work_dir, insecure)
-  if read_url.startswith("s3://"):
-    return S3RemoteSync(read_url, write_url, architecture, work_dir)
-  if read_url.startswith("b3://"):
-    return Boto3RemoteSync(read_url, write_url, architecture, work_dir)
-  if read_url.startswith("cvmfs://"):
-    return CVMFSRemoteSync(read_url, None, architecture, work_dir)
-  if read_url:
-    return RsyncRemoteSync(read_url, write_url, architecture, work_dir)
-  return NoRemoteSync()
-
+from typing import Union
 
 class NoRemoteSync:
   """Helper class which does not do anything to sync"""
-  def fetch_symlinks(self, spec):
+  def fetch_symlinks(self, spec) -> None:
     pass
-  def fetch_tarball(self, spec):
+  def fetch_tarball(self, spec) -> None:
     pass
-  def upload_symlinks_and_tarball(self, spec):
+  def upload_symlinks_and_tarball(self, spec) -> None:
     pass
 
 class PartialDownloadError(Exception):
@@ -47,7 +32,7 @@
 
 
 class HttpRemoteSync:
-  def __init__(self, remoteStore, architecture, workdir, insecure):
+  def __init__(self, remoteStore: str, architecture: str, workdir: str, insecure: bool):
     self.remoteStore = remoteStore
     self.writeStore = ""
     self.architecture = architecture
@@ -486,14 +471,14 @@
   time.
   """
 
-  def __init__(self, remoteStore, writeStore, architecture, workdir):
+  def __init__(self, remoteStore, writeStore, architecture, workdir) -> None:
     self.remoteStore = re.sub("^b3://", "", remoteStore)
     self.writeStore = re.sub("^b3://", "", writeStore)
     self.architecture = architecture
     self.workdir = workdir
     self._s3_init()
 
-  def _s3_init(self):
+  def _s3_init(self) -> None:
     # This is a separate method so that we can patch it out for unit tests.
     # Import boto3 here, so that if we don't use this remote store, we don't
     # have to install it in the first place.
@@ -530,7 +515,7 @@
       raise
     return True
 
-  def fetch_tarball(self, spec):
+  def fetch_tarball(self, spec) -> None:
     debug("Updating remote store for package %s with hashes %s", spec["package"],
           ", ".join(spec["remote_hashes"]))
 
@@ -568,7 +553,7 @@
     debug("Remote has no tarballs for %s with hashes %s", spec["package"],
           ", ".join(spec["remote_hashes"]))
 
-  def fetch_symlinks(self, spec):
+  def fetch_symlinks(self, spec) -> None:
     from botocore.exceptions import ClientError
     links_path = resolve_links_path(self.architecture, spec["package"])
     os.makedirs(os.path.join(self.workdir, links_path), exist_ok=True)
@@ -691,3 +676,21 @@
 
     self.s3.upload_file(Bucket=self.writeStore, Key=tar_path,
                         Filename=os.path.join(self.workdir, tar_path))
+
+RemoteSync = Union[HttpRemoteSync, S3RemoteSync, Boto3RemoteSync, CVMFSRemoteSync, RsyncRemoteSync, NoRemoteSync]
+
+def remote_from_url(read_url: str, write_url: str, architecture: str, work_dir: str, insecure=False) -> RemoteSync:
+  """Parse remote store URLs and return the correct RemoteSync instance for them."""
+  if read_url.startswith("http"):
+    return HttpRemoteSync(read_url, architecture, work_dir, insecure)
+  if read_url.startswith("s3://"):
+    return S3RemoteSync(read_url, write_url, architecture, work_dir)
+  if read_url.startswith("b3://"):
+    return Boto3RemoteSync(read_url, write_url, architecture, work_dir)
+  if read_url.startswith("cvmfs://"):
+    return CVMFSRemoteSync(read_url, None, architecture, work_dir)
+  if read_url:
+    return RsyncRemoteSync(read_url, write_url, architecture, work_dir)
+  return NoRemoteSync()
+
+

alibuild_helpers/templating_plugin.py

@@ -11,7 +11,7 @@
 from jinja2.sandbox import SandboxedEnvironment
 
 
-def build_plugin(specs, args, build_order):
+def build_plugin(specs, args, build_order) -> None:
     """Read a user-provided template from stdin and render it."""
     print(SandboxedEnvironment(autoescape=False)
           .from_string(sys.stdin.read())

alibuild_helpers/utilities.py

@@ -567,7 +567,7 @@ def getPackageList(packages, specs, configDir, preferSystem, noSystem,
 
 
 class Hasher:
-  def __init__(self):
+  def __init__(self) -> None:
     self.h = hashlib.sha1()
   def __call__(self, txt):
     if not type(txt) == bytes:

mypy.ini

@@ -0,0 +1,30 @@
+[mypy]
+python_version = 3.12
+platform = linux
+plugins = pydantic.mypy
+show_error_codes = true
+follow_imports = normal
+local_partial_types = true
+strict_equality = true
+no_implicit_optional = true
+warn_incomplete_stub = true
+warn_redundant_casts = true
+warn_unused_configs = true
+warn_unused_ignores = true
+enable_error_code = ignore-without-code, redundant-self, truthy-iterable
+disable_error_code = annotation-unchecked, import-not-found, import-untyped
+extra_checks = false
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
+[pydantic-mypy]
+init_forbid_extra = true
+init_typed = true
+warn_required_dynamic_aliases = true
+warn_untyped_fields = true

tests/test_analytics.py

@@ -7,11 +7,11 @@ def noAnalytics():
 def yesAnalytics():
     return True
 
-def notInvoked():
+def notInvoked() -> None:
     assert(False)
 
 class TestAnalytics(unittest.TestCase):
-    def test_analytics(self):
+    def test_analytics(self) -> None:
         self.assertEqual(False,  decideAnalytics(hasDisableFile=False,
                                                  hasUuid=False,
                                                  isTty=False,