Improve AliBuild typing and linter strictness

aliBuild

@@ -111,8 +111,8 @@ 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

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 @@ def doBuild(args, parser):
       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 @@ def doBuild(args, parser):
        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/init.py

@@ -2,7 +2,7 @@
 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
@@ -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) -> 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():

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 TypeAlias
 
 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 @@ def __str__(self):
 
 
 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
@@ -233,7 +218,7 @@ def fetch_symlinks(self, spec):
               os.path.join(self.workdir, links_path, linkname))
 
   def upload_symlinks_and_tarball(self, spec):
-    pass
+    dieOnError(True, "HTTP backend does not support uploading directly")
 
 
 class RsyncRemoteSync:
@@ -691,3 +676,21 @@ def upload_symlinks_and_tarball(self, spec):
 
     self.s3.upload_file(Bucket=self.writeStore, Key=tar_path,
                         Filename=os.path.join(self.workdir, tar_path))
+
+RemoteSync: TypeAlias = 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())

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