scripts/update-cargo-wraps.py

#!/usr/bin/env python3

# SPDX-License-Identifier: GPL-2.0-or-later

"""
update-cargo-wraps.py - Update Meson subprojects from a Cargo
registry or from the versions pinned in Cargo.lock.
"""

# Copyright (C) 2025 Red Hat, Inc.
#
# Author: Paolo Bonzini <pbonzini@redhat.com>

import argparse
import configparser
import filecmp
import glob
import os
import shutil
import subprocess
import sys
import tomllib


def get_name_and_semver(namever: str) -> tuple[str, str]:
    """Split a subproject name into its name and semantic version parts"""
    parts = namever.rsplit("-", 1)
    if len(parts) != 2:
        return namever, ""

    return parts[0], parts[1]


class CrateSource:
    """Class for locating crate versions and pointing wrap files at them.
       Subclasses know where the source of a crate comes from and how to
       rewrite the ``[wrap-file]`` to consume it."""

    origin: str

    def find(self, namever: str) -> str | None:
        """Resolve a 'name-semver' prefix to a concrete 'name-version'."""
        raise NotImplementedError

    def rewrite_source(self, section: configparser.SectionProxy, orig_namever: str, source_namever: str) -> bool:
        """Update the download-related keys of a [wrap-file] section."""
        raise NotImplementedError


class CargoRegistry(CrateSource):
    """Locate crates already extracted in a local Cargo registry directory."""

    origin = "the Cargo registry"

    def __init__(self, path: str):
        self.path = path

    def find(self, namever: str) -> str | None:
        """Find installed crate matching name and semver prefix"""
        name, semver = get_name_and_semver(namever)

        # exact version match
        path = os.path.join(self.path, f"{name}-{semver}")
        if os.path.exists(path):
            return f"{name}-{semver}"

        # semver match
        matches = sorted(glob.glob(f"{path}.*"))
        return os.path.basename(matches[0]) if matches else None

    def rewrite_source(self, section: configparser.SectionProxy, orig_namever: str, registry_namever: str) -> bool:
        # the registry already holds the extracted sources, so Meson does not
        # download anything: drop the source_* keys.
        for key in list(section.keys()):
            if key.startswith("source"):
                del section[key]
        return True


class CargoLock(CrateSource):
    """Locate crates by the versions pinned in a Cargo.lock file."""

    origin = "crates.io"

    def __init__(self, path: str):
        with open(path, "rb") as f:
            data = tomllib.load(f)

        self.versions: dict[str, list[str]] = {}
        self.checksums: dict[str, str] = {}
        for pkg in data.get("package", []):
            # workspace members have neither a checksum nor a crates.io tarball
            if "checksum" not in pkg:
                continue
            name, version = pkg["name"], pkg["version"]
            self.versions.setdefault(name, []).append(version)
            self.checksums[f"{name}-{version}"] = pkg["checksum"]

    def find(self, namever: str) -> str | None:
        """Find pinned crate matching name and semver prefix"""
        name, semver = get_name_and_semver(namever)
        versions = sorted(self.versions.get(name, []))

        # exact version match
        if semver in versions:
            return f"{name}-{semver}"

        # semver match
        matches = [v for v in versions if v.startswith(f"{semver}.")]
        return f"{name}-{matches[0]}" if matches else None

    def rewrite_source(self, section: configparser.SectionProxy, orig_namever: str, new_namever: str) -> bool:
        # rewrite the download keys to fetch the pinned version from crates.io.
        if orig_namever == new_namever:
            return False
        name, version = get_name_and_semver(new_namever)
        section["source_url"] = f"https://crates.io/api/v1/crates/{name}/{version}/download"
        section["source_filename"] = f"{new_namever}.tar.gz"
        section["source_hash"] = self.checksums[new_namever]
        return True


