Fix more mypy errors and flake8 warnings

Author: dwoz

relenv/build/common.py

@@ -5,7 +5,6 @@
 """
 from __future__ import annotations
 
-import glob
 import fnmatch
 import hashlib
 import io
@@ -26,23 +25,26 @@
 from html.parser import HTMLParser
 from types import ModuleType
 from typing import (
-    TYPE_CHECKING,
     Any,
     Callable,
-    cast,
     Dict,
     IO,
-    Iterable,
     List,
     MutableMapping,
     Optional,
     Sequence,
     Tuple,
-    TypedDict,
     Union,
-    Protocol,
+    cast,
 )
 
+try:
+    from typing import TYPE_CHECKING
+except ImportError:  # pragma: no cover
+    TYPE_CHECKING = False
+
+from typing_extensions import Protocol, TypedDict
+
 if TYPE_CHECKING:
     from multiprocessing.synchronize import Event as SyncEvent
 else:
@@ -564,11 +566,13 @@ def handle_starttag(
 
 
 class Comparable(Protocol):
+    """Protocol capturing the comparison operations we rely on."""
+
     def __lt__(self, other: Any) -> bool:
-        ...
+        """Return True when self is ordered before *other*."""
 
     def __gt__(self, other: Any) -> bool:
-        ...
+        """Return True when self is ordered after *other*."""
 
 
 def check_files(
@@ -652,7 +656,9 @@ def __init__(
         self.url_tpl = url
         self.fallback_url_tpl = fallback_url
         self.signature_tpl = signature
-        self.destination = destination
+        self._destination: pathlib.Path = pathlib.Path()
+        if destination:
+            self._destination = pathlib.Path(destination)
         self.version = version
         self.checksum = checksum
         self.checkfunc = checkfunc
@@ -849,6 +855,8 @@ def check_version(self) -> bool:
 
 
 class Recipe(TypedDict):
+    """Typed description of a build recipe entry."""
+
     build_func: Callable[[MutableMapping[str, str], "Dirs", IO[str]], None]
     wait_on: List[str]
     download: Optional[Download]

relenv/common.py

@@ -6,12 +6,14 @@
 from __future__ import annotations
 
 import http.client
+import json
 import logging
 import os as _os
 import pathlib
 import platform
 import queue
 import selectors
+import shutil
 import subprocess as _subprocess
 import sys as _sys
 import tarfile
@@ -20,6 +22,8 @@
 import time
 from typing import IO, Any, BinaryIO, Iterable, Mapping, Optional, Union, cast
 
+from typing_extensions import Literal
+
 # Re-export frequently monkeypatched modules for type checking.
 os = _os
 subprocess = _subprocess
@@ -28,6 +32,8 @@
 # relenv package version
 __version__ = "0.21.2"
 
+log = logging.getLogger(__name__)
+
 MODULE_DIR = pathlib.Path(__file__).resolve().parent
 
 DEFAULT_PYTHON = "3.10.18"
@@ -38,6 +44,83 @@
 
 MACOS_DEVELOPMENT_TARGET = "10.15"
 
+TOOLCHAIN_CACHE_ENV = "RELENV_TOOLCHAIN_CACHE"
+_TOOLCHAIN_MANIFEST = ".toolchain-manifest.json"
+
+
+# 8 GiB archives are not unusual; stick to metadata to fingerprint them.
+def _archive_metadata(path: pathlib.Path) -> dict[str, Union[str, int]]:
+    stat = path.stat()
+    return {
+        "archive": str(path.resolve()),
+        "size": stat.st_size,
+        "mtime": stat.st_mtime_ns,
+    }
+
+
+def _toolchain_cache_root() -> Optional[pathlib.Path]:
+    override = os.environ.get(TOOLCHAIN_CACHE_ENV)
+    if override:
+        if override.strip().lower() == "none":
+            return None
+        return pathlib.Path(override).expanduser()
+    cache_home = os.environ.get("XDG_CACHE_HOME")
+    if cache_home:
+        base = pathlib.Path(cache_home)
+    else:
+        base = pathlib.Path.home() / ".cache"
+    return base / "relenv" / "toolchains"
+
+
+def _toolchain_manifest_path(toolchain_path: pathlib.Path) -> pathlib.Path:
+    return toolchain_path / _TOOLCHAIN_MANIFEST
+
+
+def _load_toolchain_manifest(path: pathlib.Path) -> Optional[Mapping[str, Any]]:
+    if not path.exists():
+        return None
+    try:
+        with path.open(encoding="utf-8") as handle:
+            data = json.load(handle)
+    except (OSError, json.JSONDecodeError):
+        return None
+    if not isinstance(data, dict):
+        return None
+    return data
+
+
+def _manifest_matches(manifest: Mapping[str, Any], metadata: Mapping[str, Any]) -> bool:
+    return (
+        manifest.get("archive") == metadata.get("archive")
+        and manifest.get("size") == metadata.get("size")
+        and manifest.get("mtime") == metadata.get("mtime")
+    )
+
+
+def _write_toolchain_manifest(
+    toolchain_path: pathlib.Path, metadata: Mapping[str, Any]
+) -> None:
+    manifest_path = _toolchain_manifest_path(toolchain_path)
+    try:
+        with manifest_path.open("w", encoding="utf-8") as handle:
+            json.dump(metadata, handle, indent=2, sort_keys=True)
+            handle.write("\n")
+    except OSError as exc:  # pragma: no cover - permissions edge cases
+        log.warning(
+            "Unable to persist toolchain manifest at %s: %s", manifest_path, exc
+        )
+
+
+def toolchain_root_dir() -> pathlib.Path:
+    """Return the root directory used for cached toolchains."""
+    if sys.platform != "linux":
+        return DATA_DIR
+    root = _toolchain_cache_root()
+    if root is None:
+        return DATA_DIR / "toolchain"
+    return root
+
+
 REQUEST_HEADERS = {"User-Agent": f"relenv {__version__}"}
 
 CHECK_HOSTS = (
@@ -69,8 +152,9 @@
 SHEBANG_TPL_LINUX = textwrap.dedent(
     """#!/bin/sh
 "true" ''''
+# shellcheck disable=SC2093
 "exec" "$(dirname "$(readlink -f "$0")"){}" "$0" "$@"
-'''
+' '''
 """
 )
 
@@ -90,8 +174,10 @@
 done
 PHYS_DIR=$(pwd -P)
 REALPATH=$PHYS_DIR/$TARGET_FILE
+# shellcheck disable=SC2093
 "exec" "$(dirname "$REALPATH")"{} "$REALPATH" "$@"
-'''"""
+' '''
+"""
 )
 
 if sys.platform == "linux":
@@ -100,9 +186,6 @@
     SHEBANG_TPL = SHEBANG_TPL_MACOS
 
 
-log = logging.getLogger(__name__)
-
-
 class RelenvException(Exception):
     """
     Base class for exeptions generated from relenv.
