#!/usr/bin/python3 """Perform a createrepo test run.""" from __future__ import annotations import dataclasses import functools import pathlib import subprocess # noqa: S404 import sys import tempfile import typing import click import createrepo_c # type: ignore[import-not-found] if typing.TYPE_CHECKING: from typing import Final @dataclasses.dataclass(frozen=True) class Config: """Runtime configuration for the repotest program.""" rpmtree: pathlib.Path repo: pathlib.Path repomd: pathlib.Path source: pathlib.Path spec: pathlib.Path tempd: pathlib.Path def build_tarball(cfg: Config) -> pathlib.Path: """Build the source tarball.""" archive: Final = cfg.tempd / (cfg.source.name + ".tar.gz") if archive.exists(): sys.exit(f"Something already created {archive}") try: subprocess.check_call( # noqa: S603 ["tar", "-caf", archive, "--", cfg.source.name], # noqa: S607 shell=False, cwd=cfg.source.parent, ) except subprocess.CalledProcessError as err: sys.exit(f"Could not create {archive}: {err}") if not archive.is_file(): sys.exit(f"The tar tool did not generate {archive}") return archive def build_rpmtree(cfg: Config, source: pathlib.Path) -> None: """Build the rpmbuild tree structure.""" cfg.rpmtree.mkdir(mode=0o755) for subdir in ("BUILD", "BUILDROOT", "SOURCES"): (cfg.rpmtree / subdir).mkdir(mode=0o755) dest: Final = cfg.rpmtree / "SOURCES" / source.name try: subprocess.check_call( # noqa: S603 ["install", "-m", "644", "--", source, dest], # noqa: S607 shell=False, ) except subprocess.CalledProcessError as err: sys.exit(f"Could not copy {source} to {dest}: {err}") def create_empty_repo(cfg: Config) -> None: """Create an empty repository.""" print(f"Creating an empty repository at {cfg.repo}") cfg.repo.mkdir(mode=0o755) subprocess.check_call(["createrepo_c", "--", cfg.repo], shell=False) # noqa: S603,S607 if not cfg.repomd.is_file(): sys.exit(f"createrepo_c did not create {cfg.repomd}") def parse_repo(cfg: Config) -> dict[str, pathlib.Path]: """Parse a repository using the Python bindings for libcreaterepo-c.""" repomd: Final = createrepo_c.Repomd(str(cfg.repomd)) primary: Final = next(rec for rec in repomd.records if rec.type == "primary") def add_package(res: dict[str, pathlib.Path], pkg: createrepo_c.Package) -> None: """Record information about a single package.""" path: Final = cfg.repo / pkg.location_href idx: Final = f"{pkg.name}:{pkg.arch}" if idx in res: sys.exit(f"Duplicate package {idx}: first {res[idx]} now {path}") if not path.is_file(): sys.exit(f"Not a file: {path}") res[idx] = path res: Final[dict[str, pathlib.Path]] = {} createrepo_c.xml_parse_primary( str(cfg.repo / primary.location_href), pkgcb=functools.partial( add_package, res, ), ) return res def copy_rpm_packages(cfg: Config) -> None: """Copy the packages built by rpmbuild to the repo.""" print("Copying the source RPM package") srcs: Final = sorted((cfg.rpmtree / "SRPMS").rglob("*")) if len(srcs) != 1: sys.exit(f"Expected a single source package, got {srcs}") if not srcs[0].is_file() or not srcs[0].name.endswith(".src.rpm"): sys.exit(f"Expected a single SRPM file, got {srcs}") srcdir: Final = cfg.repo / "Sources" srcdir.mkdir(mode=0o755) print(f"- {srcs[0]} -> {srcdir}") subprocess.check_call( # noqa: S603 ["install", "-m", "644", "--", srcs[0], srcdir / srcs[0].name], # noqa: S607 shell=False, ) print("Copying the binary RPM packages") rpms: Final = sorted(path for path in (cfg.rpmtree / "RPMS").rglob("*") if path.is_file()) if len(rpms) != 2: # noqa: PLR2004 sys.exit(f"Expected two RPM packages, got {rpms!r}") if not all(path.name.endswith(".rpm") for path in rpms): sys.exit(f"Expected two RPM packages, got {rpms!r}") architectures: Final = {path.suffixes[-2].split(".", 1)[1] for path in rpms} if len(architectures) != 2 or "noarch" not in architectures: # noqa: PLR2004 sys.exit(f"Unexpected RPM architectures: {architectures!r}") for src in rpms: dstdir = cfg.repo / src.suffixes[-2].split(".", 1)[1] dstdir.mkdir(mode=0o755) print(f"- {src} -> {dstdir}") subprocess.check_call( # noqa: S603 ["install", "-m", "644", "--", src, dstdir / src.name], # noqa: S607 shell=False, ) print("Running createrepo_c again") subprocess.check_call(["createrepo_c", "--", cfg.repo], shell=False) # noqa: S603,S607 @click.command(name="repotest") @click.option( "--source", required=True, type=click.Path( exists=True, file_okay=False, dir_okay=True, resolve_path=True, path_type=pathlib.Path, ), help="path to the source directory", ) @click.option( "--spec", required=True, type=click.Path( exists=True, file_okay=True, dir_okay=False, resolve_path=True, path_type=pathlib.Path, ), help="path to the spec file", ) def main(*, source: pathlib.Path, spec: pathlib.Path) -> None: """Build the tarball, output the full path to it.""" makefile: Final = source / "GNUmakefile" if not makefile.is_file(): sys.exit(f"Not a file: {makefile}") with tempfile.TemporaryDirectory(prefix="repotest.") as tempd_obj: tempd: Final = pathlib.Path(tempd_obj) print(f"Using temporary directory {tempd}") cfg: Final = Config( repo=tempd / "repo", repomd=tempd / "repo/repodata/repomd.xml", rpmtree=tempd / "rpmbuild", source=source, spec=spec, tempd=tempd, ) source_archive: Final = build_tarball(cfg) print(f"Source archive: {source_archive}") build_rpmtree(cfg, source_archive) print(f"RPM directory structure at {cfg.rpmtree}") subprocess.check_call(["find", "--", cfg.rpmtree, "-ls"], shell=False) # noqa: S603,S607 try: subprocess.check_call( # noqa: S603 [ # noqa: S607 "rpmbuild", "-ba", f"-D_topdir {cfg.rpmtree}", "--", cfg.spec, ], shell=False, ) except subprocess.CalledProcessError as err: sys.exit(f"rpmbuild failed: {err}") subprocess.check_call(["find", "--", cfg.rpmtree, "-ls"], shell=False) # noqa: S603,S607 create_empty_repo(cfg) subprocess.check_call(["find", "--", cfg.repo, "-ls"], shell=False) # noqa: S603,S607 empty_files: Final = parse_repo(cfg) if empty_files: sys.exit(f"Files found in an empty repo: {empty_files!r}") copy_rpm_packages(cfg) subprocess.check_call(["find", "--", cfg.repo, "-ls"], shell=False) # noqa: S603,S607 files: Final = parse_repo(cfg) print(repr(files)) known: Final = {"foo:src", "foo-common:noarch"} keys: Final = set(files.keys()) missing: Final = known - keys if missing: sys.exit(f"Expected at least {known!r}, got {keys!r}") remaining: Final = sorted(keys - known) if len(remaining) != 1: sys.exit(f"Expected a single different key, got {remaining!r}") if remaining[0].split(":", 1)[0] != "foo": sys.exit(f"Expected a arch:foo key, got {remaining!r}") print("Seems fine!") if __name__ == "__main__": main()