libeufin

Integration and sandbox testing for FinTech APIs and data formats
Log | Files | Refs | Submodules | README | LICENSE

archive.py (5575B)


      1 #!/usr/bin/env python3
      2 
      3 # This file has been placed in the public domain.
      4 
      5 """Create a source archive containing this repository and its submodules."""
      6 
      7 from __future__ import annotations
      8 
      9 import argparse
     10 import io
     11 import os
     12 import subprocess
     13 import sys
     14 import tarfile
     15 from pathlib import Path
     16 
     17 
     18 def run_git(repo: Path, *args: str, input_data: bytes | None = None) -> bytes:
     19     try:
     20         return subprocess.run(
     21             ["git", *args],
     22             cwd=repo,
     23             input=input_data,
     24             check=True,
     25             stdout=subprocess.PIPE,
     26         ).stdout
     27     except subprocess.CalledProcessError as exc:
     28         command = " ".join(("git", *args))
     29         raise RuntimeError(f"'{command}' failed in {repo}") from exc
     30 
     31 
     32 def tracked_paths(repo: Path) -> list[Path]:
     33     output = run_git(repo, "ls-files", "-z", "--cached")
     34     return [
     35         Path(os.fsdecode(item)) for item in output.rstrip(b"\0").split(b"\0") if item
     36     ]
     37 
     38 
     39 def submodule_paths(repo: Path) -> set[Path]:
     40     gitmodules = repo / ".gitmodules"
     41     if not gitmodules.exists():
     42         return set()
     43     result = subprocess.run(
     44         [
     45             "git",
     46             "config",
     47             "-f",
     48             ".gitmodules",
     49             "--get-regexp",
     50             r"^submodule\..*\.path$",
     51         ],
     52         cwd=repo,
     53         check=False,
     54         stdout=subprocess.PIPE,
     55         text=True,
     56     )
     57     if result.returncode not in (0, 1):
     58         raise RuntimeError(f"could not read submodules from {gitmodules}")
     59     return {Path(line.split(maxsplit=1)[1]) for line in result.stdout.splitlines()}
     60 
     61 
     62 def export_ignored(repo: Path, paths: list[Path]) -> set[Path]:
     63     if not paths:
     64         return set()
     65     query = b"\0".join(os.fsencode(path) for path in paths) + b"\0"
     66     output = run_git(
     67         repo, "check-attr", "-z", "--stdin", "export-ignore", input_data=query
     68     )
     69     fields = output.rstrip(b"\0").split(b"\0")
     70     ignored: set[Path] = set()
     71     for index in range(0, len(fields), 3):
     72         if fields[index + 2] == b"set":
     73             ignored.add(Path(os.fsdecode(fields[index])))
     74     return ignored
     75 
     76 
     77 def repository_entries(repo: Path) -> list[tuple[Path, Path]]:
     78     """Return (filesystem path, archive-relative path) pairs for one repository."""
     79     submodules = submodule_paths(repo)
     80     candidates: list[tuple[Path, Path]] = []
     81 
     82     for relative in tracked_paths(repo):
     83         if relative not in submodules:
     84             candidates.append((repo / relative, relative))
     85 
     86     for submodule in sorted(submodules):
     87         submodule_dir = repo / submodule
     88         if not (submodule_dir / ".git").exists():
     89             raise RuntimeError(
     90                 f"submodule '{submodule}' is not initialized; run './bootstrap' first"
     91             )
     92         for source, relative in repository_entries(submodule_dir):
     93             candidates.append((source, submodule / relative))
     94 
     95     ignored = export_ignored(repo, [relative for _, relative in candidates])
     96     return [
     97         (source, relative) for source, relative in candidates if relative not in ignored
     98     ]
     99 
    100 
    101 def archive_prefix(output: Path) -> Path:
    102     name = output.name
    103     for suffix in (".tar.gz", ".tgz"):
    104         if name.endswith(suffix):
    105             prefix = name[: -len(suffix)]
    106             if prefix:
    107                 return Path(prefix)
    108             break
    109     raise ValueError("output file must end in .tar.gz or .tgz")
    110 
    111 
    112 def parse_args() -> argparse.Namespace:
    113     parser = argparse.ArgumentParser(description=__doc__)
    114     parser.add_argument(
    115         "--include",
    116         action="append",
    117         default=[],
    118         metavar="FILE",
    119         help="include an untracked or export-ignored file (may be repeated)",
    120     )
    121     parser.add_argument("output", type=Path, help="output .tar.gz file")
    122     return parser.parse_args()
    123 
    124 
    125 def main() -> int:
    126     args = parse_args()
    127     root = Path(run_git(Path.cwd(), "rev-parse", "--show-toplevel").decode().strip())
    128     output = args.output.resolve()
    129     prefix = archive_prefix(output)
    130     entries = repository_entries(root)
    131     version = run_git(root, "describe", "--tags", "--always", "--abbrev=8").strip()
    132     if not version or any(character in version for character in (b"\n", b"\r", b"\0")):
    133         raise ValueError("program version must be a nonempty single line")
    134     version += b"\n"
    135 
    136     included: list[tuple[Path, Path]] = []
    137     for name in args.include:
    138         relative = Path(name)
    139         if relative.is_absolute() or ".." in relative.parts:
    140             raise ValueError(
    141                 f"included path must be relative to the repository: {name}"
    142             )
    143         source = (root / relative).absolute()
    144         if not source.is_file():
    145             raise FileNotFoundError(f"included file does not exist: {name}")
    146         included.append((source, relative))
    147 
    148     with tarfile.open(output, "w:gz") as archive:
    149         stamp = tarfile.TarInfo(str(prefix / ".version"))
    150         stamp.size = len(version)
    151         stamp.mode = 0o644
    152         stamp.mtime = int(run_git(root, "log", "-1", "--format=%ct"))
    153         archive.addfile(stamp, io.BytesIO(version))
    154         for source, relative in [*included, *entries]:
    155             # A stale stamp in the checkout must not override the Git version.
    156             if relative == Path(".version"):
    157                 continue
    158             archive.add(source, arcname=prefix / relative, recursive=False)
    159 
    160     print(f"created {output}")
    161     return 0
    162 
    163 
    164 if __name__ == "__main__":
    165     try:
    166         sys.exit(main())
    167     except (OSError, RuntimeError, ValueError) as exc:
    168         print(f"archive: {exc}", file=sys.stderr)
    169         sys.exit(1)