@@ -180,7 +263,7 @@ def __init__(self: "WorkDirs", root: Union[str, os.PathLike[str]]) -> None:
         self.root: pathlib.Path = pathlib.Path(root)
         self.data: pathlib.Path = DATA_DIR
         self.toolchain_config: pathlib.Path = work_dir("toolchain", self.root)
-        self.toolchain: pathlib.Path = work_dir("toolchain", DATA_DIR)
+        self.toolchain: pathlib.Path = toolchain_root_dir()
         self.build: pathlib.Path = work_dir("build", DATA_DIR)
         self.src: pathlib.Path = work_dir("src", DATA_DIR)
         self.logs: pathlib.Path = work_dir("logs", DATA_DIR)
@@ -251,16 +334,17 @@ def get_toolchain(
     del root  # Kept for backward compatibility; location driven by DATA_DIR
     os.makedirs(DATA_DIR, exist_ok=True)
     if sys.platform != "linux":
-        return DATA_DIR
+        return toolchain_root_dir()
 
-    toolchain_root = DATA_DIR / "toolchain"
+    toolchain_root = toolchain_root_dir()
     try:
         triplet = get_triplet(machine=arch)
     except TypeError:
         triplet = get_triplet()
     toolchain_path = toolchain_root / triplet
+    metadata: Optional[Mapping[str, Any]] = None
     if toolchain_path.exists():
-        return toolchain_path
+        metadata = _load_toolchain_manifest(_toolchain_manifest_path(toolchain_path))
 
     try:
         from importlib import import_module
@@ -275,7 +359,24 @@ def get_toolchain(
 
     toolchain_root.mkdir(parents=True, exist_ok=True)
     archive_path = pathlib.Path(archive_attr)
+    archive_meta = _archive_metadata(archive_path)
+
+    if (
+        toolchain_path.exists()
+        and metadata
+        and _manifest_matches(metadata, archive_meta)
+    ):
+        return toolchain_path
+
+    if toolchain_path.exists():
+        shutil.rmtree(toolchain_path)
+
     extract(str(toolchain_root), str(archive_path))
+    if not toolchain_path.exists():
+        raise RelenvException(
+            f"Toolchain archive {archive_path} did not produce {toolchain_path}"
+        )
+    _write_toolchain_manifest(toolchain_path, archive_meta)
     return toolchain_path
 
 
@@ -370,6 +471,8 @@ def extract_archive(
     archive_path = pathlib.Path(archive)
     archive_str = str(archive_path)
     to_path = pathlib.Path(to_dir)
+    TarReadMode = Literal["r:gz", "r:xz", "r:bz2", "r"]
+    read_type: TarReadMode = "r"
     if archive_str.endswith(".tgz"):
         log.debug("Found tgz archive")
         read_type = "r:gz"
@@ -384,9 +487,8 @@ def extract_archive(
         read_type = "r:bz2"
     else:
         log.warning("Found unknown archive type: %s", archive_path)
-        read_type = "r"
-    with tarfile.open(archive_path, read_type) as tar:
-        tar.extractall(to_path)
+    with tarfile.open(str(archive_path), mode=read_type) as tar:
+        tar.extractall(str(to_path))
 
 
 def get_download_location(url: str, dest: Union[str, os.PathLike[str]]) -> str:

relenv/create.py

@@ -15,7 +15,14 @@
 import tarfile
 from collections.abc import Iterator
 
-from .common import RelenvException, arches, archived_build, build_arch
+from .common import (
+    RelenvException,
+    arches,
+    archived_build,
+    build_arch,
+    format_shebang,
+    relative_interpreter,
+)
 
 
 @contextlib.contextmanager
@@ -135,6 +142,7 @@ def create(
         for f in fp:
             fp.extract(f, writeto)
     _sync_relenv_package(writeto, version)
+    _repair_script_shebangs(writeto, version)
 
 
 def _site_packages_dir(root: pathlib.Path, version: str) -> pathlib.Path:
@@ -166,6 +174,71 @@ def _sync_relenv_package(root: pathlib.Path, version: str) -> None:
     )
 
 
+def _repair_script_shebangs(root: pathlib.Path, version: str) -> None:
+    """
+    Update legacy shell-wrapped entry points to the current shebang format.
+
+    Older archives shipped scripts that started with the ``"true" ''''`` preamble.
+    Those files break when executed directly under Python (the parser sees the
+    unmatched triple-quoted literal). Patch any remaining copies to the new
+    `format_shebang` layout so fresh installs do not inherit stale loaders.
+    """
+    if sys.platform == "win32":
+        return
+
+    scripts_dir = root / "bin"
+    if not scripts_dir.is_dir():
+        return
+
+    major_minor = ".".join(version.split(".")[:2])
+    interpreter_candidates = [
+        scripts_dir / f"python{major_minor}",
+        scripts_dir / f"python{major_minor.split('.')[0]}",
+        scripts_dir / "python3",
+        scripts_dir / "python",
+    ]
+    interpreter_path: pathlib.Path | None = None
+    for candidate in interpreter_candidates:
+        if candidate.exists():
+            interpreter_path = candidate
+            break
+    if interpreter_path is None:
+        return
+
+    try:
+        rel_interpreter = relative_interpreter(root, scripts_dir, interpreter_path)
+    except ValueError:
+        # Paths are not relative to the install root; abandon the rewrite.
+        return
+
+    try:
+        shebang = format_shebang(str(pathlib.PurePosixPath("/") / rel_interpreter))
+    except Exception:
+        return
+
+    legacy_prefix = "#!/bin/sh\n\"true\" ''''\n"
+    marker = "\n'''"
+    for script in scripts_dir.iterdir():
+        if not script.is_file():
+            continue
+        try:
+            text = script.read_text(encoding="utf-8")
+        except (OSError, UnicodeDecodeError):
+            continue
+        if not text.startswith(legacy_prefix):
+            continue
+        idx = text.find(marker)
+        if idx == -1:
+            continue
+        idy = idx + len(marker)
+        rest = text[idy:]
+        updated = shebang + rest.lstrip("\n")
+        try:
+            script.write_text(updated, encoding="utf-8")
+        except OSError:
+            continue
+
+
 def main(args: argparse.Namespace) -> None:
     """
     The entrypoint into the ``relenv create`` command.