class UpdateSubprojects:
    cargo_registry: str
    source: CrateSource
    top_srcdir: str
    dry_run: bool
    changes: int = 0

    def compare_build_rs(self, orig_dir: str, source_namever: str) -> None:
        """Warn if the build.rs in the original directory differs from the registry version."""
        orig_build_rs = os.path.join(orig_dir, "build.rs")
        new_build_rs = os.path.join(source_namever, "build.rs")

        msg = None
        if os.path.isfile(orig_build_rs) != os.path.isfile(new_build_rs):
            if os.path.isfile(orig_build_rs):
                msg = f"build.rs removed in {source_namever}"
            if os.path.isfile(new_build_rs):
                msg = f"build.rs added in {source_namever}"

        elif os.path.isfile(orig_build_rs) and not filecmp.cmp(orig_build_rs, new_build_rs, shallow=False):
            msg = f"build.rs changed from {orig_dir} to {source_namever}"
            # diff exits non-zero when the files differ, which is expected here
            subprocess.run(["diff", "-u", orig_build_rs, new_build_rs])

        if msg:
            print(f"⚠️  Warning: {msg}")
            print("   This may affect the build process - please review the differences.")

    def update_subproject(self, wrap_file: str, source_namever: str) -> None:
        """Modify [wrap-file] section to use the crate resolved as `source_namever`."""
        assert wrap_file.endswith("-rs.wrap")
        wrap_name = wrap_file[:-5]

        env = os.environ.copy()
        if self.cargo_registry:
            env["MESON_PACKAGE_CACHE_DIR"] = self.cargo_registry

        config = configparser.ConfigParser()
        config.read(wrap_file)
        if "wrap-file" not in config:
            return

        section = config["wrap-file"]
        orig_dir = section["directory"]

        if self.dry_run:
            if orig_dir != source_namever:
                print(f"Will replace {orig_dir} with {source_namever}.")
            elif not os.path.exists(orig_dir) or self.cargo_registry:
                print(f"Will install {orig_dir} from {self.source.origin}.")
            else:
                print(f"Will update {orig_dir} from cache.")
                return
            self.changes += 1
            return

        section["directory"] = source_namever
        if self.source.rewrite_source(section, orig_dir, source_namever):
            with open(wrap_file, "w") as f:
                config.write(f)

        with open(wrap_file, "w") as f:
            config.write(f)

        if orig_dir != source_namever:
            print(f"👉 Replacing {orig_dir} with {source_namever}.")
        elif not os.path.exists(orig_dir) or self.cargo_registry:
            print(f"👉 Installing {orig_dir} from {self.source.origin}.")
        else:
            print(f"👉 Updating {orig_dir} from cache.")
            subprocess.run(
                ["meson", "subprojects", "update", "--reset", wrap_name],
                cwd=self.top_srcdir,
                env=env,
                check=True,
            )
            return

        subprocess.run(
            ["meson", "subprojects", "download", wrap_name],
            cwd=self.top_srcdir,
            env=env,
            check=True,
        )
        self.changes += 1

        if os.path.exists(orig_dir) and orig_dir != source_namever:
            self.compare_build_rs(orig_dir, source_namever)
            shutil.rmtree(orig_dir)

    @staticmethod
    def parse_cmdline() -> argparse.Namespace:
        parser = argparse.ArgumentParser(
            description="Replace Meson subprojects with packages in a Cargo registry"
        )
        parser.add_argument(
            "--cargo-lock",
            action='store_true',
            default=False,
            help="Update wraps from Cargo.lock",
        )
        parser.add_argument(
            "--cargo-registry",
            default=None,
            help="Path to Cargo registry (default: CARGO_REGISTRY env var)",
        )
        parser.add_argument(
            "--dry-run",
            action="store_true",
            default=False,
            help="Do not actually replace anything",
        )

        args = parser.parse_args()
        if args.cargo_registry and args.cargo_lock:
            print("error: --cargo-registry and --cargo-lock are incompatible")
            sys.exit(1)
        if not args.cargo_registry and not args.cargo_lock:
            args.cargo_registry = os.environ.get("CARGO_REGISTRY")
            if not args.cargo_registry:
                print("error: CARGO_REGISTRY environment variable not set and " +
                      "--cargo-registry or --cargo-lock not provided")
                sys.exit(1)

        return args

    def __init__(self, args: argparse.Namespace):
        self.cargo_registry = args.cargo_registry
        self.dry_run = args.dry_run
        self.top_srcdir = os.getcwd()
        if args.cargo_lock:
            self.source = CargoLock(os.path.join(self.top_srcdir, "Cargo.lock"))
        else:
            self.source = CargoRegistry(args.cargo_registry)

    def main(self) -> None:
        if not os.path.exists("subprojects"):
            print("'subprojects' directory not found, nothing to do.")
            return

        os.chdir("subprojects")
        for wrap_file in sorted(glob.glob("*-rs.wrap")):
            namever = wrap_file[:-8]  # Remove '-rs.wrap'

            source_namever = self.source.find(namever)
            if not source_namever:
                print(f"No crate found for {wrap_file}")
                continue

            self.update_subproject(wrap_file, source_namever)

        if self.changes:
            if self.dry_run:
                print("Rerun without --dry-run to apply changes.")
            else:
                print(f"✨ {self.changes} subproject(s) updated!")
        else:
            print("No changes.")


if __name__ == "__main__":
    args = UpdateSubprojects.parse_cmdline()
    UpdateSubprojects(args).main()