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/cmd.py

@@ -8,7 +8,7 @@
 
 from alibuild_helpers.log import debug, warning, dieOnError
 
-def decode_with_fallback(data):
+def decode_with_fallback(data : 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 +23,7 @@ def decode_with_fallback(data):
     return str(data)
 
 
-def getoutput(command, timeout=None):
+def getoutput(command:str, timeout=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,7 +37,7 @@ def getoutput(command, timeout=None):
   return decode_with_fallback(stdout)
 
 
-def getstatusoutput(command, timeout=None):
+def getstatusoutput(command:str, timeout: 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:
@@ -54,9 +54,10 @@ def getstatusoutput(command, timeout=None):
   return proc.returncode, merged_output
 
 
-def execute(command, printer=debug, timeout=None):
+def execute(command: str | list[str], printer=debug, timeout: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) -> 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