diff --git a/scripts/create_release_branch.py b/scripts/create_release_branch.py index 0d3b6af9..b4579d9a 100644 --- a/scripts/create_release_branch.py +++ b/scripts/create_release_branch.py @@ -14,23 +14,19 @@ `git remote set-url` (with `add` fallback). High-level workflow: -1. Reuse (or populate) the cached TheRock clone; reclone only when the cache - is missing or corrupt (`--force-clone` deletes and reclones when the cache - directory exists but is not a valid git repo). Otherwise fetch/prune to - pick up new commits. -2. Hard-reset to the requested commit and populate submodules via - `fetch_sources.py` when available (fallback to `git submodule update`). -3. Build an execution plan from `.gitmodules` + `git submodule status`, - capturing repo URL, commit SHA, and working tree path for each component - plus TheRock itself. Repos listed in `--exclude-list` and repos outside - the ROCm GitHub org are filtered out. -4. For each component: - a. Set up the SSH `rocm-github` remote. - b. Check if the release branch already exists on the remote; if so, skip - the repo entirely (recorded as skipped, not a failure). - c. Create (or reset) the branch at the recorded commit. - d. Push to `rocm-github` (skipped in dry-run mode). -5. Log a summary of successful and failed repos. +1. Verify GitHub push/admin permissions for all repos (via GitHub API, no + clone required). +2. Reuse (or populate) the cached TheRock clone; reclone only when the cache + is missing or corrupt (`--force-clone` deletes and reclones). +3. Hard-reset to the requested commit and populate submodules via + `fetch_sources.py` when available (fallback to `git submodule update`). +4. Build an execution plan from `.gitmodules` + `git submodule status`. +5. For each component: + a. Set up the SSH `rocm-github` remote. + b. Check if the release branch already exists on the remote; skip if so. + c. Create (or reset) the branch at the recorded commit. + d. Push to `rocm-github` (skipped in dry-run mode). +6. Log a summary of successful, skipped, and failed repos. Dry-run mode (the default) logs every action without touching remotes; `--no-dry-run` enables actual pushes. @@ -54,32 +50,27 @@ import argparse import logging import re -import shlex -import shutil import subprocess import sys -import tempfile -from dataclasses import dataclass from pathlib import Path from pprint import pformat +from release_utils import ( + RockBase, + RepoInfo, + check_permissions, + fetch_lightweight_plan, + get_gh_token, +) -@dataclass -class RepoInfo: - """Information about a repository to branch.""" - url: str - commit: str - path: Path - - -class RockBranchingAutomation: +class RockBranchingAutomation(RockBase): """Automates creation of release branches for TheRock and its ROCm submodules.""" + _cache_dir_name = "rock-branching-cache" + def __init__(self, cli_args: argparse.Namespace) -> None: - self.release_branch: str = cli_args.branch_name - self.dry_run: bool = cli_args.dry_run - self.commitid: str = cli_args.commitid + super().__init__(cli_args) if not re.fullmatch(r"[0-9a-f]{40}", self.commitid): raise SystemExit( @@ -87,155 +78,13 @@ def __init__(self, cli_args: argparse.Namespace) -> None: f"SHA-1 hash, got: {self.commitid!r}" ) - self.exclude_list: set[str] = set(cli_args.exclude_list or []) - self.force_clone: bool = cli_args.force_clone - self.cache_dir: Path | None = ( - Path(cli_args.cache_dir) if cli_args.cache_dir else None - ) - self.rock_url: str = "https://github.com/ROCm/TheRock.git" - self.cache_root: Path | None = None - - self._logger = logging.getLogger("rock_branching") - self.log("Authentication Mode: SSH") self.log(f"Dry run mode = {self.dry_run}") if self.exclude_list: self.log(f"Exclude list: {self.exclude_list}") - def log(self, msg: str) -> None: - """Log an info-level message.""" - self._logger.info(msg) - - def run_command( - self, - args: list[str | Path], - cwd: Path, - *, - input_data: bytes | None = None, - stream: bool = False, - timeout: int | None = None, - ) -> None: - """Execute a subprocess command, raising CalledProcessError on failure. - - Args: - args: Command and arguments to execute. - cwd: Working directory for the command. - input_data: Optional bytes piped to stdin. - stream: If True, print stdout/stderr line-by-line as it arrives - (useful for long-running operations like clone/fetch). - If False, buffer output and log after completion. - timeout: Maximum seconds to wait before raising TimeoutExpired. - """ - cmd = args if isinstance(args, list) else [args] - self.log(f"++ Exec [{cwd}]$ {shlex.join(map(str, cmd))}") - sys.stdout.flush() - - if stream: - process = subprocess.Popen( - cmd, - cwd=str(cwd), - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - ) - - for line in process.stdout: - self.log(line.rstrip()) - - try: - ret = process.wait(timeout=timeout) - except subprocess.TimeoutExpired: - process.kill() - process.wait() - raise - if ret != 0: - raise subprocess.CalledProcessError(ret, cmd) - - return - - try: - result = subprocess.run( - cmd, - cwd=str(cwd), - shell=False, - input=input_data, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - check=True, - stdin=None if input_data else subprocess.DEVNULL, - text=False, - timeout=timeout, - ) - - if result.stdout: - self.log( - result.stdout - if isinstance(result.stdout, str) - else result.stdout.decode(errors="ignore") - ) - if result.stderr: - self.log( - result.stderr - if isinstance(result.stderr, str) - else result.stderr.decode(errors="ignore") - ) - - except subprocess.CalledProcessError as exc: - self.log( - (exc.stdout or b"").decode(errors="ignore") - if isinstance(exc.stdout, bytes) - else (exc.stdout or "") - ) - self.log( - (exc.stderr or b"").decode(errors="ignore") - if isinstance(exc.stderr, bytes) - else (exc.stderr or "") - ) - raise - - def run_command_output( - self, args: list[str | Path], cwd: Path, timeout: int | None = None - ) -> str: - """Run a command and return its stripped stdout as a string. - - Raises CalledProcessError on non-zero exit. - Raises subprocess.TimeoutExpired when *timeout* seconds elapse. - """ - cmd = args if isinstance(args, list) else [args] - self.log(f"++ Exec [{cwd}]$ {shlex.join(map(str, cmd))}") - - result = subprocess.run( - cmd, - cwd=str(cwd), - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - check=True, - stdin=subprocess.DEVNULL, - timeout=timeout, - ) - return result.stdout.strip() - - def _setup_remote(self, url: str, repo_dir: Path) -> None: - """Add or update the rocm-github remote for a repo.""" - remote_url = self.convert_to_ssh(url) - try: - self.run_command( - ["git", "remote", "set-url", "rocm-github", remote_url], - cwd=repo_dir, - ) - except subprocess.CalledProcessError: - self.run_command( - ["git", "remote", "add", "rocm-github", remote_url], - cwd=repo_dir, - ) - def _remote_branch_exists(self, repo_dir: Path) -> bool: - """Return True if the release branch already exists on rocm-github. - - Raises CalledProcessError if the remote check itself fails. - Raises subprocess.TimeoutExpired if the network call hangs (60 s). - """ + """Return True if the release branch already exists on rocm-github.""" output = self.run_command_output( ["git", "ls-remote", "--heads", "rocm-github", self.release_branch], cwd=repo_dir, @@ -254,8 +103,7 @@ def _push_branch(self, repo_name: str, repo_dir: Path) -> None: """Push the release branch to rocm-github, respecting dry-run mode.""" if self.dry_run: self.log( - f"[DRY RUN] Skipping push of {self.release_branch} " - f"for {repo_name}" + f"[DRY RUN] Skipping push of {self.release_branch} for {repo_name}" ) else: self.run_command( @@ -265,14 +113,7 @@ def _push_branch(self, repo_name: str, repo_dir: Path) -> None: ) def execute_plan(self, plan: dict[str, RepoInfo]) -> None: - """Execute the branching plan for every repo in *plan*. - - For each repo: - 1. Set up the ``rocm-github`` remote with the SSH URL. - 2. Guard against a pre-existing remote branch (recorded as skipped, not failed, if found). - 3. Create (or reset) the release branch at the recorded commit SHA. - 4. Push to ``rocm-github`` (skipped in dry-run mode). - """ + """Create and push release branches for every repo in the plan.""" successful_repos: dict[str, RepoInfo] = {} skipped_repos: dict[str, str] = {} failed_repos: dict[str, str] = {} @@ -281,9 +122,7 @@ def execute_plan(self, plan: dict[str, RepoInfo]) -> None: self.log(f"Processing {repo_name} at {info.path}") if not info.path.exists(): - failed_repos[repo_name] = ( - f"Repo path does not exist: {info.path}" - ) + failed_repos[repo_name] = f"Repo path does not exist: {info.path}" continue try: @@ -295,15 +134,12 @@ def execute_plan(self, plan: dict[str, RepoInfo]) -> None: try: branch_exists = self._remote_branch_exists(info.path) except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: - failed_repos[repo_name] = ( - f"Remote branch check failed: {exc}" - ) + failed_repos[repo_name] = f"Remote branch check failed: {exc}" continue if branch_exists: msg = ( - f"Remote branch {self.release_branch} already exists " - "on rocm-github" + f"Remote branch {self.release_branch} already exists on rocm-github" ) self.log(msg) skipped_repos[repo_name] = msg @@ -335,261 +171,26 @@ def execute_plan(self, plan: dict[str, RepoInfo]) -> None: if failed_repos: self.log(f"Failed repos: {pformat(failed_repos)}") - def convert_to_ssh(self, url: str) -> str: - """Convert https://github.com/X/Y.git to git@github.com:X/Y.git.""" - if url.startswith("https://github.com/"): - path = url.replace("https://github.com/", "") - return f"git@github.com:{path}" - return url - - def get_submodule_url_map(self, repo_dir: Path) -> dict[str, str]: - """Return mapping of submodule working-tree paths to remote URLs.""" - gitmodules_path = repo_dir / ".gitmodules" - if not gitmodules_path.exists(): - return {} - - try: - path_entries = self.run_command_output( - [ - "git", - "config", - "--file", - str(gitmodules_path), - "--get-regexp", - r"submodule\..*\.path", - ], - cwd=repo_dir, - ) - except subprocess.CalledProcessError: - return {} - - url_map: dict[str, str] = {} - for line in path_entries.splitlines(): - # Each line looks like: - # "submodule.external/hipcc.path external/hipcc" - parts = line.strip().split(None, 1) - if len(parts) != 2: - continue - key, path_value = parts - section = key.rsplit(".", 1)[0] - try: - url = self.run_command_output( - [ - "git", - "config", - "--file", - str(gitmodules_path), - "--get", - f"{section}.url", - ], - cwd=repo_dir, - ) - except subprocess.CalledProcessError: - self.log(f"No URL entry for {section}; skipping") - continue - url_map[path_value.strip()] = url - - return url_map - - def build_plan(self) -> dict[str, RepoInfo]: - """Build the branching execution plan. - - 1. Clone (or reuse cached clone of) TheRock. - 2. Check out and hard-reset to ``self.commitid``. - 3. Populate submodules via ``fetch_sources.py`` (or ``git submodule update``). - 4. Read ``git submodule status`` and ``.gitmodules`` to collect each - submodule's commit SHA, remote URL, and local path. - 5. Return a dict keyed by repo name, including TheRock itself. - """ - cache_root = ( - self.cache_dir - or Path(tempfile.gettempdir()) / "rock-branching-cache" - ) - cache_root.mkdir(parents=True, exist_ok=True) - clone_dir = cache_root / "TheRock" - self.cache_root = cache_root - - needs_clone = not clone_dir.exists() - if not needs_clone and not (clone_dir / ".git").exists(): - if not self.force_clone: - raise RuntimeError( - f"Cache directory {clone_dir} exists but is not a git " - "repo. Use --force-clone to delete it and reclone." - ) - self.log( - f"Cache directory {clone_dir} is not a git repo; " - "removing before reclone (--force-clone)" - ) - shutil.rmtree(clone_dir) - needs_clone = True - - if needs_clone: - self.log( - f"Cloning TheRock repo from {self.rock_url} into {clone_dir}" - ) - self.run_command( - ["git", "clone", str(self.rock_url), str(clone_dir)], - cwd=cache_root, - stream=True, - timeout=600, - ) - else: - self.log(f"Reusing existing TheRock repo at {clone_dir}") - - try: - remote_url = self.run_command_output( - ["git", "remote", "get-url", "origin"], - cwd=clone_dir, - ) - if "TheRock" not in remote_url: - raise RuntimeError( - f"Existing repo at {clone_dir} does not look like " - f"TheRock (origin={remote_url})" - ) - except subprocess.CalledProcessError as exc: - raise RuntimeError( - f"Failed to inspect existing repo at {clone_dir}: {exc}" - ) from exc - - self.log("Fetching latest changes for existing TheRock clone...") - self.run_command( - [ - "git", - "fetch", - "origin", - "--prune", - "--recurse-submodules=on-demand", - ], - cwd=clone_dir, - stream=True, - timeout=600, - ) - - fetch_script = clone_dir / "build_tools" / "fetch_sources.py" - rock_commit = self.commitid - - self.log(f"Checking out TheRock at commit {rock_commit}") - self.run_command(["git", "checkout", rock_commit], cwd=clone_dir) - self.run_command( - ["git", "reset", "--hard", rock_commit], cwd=clone_dir - ) - - if fetch_script.exists(): - self.log( - "Updating submodules via fetch_sources.py " - "(jobs=10, no patches)..." - ) - self.run_command( - [ - "python3", - str(fetch_script), - "--jobs", - "10", - "--no-apply-patches", - ], - cwd=clone_dir, - stream=True, - ) - else: - self.log( - "fetch_sources.py not found; " - "falling back to git submodule update" - ) - self.run_command( - ["git", "submodule", "update", "--init", "--recursive"], - cwd=clone_dir, - stream=True, - ) - - self.log("Reading submodule status...") - try: - status_output = self.run_command_output( - ["git", "submodule", "status"], - cwd=clone_dir, - ) - lines = status_output.split("\n") if status_output else [] - except subprocess.CalledProcessError as exc: - raise RuntimeError( - f"Failed to read submodule status: {exc}" - ) from exc - - url_map = self.get_submodule_url_map(clone_dir) - - plan: dict[str, RepoInfo] = {} - - # Each line from `git submodule status` looks like: - # " ()" or "- " (not initialized) - for line in lines: - if not line: - continue - - parts = line.split() - if len(parts) < 2: - continue - sha = parts[0].lstrip("-+") - path = parts[1] - - repo_name = Path(path).name - repo_url = url_map.get(path) - - if not repo_url: - self.log( - f"No URL found for submodule {path} in .gitmodules" - ) - continue - - if repo_name in self.exclude_list: - self.log(f"Skipping {repo_name} (in exclude list)") - continue - - url_lower = repo_url.lower() - if ( - "github.com/rocm/" not in url_lower - and "github.com:rocm/" not in url_lower - ): - self.log( - f"Skipping {repo_name} " - f"(not a ROCm org repo: {repo_url})" - ) - continue - - plan[repo_name] = RepoInfo( - url=repo_url, - commit=sha, - path=clone_dir / path, - ) - - plan["TheRock"] = RepoInfo( - url=self.rock_url, - commit=rock_commit, - path=clone_dir, - ) - - return plan - def run(self) -> None: - """Build the execution plan and execute it.""" + """Check permissions, build the plan, and execute it.""" + token = get_gh_token() + repo_map = fetch_lightweight_plan(token, self.commitid, self.exclude_list) + check_permissions(token, repo_map, self._logger, action="branches") + plan = self.build_plan() self.log(f"Execution plan:\n{pformat(plan)}") self.execute_plan(plan) def main(argv: list[str]) -> int: - """Parse arguments and run the branching automation.""" - parser = argparse.ArgumentParser( - description="Rock Branching Automation Tool", - ) + parser = argparse.ArgumentParser(description="Rock Branching Automation Tool") parser.add_argument( - "-B", - "--branch-name", - required=True, + "-B", "--branch-name", required=True, help="Name of the release branch to create", ) parser.add_argument( - "-C", - "--commitid", - required=True, - help="Commit ID of TheRock to branch from", + "-C", "--commitid", required=True, + help="Commit SHA of TheRock to branch from", ) parser.add_argument( "--dry-run", @@ -598,29 +199,20 @@ def main(argv: list[str]) -> int: help="Log actions without pushing to remotes (default: enabled)", ) parser.add_argument( - "--exclude-list", - nargs="*", - default=[], - help="List of submodule repo names to exclude from branching", + "--exclude-list", nargs="*", default=[], + help="Submodule repo names to exclude from branching", ) parser.add_argument( - "--force-clone", - action="store_true", - default=False, - help="Delete and reclone if cache directory exists but is not a " - "valid git repo", + "--force-clone", action="store_true", default=False, + help="Delete and reclone if cache dir exists but is not a valid git repo", ) parser.add_argument( - "--cache-dir", - default=None, - help="Directory to cache the TheRock clone " - "(default: /tmp/rock-branching-cache)", + "--cache-dir", default=None, + help="Directory to cache the TheRock clone (default: /tmp/rock-branching-cache)", ) args = parser.parse_args(argv) - logging.basicConfig( - level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" - ) + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") try: RockBranchingAutomation(args).run() diff --git a/scripts/release_utils.py b/scripts/release_utils.py new file mode 100644 index 00000000..19737fc7 --- /dev/null +++ b/scripts/release_utils.py @@ -0,0 +1,616 @@ +#!/usr/bin/env python3 +# Copyright Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT +""" +Shared utilities for ROCm release automation scripts. + +Provides: + - RepoInfo dataclass describing a single repo in an execution plan + - RockBase base class with subprocess helpers, git helpers, and + build_plan() for cloning/reading TheRock submodules + - extract_owner_repo parse (owner, repo) from a GitHub HTTPS or SSH URL + - get_gh_token retrieve the active gh CLI token + - fetch_lightweight_plan read .gitmodules from the GitHub API (no clone) + - check_permissions verify push/admin access for every repo in a plan +""" +import base64 +import json +import logging +import re +import shlex +import shutil +import subprocess +import sys +import tempfile +import urllib.error +import urllib.request +from dataclasses import dataclass +from pathlib import Path +from pprint import pformat + + +ROCK_URL = "https://github.com/ROCm/TheRock.git" + +TIMEOUT_LONG = 1800 # clone, fetch, submodule update +TIMEOUT_SHORT = 60 # tag, push, git config reads + + +@dataclass +class RepoInfo: + """A single repository entry in an execution plan.""" + + url: str + commit: str + path: Path + + +# --------------------------------------------------------------------------- +# Standalone GitHub helpers (no class state required) +# --------------------------------------------------------------------------- + +def extract_owner_repo(url: str) -> tuple[str, str]: + """Return (owner, repo) from a GitHub HTTPS or SSH URL. + + Accepts: + https://github.com/ROCm/hip.git → ("ROCm", "hip") + git@github.com:ROCm/hip.git → ("ROCm", "hip") + + The .git suffix is optional. Raises ValueError for non-matching URLs. + """ + m = re.match(r"https://github\.com/([^/]+)/([^/]+?)(?:\.git)?$", url) + if m: + return m.group(1), m.group(2) + m = re.match(r"git@github\.com:([^/]+)/([^/]+?)(?:\.git)?$", url) + if m: + return m.group(1), m.group(2) + raise ValueError(f"Cannot extract owner/repo from URL: {url!r}") + + +def get_gh_token() -> str: + """Return the GitHub token from the active gh CLI session. + + Raises SystemExit with a clear message if gh is missing or not logged in. + """ + try: + result = subprocess.run( + ["gh", "auth", "token"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=True, + ) + token = result.stdout.strip() + except FileNotFoundError: + raise SystemExit( + "ERROR: gh CLI not found. Install it from https://cli.github.com " + "and run: gh auth login" + ) + except subprocess.CalledProcessError: + raise SystemExit( + "ERROR: Not authenticated with gh CLI. Run: gh auth login" + ) + if not token: + raise SystemExit( + "ERROR: gh auth token returned an empty token. Run: gh auth login" + ) + return token + + +def fetch_lightweight_plan( + token: str, + commitid: str, + exclude_list: set[str], +) -> dict[str, str]: + """Return a repo-name → URL map by reading .gitmodules from the GitHub API. + + Fetches GET /repos/ROCm/TheRock/contents/.gitmodules?ref= so the + caller can build a repo list without cloning anything locally. Filters out + repos outside the ROCm org and repos in exclude_list. TheRock itself is + included unless it appears in exclude_list. + """ + api_url = ( + f"https://api.github.com/repos/ROCm/TheRock/contents/.gitmodules" + f"?ref={commitid}" + ) + req = urllib.request.Request( + api_url, + headers={ + "Authorization": f"token {token}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + }, + ) + try: + with urllib.request.urlopen(req, timeout=15) as resp: + data = json.loads(resp.read()) + except urllib.error.HTTPError as exc: + raise SystemExit( + f"ERROR: Failed to fetch .gitmodules from GitHub API " + f"(HTTP {exc.code}). Check that commit {commitid!r} exists " + f"and the token has repo read access." + ) + except urllib.error.URLError as exc: + raise SystemExit( + f"ERROR: Network error fetching .gitmodules: {exc.reason}" + ) + + raw = base64.b64decode(data["content"]).decode() + + repo_map: dict[str, str] = {} + current_path: str | None = None + current_url: str | None = None + + def _flush(path: str | None, url: str | None) -> None: + if not path or not url: + return + repo_name = Path(path).name + url_lower = url.lower() + is_rocm = ( + "github.com/rocm/" in url_lower or "github.com:rocm/" in url_lower + ) + if is_rocm and repo_name not in exclude_list: + repo_map[repo_name] = url + + for line in raw.splitlines(): + line = line.strip() + if line.startswith("[submodule"): + _flush(current_path, current_url) + current_path = None + current_url = None + elif "=" in line: + key, _, value = line.partition("=") + key = key.strip() + value = value.strip() + if key == "path": + current_path = value + elif key == "url": + current_url = value + + _flush(current_path, current_url) + + if "TheRock" not in exclude_list: + repo_map["TheRock"] = ROCK_URL + return repo_map + + +def check_permissions( + token: str, + repo_map: dict[str, str], + logger: logging.Logger, + action: str = "branches", +) -> None: + """Verify push/admin access for every repo in repo_map. + + Checks GET /repos/{owner}/{repo} for each entry. Collects all failures + before raising so the caller sees the complete list in one run. + Raises SystemExit if any repo lacks the required access. + + Args: + token: GitHub personal access token. + repo_map: Mapping of repo name → GitHub URL. + logger: Logger to use for progress output. + action: Short noun used in the abort message ("branches" or "tags"). + """ + logger.info("=" * 60) + logger.info(" GitHub Permission Check") + logger.info(" Verifying push/admin access for %d repo(s)", len(repo_map)) + logger.info("=" * 60) + + failed: dict[str, str] = {} + for repo_name, url in repo_map.items(): + try: + owner, repo = extract_owner_repo(url) + except ValueError as exc: + failed[repo_name] = f"URL parse error: {exc}" + continue + + api_url = f"https://api.github.com/repos/{owner}/{repo}" + req = urllib.request.Request( + api_url, + headers={ + "Authorization": f"token {token}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + }, + ) + try: + with urllib.request.urlopen(req, timeout=15) as resp: + data = json.loads(resp.read()) + except urllib.error.HTTPError as exc: + if exc.code == 403: + failed[repo_name] = ( + f"HTTP 403 Forbidden — token lacks access to " + f"{owner}/{repo}" + ) + elif exc.code == 404: + failed[repo_name] = ( + f"HTTP 404 — repo {owner}/{repo} not found " + "(token may lack visibility)" + ) + else: + failed[repo_name] = ( + f"HTTP {exc.code} from GitHub API for {owner}/{repo}" + ) + continue + except urllib.error.URLError as exc: + failed[repo_name] = ( + f"Network error checking {owner}/{repo}: {exc.reason}" + ) + continue + + perms = data.get("permissions", {}) + if perms.get("push", False) or perms.get("admin", False): + logger.info("Permission check OK: %s/%s", owner, repo) + else: + failed[repo_name] = ( + f"Insufficient permissions for {owner}/{repo}: " + f"push={perms.get('push')}, admin={perms.get('admin')}" + ) + + total = len(repo_map) + passed = total - len(failed) + logger.info("=" * 60) + logger.info(" Permission Check Summary") + logger.info(" Passed : %d / %d repo(s)", passed, total) + logger.info(" Failed : %d / %d repo(s)", len(failed), total) + logger.info("=" * 60) + + if failed: + lines = [f" {name}: {reason}" for name, reason in failed.items()] + raise SystemExit( + f"ERROR: Permission check failed for {len(failed)} repo(s). " + f"Aborting before any {action} are created.\n" + "\n".join(lines) + ) + + +# --------------------------------------------------------------------------- +# Base class +# --------------------------------------------------------------------------- + +class RockBase: + """Shared subprocess helpers, git helpers, and plan builder. + + Subclasses set ``_cache_dir_name`` (used as the default cache directory + suffix under /tmp) and call ``super().__init__()`` after setting their + own attributes. + """ + + _cache_dir_name: str = "rock-cache" + + def __init__(self, cli_args) -> None: + self.release_branch: str = cli_args.branch_name + self.dry_run: bool = cli_args.dry_run + self.commitid: str = cli_args.commitid + # Accept both space-separated (--exclude-list a b c) and + # comma-separated (--exclude-list a,b,c) values. + self.exclude_list: set[str] = { + item.strip() + for val in (cli_args.exclude_list or []) + for item in val.split(",") + if item.strip() + } + self.force_clone: bool = cli_args.force_clone + self.cache_dir: Path | None = ( + Path(cli_args.cache_dir) if cli_args.cache_dir else None + ) + self.rock_url: str = ROCK_URL + self.cache_root: Path | None = None + self._logger = logging.getLogger(self.__class__.__name__) + + # ------------------------------------------------------------------ + # Logging + # ------------------------------------------------------------------ + + def log(self, msg: str) -> None: + self._logger.info(msg) + + # ------------------------------------------------------------------ + # Subprocess helpers + # ------------------------------------------------------------------ + + def run_command( + self, + args: list[str | Path], + cwd: Path, + *, + input_data: bytes | None = None, + stream: bool = False, + timeout: int | None = None, + ) -> None: + """Execute a command, raising CalledProcessError on failure.""" + cmd = args if isinstance(args, list) else [args] + self.log(f"++ Exec [{cwd}]$ {shlex.join(map(str, cmd))}") + sys.stdout.flush() + + if stream: + process = subprocess.Popen( + cmd, + cwd=str(cwd), + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + try: + for line in process.stdout: + self.log(line.rstrip()) + ret = process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + raise subprocess.TimeoutExpired(cmd, timeout) + if ret != 0: + raise subprocess.CalledProcessError(ret, cmd) + return + + try: + result = subprocess.run( + cmd, + cwd=str(cwd), + shell=False, + input=input_data, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True, + stdin=None if input_data else subprocess.DEVNULL, + text=False, + timeout=timeout, + ) + if result.stdout: + self.log( + result.stdout + if isinstance(result.stdout, str) + else result.stdout.decode(errors="ignore") + ) + if result.stderr: + self.log( + result.stderr + if isinstance(result.stderr, str) + else result.stderr.decode(errors="ignore") + ) + except subprocess.CalledProcessError as exc: + self.log( + (exc.stdout or b"").decode(errors="ignore") + if isinstance(exc.stdout, bytes) + else (exc.stdout or "") + ) + self.log( + (exc.stderr or b"").decode(errors="ignore") + if isinstance(exc.stderr, bytes) + else (exc.stderr or "") + ) + raise + + def run_command_output( + self, + args: list[str | Path], + cwd: Path, + timeout: int | None = None, + ) -> str: + """Run a command and return its stripped stdout.""" + cmd = args if isinstance(args, list) else [args] + self.log(f"++ Exec [{cwd}]$ {shlex.join(map(str, cmd))}") + result = subprocess.run( + cmd, + cwd=str(cwd), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=True, + stdin=subprocess.DEVNULL, + timeout=timeout, + ) + return result.stdout.strip() + + # ------------------------------------------------------------------ + # Git helpers + # ------------------------------------------------------------------ + + def convert_to_ssh(self, url: str) -> str: + """Convert https://github.com/X/Y.git to git@github.com:X/Y.git.""" + if url.startswith("https://github.com/"): + return "git@github.com:" + url.replace("https://github.com/", "") + return url + + def _setup_remote(self, url: str, repo_dir: Path) -> None: + """Add or update the rocm-github remote.""" + remote_url = self.convert_to_ssh(url) + try: + self.run_command( + ["git", "remote", "set-url", "rocm-github", remote_url], + cwd=repo_dir, + ) + except subprocess.CalledProcessError: + self.run_command( + ["git", "remote", "add", "rocm-github", remote_url], + cwd=repo_dir, + ) + + def get_submodule_url_map(self, repo_dir: Path) -> dict[str, str]: + """Return mapping of submodule working-tree paths to remote URLs.""" + gitmodules_path = repo_dir / ".gitmodules" + if not gitmodules_path.exists(): + return {} + + try: + path_entries = self.run_command_output( + [ + "git", "config", + "--file", str(gitmodules_path), + "--get-regexp", r"submodule\..*\.path", + ], + cwd=repo_dir, + ) + except subprocess.CalledProcessError: + return {} + + url_map: dict[str, str] = {} + for line in path_entries.splitlines(): + parts = line.strip().split(None, 1) + if len(parts) != 2: + continue + key, path_value = parts + section = key.rsplit(".", 1)[0] + try: + url = self.run_command_output( + [ + "git", "config", + "--file", str(gitmodules_path), + "--get", f"{section}.url", + ], + cwd=repo_dir, + ) + except subprocess.CalledProcessError: + self.log(f"No URL entry for {section}; skipping") + continue + url_map[path_value.strip()] = url + + return url_map + + # ------------------------------------------------------------------ + # Plan builder + # ------------------------------------------------------------------ + + def _prepare_clone(self, cache_root: Path) -> Path: + """Ensure a valid TheRock clone exists under cache_root; return its path.""" + clone_dir = cache_root / "TheRock" + needs_clone = not clone_dir.exists() + + if not needs_clone and not (clone_dir / ".git").exists(): + if not self.force_clone: + raise RuntimeError( + f"Cache directory {clone_dir} exists but is not a git repo. " + "Use --force-clone to delete it and reclone." + ) + self.log( + f"Cache directory {clone_dir} is not a git repo; " + "removing before reclone (--force-clone)" + ) + shutil.rmtree(clone_dir) + needs_clone = True + + if needs_clone: + self.log(f"Cloning TheRock from {self.rock_url} into {clone_dir}") + self.run_command( + ["git", "clone", str(self.rock_url), str(clone_dir)], + cwd=cache_root, + stream=True, + timeout=TIMEOUT_LONG, + ) + else: + self.log(f"Reusing existing TheRock repo at {clone_dir}") + try: + remote_url = self.run_command_output( + ["git", "remote", "get-url", "origin"], + cwd=clone_dir, + ) + if "TheRock" not in remote_url: + raise RuntimeError( + f"Existing repo at {clone_dir} does not look like " + f"TheRock (origin={remote_url})" + ) + except subprocess.CalledProcessError as exc: + raise RuntimeError( + f"Failed to inspect existing repo at {clone_dir}: {exc}" + ) from exc + + self.log("Fetching latest changes for existing TheRock clone...") + self.run_command( + ["git", "fetch", "origin", "--prune", "--recurse-submodules=on-demand"], + cwd=clone_dir, + stream=True, + timeout=TIMEOUT_LONG, + ) + + return clone_dir + + def _checkout_and_update_submodules(self, clone_dir: Path) -> None: + """Hard-reset to self.commitid and populate submodules.""" + rock_commit = self.commitid + self.log(f"Checking out TheRock at commit {rock_commit}") + self.run_command(["git", "checkout", rock_commit], cwd=clone_dir) + self.run_command(["git", "reset", "--hard", rock_commit], cwd=clone_dir) + + fetch_script = clone_dir / "build_tools" / "fetch_sources.py" + if fetch_script.exists(): + self.log("Updating submodules via fetch_sources.py (jobs=10, no patches)...") + self.run_command( + ["python3", str(fetch_script), "--jobs", "10", "--no-apply-patches"], + cwd=clone_dir, + stream=True, + timeout=TIMEOUT_LONG, + ) + else: + self.log("fetch_sources.py not found; falling back to git submodule update") + self.run_command( + ["git", "submodule", "update", "--init", "--recursive"], + cwd=clone_dir, + stream=True, + timeout=TIMEOUT_LONG, + ) + + def build_plan(self) -> dict[str, RepoInfo]: + """Clone/reuse TheRock, populate submodules, and return the execution plan. + + Subclasses may call super().build_plan() and extend the result, or + override _prepare_clone / _checkout_and_update_submodules for extra + git steps (e.g. fetching a release branch before checkout). + """ + cache_root = ( + self.cache_dir + or Path(tempfile.gettempdir()) / self._cache_dir_name + ) + cache_root.mkdir(parents=True, exist_ok=True) + self.cache_root = cache_root + + clone_dir = self._prepare_clone(cache_root) + self._checkout_and_update_submodules(clone_dir) + + self.log("Reading submodule status...") + try: + status_output = self.run_command_output( + ["git", "submodule", "status"], + cwd=clone_dir, + ) + lines = status_output.split("\n") if status_output else [] + except subprocess.CalledProcessError as exc: + raise RuntimeError(f"Failed to read submodule status: {exc}") from exc + + url_map = self.get_submodule_url_map(clone_dir) + plan: dict[str, RepoInfo] = {} + + for line in lines: + if not line: + continue + parts = line.split() + if len(parts) < 2: + continue + sha = parts[0].lstrip("-+") + path = parts[1] + repo_name = Path(path).name + repo_url = url_map.get(path) + + if not repo_url: + self.log(f"No URL found for submodule {path} in .gitmodules") + continue + if repo_name in self.exclude_list: + self.log(f"Skipping {repo_name} (in exclude list)") + continue + url_lower = repo_url.lower() + if ( + "github.com/rocm/" not in url_lower + and "github.com:rocm/" not in url_lower + ): + self.log(f"Skipping {repo_name} (not a ROCm org repo: {repo_url})") + continue + + plan[repo_name] = RepoInfo(url=repo_url, commit=sha, path=clone_dir / path) + + if "TheRock" not in self.exclude_list: + plan["TheRock"] = RepoInfo( + url=self.rock_url, + commit=self.commitid, + path=clone_dir, + ) + else: + self.log("Skipping TheRock (in exclude list)") + return plan diff --git a/scripts/rock_tagging.py b/scripts/rock_tagging.py index 9a85a0cd..3461bc06 100644 --- a/scripts/rock_tagging.py +++ b/scripts/rock_tagging.py @@ -9,30 +9,28 @@ submodule. Authentication is done via SSH. High-level workflow: -1. Reuse (or populate) a cached clone under a configurable directory - (default: `/tmp/rock-tagging-cache`, overridable via `--cache-dir`), - fetch the latest refs, and hard-reset to the user-specified commit. -2. Update submodules via `fetch_sources.py` when available (fallback to - `git submodule update`) and build a plan by combining `.gitmodules` - metadata with `git submodule status` output. Repos listed in - `--exclude-list` and repos outside the ROCm GitHub org are skipped. -3. For each component (inside a single loop): - a. Configure an SSH `rocm-github` remote. - b. Create an annotated tag (`therock-`) at the recorded - commit, skipping components where the tag already exists. - c. For mono-repos (`rocm-libraries`, `rocm-systems`), generate - tarballs for the `projects/` and `shared/` directories (tarballs - are created even in dry-run mode). - d. When not in dry-run mode, push the tag and invoke - `gh release create` with the appropriate notes and tarball assets. - -Use `--force-clone` to delete and reclone when the cache directory exists -but is not a valid git repo. All actions are logged for traceability, and -dry-run mode (the default) lets you preview the plan without touching -remotes. +1. Verify GitHub push/admin permissions for all repos (via GitHub API, no + clone required). +2. Reuse (or populate) a cached clone under a configurable directory + (default: `/tmp/rock-tagging-cache`, overridable via `--cache-dir`), + fetch the latest refs, and hard-reset to the user-specified commit. +3. Update submodules via `fetch_sources.py` when available (fallback to + `git submodule update`) and build a plan by combining `.gitmodules` + metadata with `git submodule status` output. Repos listed in + `--exclude-list` and repos outside the ROCm GitHub org are skipped. +4. For each component: + a. Configure an SSH `rocm-github` remote. + b. Create an annotated tag (`therock-`) at the recorded + commit, skipping components where the tag already exists. + c. For mono-repos (`rocm-libraries`, `rocm-systems`), generate + tarballs for the `projects/` and `shared/` directories. + d. When not in dry-run mode, push the tag and invoke + `gh release create` with the appropriate notes and tarball assets. + +Dry-run mode (the default) lets you preview the plan without touching remotes. Usage: - python rock-tagging.py \\ + python rock_tagging.py \\ --branch-name \\ --release-version \\ --commitid \\ @@ -52,418 +50,57 @@ """ import argparse import logging -import shlex -import shutil import subprocess import sys import tarfile import tempfile -from dataclasses import dataclass from pathlib import Path from pprint import pformat +from release_utils import ( + RockBase, + RepoInfo, + TIMEOUT_LONG, + check_permissions, + fetch_lightweight_plan, + get_gh_token, +) -@dataclass -class RepoInfo: - """Information about a repository to tag.""" - url: str - commit: str - path: Path - - -class RockTagging: +class RockTagging(RockBase): """Automates tagging and release uploading for TheRock.""" - MONO_REPOS = frozenset({"rocm-libraries", "rocm-systems"}) + _cache_dir_name = "rock-tagging-cache" - # Timeouts (seconds) for subprocess calls. - TIMEOUT_LONG = 1800 # clone, fetch, submodule update - TIMEOUT_SHORT = 60 # tag, push, git config reads + MONO_REPOS = frozenset({"rocm-libraries", "rocm-systems"}) def __init__(self, cli_args: argparse.Namespace) -> None: - self.release_branch: str = cli_args.branch_name + super().__init__(cli_args) self.release_version: str = cli_args.release_version - self.dry_run: bool = cli_args.dry_run - self.commitid: str = cli_args.commitid - self.exclude_list: set[str] = set(cli_args.exclude_list or []) - self.force_clone: bool = cli_args.force_clone - self.cache_dir: Path | None = ( - Path(cli_args.cache_dir) if cli_args.cache_dir else None - ) - self.rock_url: str = "https://github.com/ROCm/TheRock.git" - self.cache_root: Path | None = None - - self._logger = logging.getLogger("rock_tagging") self.log("Authentication Mode: SSH") self.log(f"Dry run mode = {self.dry_run}") if self.exclude_list: self.log(f"Exclude list: {self.exclude_list}") - def log(self, msg: str) -> None: - """Log an info-level message.""" - self._logger.info(msg) - - def run_command( - self, - args: list[str | Path], - cwd: Path, - *, - input_data: bytes | None = None, - stream: bool = False, - timeout: int | None = TIMEOUT_SHORT, - ) -> None: - """Execute a subprocess command, raising CalledProcessError on failure. - - Args: - args: Command and arguments to execute. - cwd: Working directory for the command. - input_data: Optional bytes piped to stdin. - stream: If True, print stdout/stderr line-by-line as it arrives - (useful for long-running operations like clone/fetch). - If False, buffer output and log after completion. - timeout: Seconds before the process is killed (default: TIMEOUT_SHORT). - Pass TIMEOUT_LONG for clone/fetch/submodule operations. - """ - cmd = args if isinstance(args, list) else [args] - self.log(f"++ Exec [{cwd}]$ {shlex.join(map(str, cmd))}") - sys.stdout.flush() - - if stream: - process = subprocess.Popen( - cmd, - cwd=str(cwd), - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - ) - - try: - for line in process.stdout: - self.log(line.rstrip()) - ret = process.wait(timeout=timeout) - except subprocess.TimeoutExpired: - process.kill() - process.wait() - raise subprocess.TimeoutExpired(cmd, timeout) - if ret != 0: - raise subprocess.CalledProcessError(ret, cmd) - - return - - try: - result = subprocess.run( - cmd, - cwd=str(cwd), - shell=False, - input=input_data, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - check=True, - stdin=None if input_data else subprocess.DEVNULL, - text=False, - timeout=timeout, - ) - - if result.stdout: - self.log( - result.stdout - if isinstance(result.stdout, str) - else result.stdout.decode(errors="ignore") - ) - if result.stderr: - self.log( - result.stderr - if isinstance(result.stderr, str) - else result.stderr.decode(errors="ignore") - ) - - except subprocess.CalledProcessError as exc: - self.log( - (exc.stdout or b"").decode(errors="ignore") - if isinstance(exc.stdout, bytes) - else (exc.stdout or "") - ) - self.log( - (exc.stderr or b"").decode(errors="ignore") - if isinstance(exc.stderr, bytes) - else (exc.stderr or "") - ) - raise - - def run_command_output( - self, - args: list[str | Path], - cwd: Path, - timeout: int | None = TIMEOUT_SHORT, - ) -> str: - """Run a command and return its stripped stdout as a string. - - Raises CalledProcessError on non-zero exit. - """ - cmd = args if isinstance(args, list) else [args] - self.log(f"++ Exec [{cwd}]$ {shlex.join(map(str, cmd))}") - - result = subprocess.run( - cmd, - cwd=str(cwd), - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - check=True, - stdin=subprocess.DEVNULL, - timeout=timeout, - ) - return result.stdout.strip() - - def convert_to_ssh(self, url: str) -> str: - """Convert https://github.com/X/Y.git to git@github.com:X/Y.git.""" - if url.startswith("https://github.com/"): - path = url.replace("https://github.com/", "") - return f"git@github.com:{path}" - return url - - def get_submodule_url_map(self, repo_dir: Path) -> dict[str, str]: - """Return mapping of submodule working-tree paths to remote URLs.""" - gitmodules_path = repo_dir / ".gitmodules" - if not gitmodules_path.exists(): - return {} - - try: - path_entries = self.run_command_output( - [ - "git", - "config", - "--file", - str(gitmodules_path), - "--get-regexp", - r"submodule\..*\.path", - ], - cwd=repo_dir, - ) - except subprocess.CalledProcessError: - return {} - - url_map: dict[str, str] = {} - for line in path_entries.splitlines(): - # Each line looks like: - # "submodule.external/hipcc.path external/hipcc" - parts = line.strip().split(None, 1) - if len(parts) != 2: - continue - key, path_value = parts - section = key.rsplit(".", 1)[0] - try: - url = self.run_command_output( - [ - "git", - "config", - "--file", - str(gitmodules_path), - "--get", - f"{section}.url", - ], - cwd=repo_dir, - ) - except subprocess.CalledProcessError: - self.log(f"No URL entry for {section}; skipping") - continue - url_map[path_value.strip()] = url - - return url_map - - def build_plan(self) -> dict[str, RepoInfo]: - """Build the tagging execution plan. - - 1. Clone (or reuse cached clone of) TheRock. - 2. Check out and hard-reset to ``self.commitid``. - 3. Populate submodules via ``fetch_sources.py`` (or ``git submodule update``). - 4. Read ``git submodule status`` and ``.gitmodules`` to collect each - submodule's commit SHA, remote URL, and local path. - 5. Return a dict keyed by repo name, including TheRock itself. - """ - cache_root = ( - self.cache_dir - or Path(tempfile.gettempdir()) / "rock-tagging-cache" - ) - cache_root.mkdir(parents=True, exist_ok=True) - clone_dir = cache_root / "TheRock" - self.cache_root = cache_root - - needs_clone = not clone_dir.exists() - if not needs_clone and not (clone_dir / ".git").exists(): - if not self.force_clone: - raise RuntimeError( - f"Cache directory {clone_dir} exists but is not a git " - "repo. Use --force-clone to delete it and reclone." - ) - self.log( - f"Cache directory {clone_dir} is not a git repo; " - "removing before reclone (--force-clone)" - ) - shutil.rmtree(clone_dir) - needs_clone = True - - if needs_clone: - self.log( - f"Cloning TheRock repo from {self.rock_url} into {clone_dir}" - ) - self.run_command( - ["git", "clone", str(self.rock_url), str(clone_dir)], - cwd=cache_root, - stream=True, - timeout=self.TIMEOUT_LONG, - ) - else: - self.log(f"Reusing existing TheRock repo at {clone_dir}") - - try: - remote_url = self.run_command_output( - ["git", "remote", "get-url", "origin"], - cwd=clone_dir, - ) - if "TheRock" not in remote_url: - raise RuntimeError( - f"Existing repo at {clone_dir} does not look like " - f"TheRock (origin={remote_url})" - ) - except subprocess.CalledProcessError as exc: - raise RuntimeError( - f"Failed to inspect existing repo at {clone_dir}: {exc}" - ) from exc - - self.log("Fetching latest changes for existing TheRock clone...") - self.run_command( - [ - "git", - "fetch", - "origin", - "--prune", - "--recurse-submodules=on-demand", - ], - cwd=clone_dir, - stream=True, - timeout=self.TIMEOUT_LONG, - ) - - fetch_script = clone_dir / "build_tools" / "fetch_sources.py" + def _checkout_and_update_submodules(self, clone_dir: Path) -> None: + """Fetch the release branch then delegate to the base checkout logic.""" rock_commit = self.commitid - self.log( f"Fetching release branch '{self.release_branch}' to ensure " f"commit {rock_commit} is reachable..." ) self.run_command( [ - "git", - "fetch", - "origin", - f"refs/heads/{self.release_branch}:refs/remotes/origin/{self.release_branch}", + "git", "fetch", "origin", + f"refs/heads/{self.release_branch}:" + f"refs/remotes/origin/{self.release_branch}", ], cwd=clone_dir, stream=True, - timeout=self.TIMEOUT_LONG, + timeout=TIMEOUT_LONG, ) - - self.log(f"Checking out TheRock at commit {rock_commit}") - self.run_command(["git", "checkout", rock_commit], cwd=clone_dir) - self.run_command( - ["git", "reset", "--hard", rock_commit], cwd=clone_dir - ) - - if fetch_script.exists(): - self.log( - "Updating submodules via fetch_sources.py " - "(jobs=10, no patches)..." - ) - self.run_command( - [ - "python3", - str(fetch_script), - "--jobs", - "10", - "--no-apply-patches", - ], - cwd=clone_dir, - stream=True, - timeout=self.TIMEOUT_LONG, - ) - else: - self.log( - "fetch_sources.py not found; " - "falling back to git submodule update" - ) - self.run_command( - ["git", "submodule", "update", "--init", "--recursive"], - cwd=clone_dir, - stream=True, - timeout=self.TIMEOUT_LONG, - ) - - self.log("Reading submodule status...") - try: - status_output = self.run_command_output( - ["git", "submodule", "status"], - cwd=clone_dir, - ) - lines = status_output.split("\n") if status_output else [] - except subprocess.CalledProcessError as exc: - raise RuntimeError( - f"Failed to read submodule status: {exc}" - ) from exc - - url_map = self.get_submodule_url_map(clone_dir) - - plan: dict[str, RepoInfo] = {} - - for line in lines: - if not line: - continue - - parts = line.split() - if len(parts) < 2: - continue - sha = parts[0].lstrip("-+") - path = parts[1] - - repo_name = Path(path).name - repo_url = url_map.get(path) - - if not repo_url: - self.log( - f"No URL found for submodule {path} in .gitmodules" - ) - continue - - if repo_name in self.exclude_list: - self.log(f"Skipping {repo_name} (in exclude list)") - continue - - url_lower = repo_url.lower() - if ( - "github.com/rocm/" not in url_lower - and "github.com:rocm/" not in url_lower - ): - self.log( - f"Skipping {repo_name} " - f"(not a ROCm org repo: {repo_url})" - ) - continue - - plan[repo_name] = RepoInfo( - url=repo_url, - commit=sha, - path=clone_dir / path, - ) - - plan["TheRock"] = RepoInfo( - url=self.rock_url, - commit=rock_commit, - path=clone_dir, - ) - - return plan + super()._checkout_and_update_submodules(clone_dir) def _create_tarballs( self, @@ -472,7 +109,7 @@ def _create_tarballs( tarball_paths: list[Path], label: str, ) -> None: - """Create per-subdirectory tarballs from *source_dir*.""" + """Create per-subdirectory tarballs from source_dir.""" if not source_dir.is_dir(): self.log(f"Source directory not found for {label}: {source_dir}") return @@ -481,39 +118,16 @@ def _create_tarballs( for entry in sorted(source_dir.iterdir()): if entry.name.startswith(".") or not entry.is_dir(): continue - tarball_path = root_dir / f"{entry.name}.tar.gz" if tarball_path in tarball_paths: continue - with tarfile.open(tarball_path, "w:gz") as tf: tf.add(str(entry), arcname=entry.name) tarball_paths.append(tarball_path) self.log(f"Tarball created: {tarball_path}") - def _setup_remote(self, url: str, repo_dir: Path) -> None: - """Add or update the rocm-github remote for a repo.""" - remote_url = self.convert_to_ssh(url) - try: - self.run_command( - ["git", "remote", "set-url", "rocm-github", remote_url], - cwd=repo_dir, - ) - except subprocess.CalledProcessError: - self.run_command( - ["git", "remote", "add", "rocm-github", remote_url], - cwd=repo_dir, - ) - def execute_plan(self, plan: dict[str, RepoInfo]) -> None: - """Execute the tagging plan for every repo in *plan*. - - For each repo: - 1. Configure the SSH ``rocm-github`` remote. - 2. Create an annotated tag at the recorded commit. - 3. For mono-repos, generate tarballs. - 4. Push the tag and create a GitHub release (skipped in dry-run mode). - """ + """Create and push tags, and publish GitHub releases, for every repo.""" successful_components: dict[str, RepoInfo] = {} failed_components: dict[str, str] = {} work_dir = self.cache_root or Path(tempfile.gettempdir()) @@ -529,7 +143,6 @@ def execute_plan(self, plan: dict[str, RepoInfo]) -> None: failed_components[comp] = f"Remote setup failed: {exc}" continue - # Skip if tag already exists locally tag_exists = subprocess.run( ["git", "rev-parse", "-q", "--verify", tag_name], cwd=info.path, @@ -538,41 +151,27 @@ def execute_plan(self, plan: dict[str, RepoInfo]) -> None: ).returncode == 0 if tag_exists: - self.log( - f"Tag {tag_name} already exists for {comp}; " - "skipping creation" - ) + self.log(f"Tag {tag_name} already exists for {comp}; skipping creation") successful_components[comp] = info continue try: self.run_command( [ - "git", - "tag", - "-a", - tag_name, - info.commit, - "-m", - f"therock release v{self.release_version}", + "git", "tag", "-a", tag_name, info.commit, + "-m", f"therock release v{self.release_version}", ], cwd=info.path, ) if not self.dry_run: self.run_command( - [ - "git", - "push", - "rocm-github", - f"{tag_name}:refs/tags/{tag_name}", - ], + ["git", "push", "rocm-github", f"{tag_name}:refs/tags/{tag_name}"], cwd=info.path, ) except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: failed_components[comp] = f"Tag failed: {exc}" continue - # Tarballs only for mono-repos tarballs: list[Path] = [] if comp in self.MONO_REPOS: self._create_tarballs( @@ -586,21 +185,18 @@ def execute_plan(self, plan: dict[str, RepoInfo]) -> None: self.log(f"[DRY RUN] Would create release with: {tarballs}") else: try: - release_cmd = [ - "gh", - "release", - "create", - tag_name, - "--notes", - f"therock release v{self.release_version}", - *[str(p) for p in tarballs], - ] - self.run_command(release_cmd, cwd=info.path) - except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: - failed_components[comp] = ( - f"Release creation failed: {exc}" + self.run_command( + [ + "gh", "release", "create", tag_name, + "--notes", f"therock release v{self.release_version}", + *[str(p) for p in tarballs], + ], + cwd=info.path, ) + except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: + failed_components[comp] = f"Release creation failed: {exc}" continue + successful_components[comp] = info self.log( @@ -613,34 +209,29 @@ def execute_plan(self, plan: dict[str, RepoInfo]) -> None: self.log(f"Failed components: {pformat(failed_components)}") def run(self) -> None: - """Build the execution plan and execute it.""" + """Check permissions, build the plan, and execute it.""" + token = get_gh_token() + repo_map = fetch_lightweight_plan(token, self.commitid, self.exclude_list) + check_permissions(token, repo_map, self._logger, action="tags") + plan = self.build_plan() self.log(f"Execution plan: {pformat(plan)}") self.execute_plan(plan) def main(argv: list[str]) -> int: - """Parse arguments and run the tagging automation.""" - parser = argparse.ArgumentParser( - description="Rock Tagging Automation Tool", - ) + parser = argparse.ArgumentParser(description="Rock Tagging Automation Tool") parser.add_argument( - "-B", - "--branch-name", - required=True, + "-B", "--branch-name", required=True, help="Name of the release branch", ) parser.add_argument( - "-V", - "--release-version", - required=True, + "-V", "--release-version", required=True, help="Release version string (used for tag names)", ) parser.add_argument( - "-C", - "--commitid", - required=True, - help="Commit ID of TheRock to tag from", + "-C", "--commitid", required=True, + help="Commit SHA of TheRock to tag from", ) parser.add_argument( "--dry-run", @@ -649,29 +240,20 @@ def main(argv: list[str]) -> int: help="Log actions without pushing to remotes (default: enabled)", ) parser.add_argument( - "--exclude-list", - nargs="*", - default=[], - help="List of submodule repo names to exclude from tagging", + "--exclude-list", nargs="*", default=[], + help="Submodule repo names to exclude from tagging", ) parser.add_argument( - "--force-clone", - action="store_true", - default=False, - help="Delete and reclone if cache directory exists but is not a " - "valid git repo", + "--force-clone", action="store_true", default=False, + help="Delete and reclone if cache dir exists but is not a valid git repo", ) parser.add_argument( - "--cache-dir", - default=None, - help="Directory to cache the TheRock clone " - "(default: /tmp/rock-tagging-cache)", + "--cache-dir", default=None, + help="Directory to cache the TheRock clone (default: /tmp/rock-tagging-cache)", ) args = parser.parse_args(argv) - logging.basicConfig( - level=logging.INFO, format="[%(levelname)s] %(message)s" - ) + logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s") try: RockTagging(args).run() diff --git a/scripts/tests/test_create_release_branch.py b/scripts/tests/test_create_release_branch.py index 186e3db8..274c7e83 100644 --- a/scripts/tests/test_create_release_branch.py +++ b/scripts/tests/test_create_release_branch.py @@ -10,8 +10,11 @@ - get_submodule_url_map parsing - execute_plan behaviour with mocked subprocess calls """ +import json import subprocess import textwrap +import urllib.error +import urllib.request from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, call, patch @@ -20,7 +23,12 @@ import sys sys.path.insert(0, str(Path(__file__).parent.parent.parent)) -from scripts.create_release_branch import RepoInfo, RockBranchingAutomation +from scripts.create_release_branch import RockBranchingAutomation +from scripts.release_utils import ( + RepoInfo, + check_permissions, + extract_owner_repo, +) _FAKE_COMMIT = "a" * 40 @@ -235,6 +243,118 @@ def test_no_dry_run_calls_push(self, tmp_path): assert len(push_calls) == 1 +# --------------------------------------------------------------------------- +# _extract_owner_repo +# --------------------------------------------------------------------------- + +class TestExtractOwnerRepo: + @pytest.mark.parametrize("url,owner,repo", [ + ("https://github.com/ROCm/hip.git", "ROCm", "hip"), + ("https://github.com/ROCm/hip", "ROCm", "hip"), + ("git@github.com:ROCm/hip.git", "ROCm", "hip"), + ("git@github.com:ROCm/hip", "ROCm", "hip"), + ("https://github.com/ROCm/TheRock.git", "ROCm", "TheRock"), + ]) + def test_extraction(self, url, owner, repo): + assert extract_owner_repo(url) == (owner, repo) + + def test_invalid_url_raises_value_error(self): + with pytest.raises(ValueError): + extract_owner_repo("not-a-url") + + +# --------------------------------------------------------------------------- +# _check_permissions +# --------------------------------------------------------------------------- + +def _make_mock_response(permissions: dict) -> MagicMock: + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps({"permissions": permissions}).encode() + mock_resp.__enter__ = lambda s: s + mock_resp.__exit__ = MagicMock(return_value=False) + return mock_resp + + +def _repo_map() -> dict[str, str]: + return {"hip": "https://github.com/ROCm/hip.git"} + + +class TestCheckPermissions: + """check_permissions is now a standalone function in release_utils.""" + + def test_insufficient_permission_raises_with_repo_name(self): + mock_resp = _make_mock_response({"push": False, "admin": False}) + auto = make_automation() + with patch("urllib.request.urlopen", return_value=mock_resp): + with pytest.raises(SystemExit) as exc_info: + check_permissions("fake-token", _repo_map(), auto._logger) + assert "hip" in str(exc_info.value) + + def test_all_push_access_passes(self): + repo_map = { + "hip": "https://github.com/ROCm/hip.git", + "clr": "https://github.com/ROCm/clr.git", + } + mock_resp = _make_mock_response({"push": True, "admin": False}) + auto = make_automation() + with patch("urllib.request.urlopen", return_value=mock_resp): + check_permissions("fake-token", repo_map, auto._logger) # must not raise + + def test_admin_access_passes(self): + mock_resp = _make_mock_response({"push": False, "admin": True}) + auto = make_automation() + with patch("urllib.request.urlopen", return_value=mock_resp): + check_permissions("fake-token", _repo_map(), auto._logger) # must not raise + + def test_http_403_raises_with_repo_name(self): + auto = make_automation() + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.HTTPError( + url=None, code=403, msg="Forbidden", hdrs=None, fp=None + ), + ): + with pytest.raises(SystemExit) as exc_info: + check_permissions("fake-token", _repo_map(), auto._logger) + assert "hip" in str(exc_info.value) + + def test_http_404_raises_with_repo_name(self): + auto = make_automation() + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.HTTPError( + url=None, code=404, msg="Not Found", hdrs=None, fp=None + ), + ): + with pytest.raises(SystemExit) as exc_info: + check_permissions("fake-token", _repo_map(), auto._logger) + assert "hip" in str(exc_info.value) + + def test_multiple_failures_summary_lists_all(self): + repo_map = { + "hip": "https://github.com/ROCm/hip.git", + "clr": "https://github.com/ROCm/clr.git", + } + mock_resp = _make_mock_response({"push": False, "admin": False}) + auto = make_automation() + with patch("urllib.request.urlopen", return_value=mock_resp): + with pytest.raises(SystemExit) as exc_info: + check_permissions("fake-token", repo_map, auto._logger) + msg = str(exc_info.value) + assert "hip" in msg + assert "clr" in msg + + def test_network_error_recorded_as_failure(self): + auto = make_automation() + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.URLError("connection refused"), + ): + with pytest.raises(SystemExit) as exc_info: + check_permissions("fake-token", _repo_map(), auto._logger) + assert "hip" in str(exc_info.value) + + # --------------------------------------------------------------------------- # commitid validation # --------------------------------------------------------------------------- diff --git a/scripts/tests/test_release_utils.py b/scripts/tests/test_release_utils.py new file mode 100644 index 00000000..44f3e68a --- /dev/null +++ b/scripts/tests/test_release_utils.py @@ -0,0 +1,534 @@ +#!/usr/bin/env python3 +# Copyright Advanced Micro Devices, Inc. +# SPDX-License-Identifier: MIT +""" +Tests for release_utils.py. + +Covers: +- extract_owner_repo +- get_gh_token +- fetch_lightweight_plan +- check_permissions +- RockBase.convert_to_ssh +- RockBase.get_submodule_url_map +- RockBase.build_plan (clone / reuse / submodule parsing) +""" +import base64 +import json +import subprocess +import textwrap +import urllib.error +import urllib.request +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, call, patch + +import pytest + +import sys +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) +from scripts.release_utils import ( + RepoInfo, + RockBase, + check_permissions, + extract_owner_repo, + fetch_lightweight_plan, + get_gh_token, + ROCK_URL, + TIMEOUT_LONG, +) + + +_FAKE_COMMIT = "a" * 40 + + +def make_base(**kwargs) -> RockBase: + """Instantiate a bare RockBase (concrete enough for testing shared methods).""" + defaults = dict( + branch_name="release/6.4", + commitid=_FAKE_COMMIT, + dry_run=True, + exclude_list=[], + force_clone=False, + cache_dir=None, + ) + defaults.update(kwargs) + return RockBase(SimpleNamespace(**defaults)) + + +def _make_mock_response(payload: dict) -> MagicMock: + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps(payload).encode() + mock_resp.__enter__ = lambda s: s + mock_resp.__exit__ = MagicMock(return_value=False) + return mock_resp + + +# --------------------------------------------------------------------------- +# extract_owner_repo +# --------------------------------------------------------------------------- + +class TestExtractOwnerRepo: + @pytest.mark.parametrize("url,owner,repo", [ + ("https://github.com/ROCm/hip.git", "ROCm", "hip"), + ("https://github.com/ROCm/hip", "ROCm", "hip"), + ("git@github.com:ROCm/hip.git", "ROCm", "hip"), + ("git@github.com:ROCm/hip", "ROCm", "hip"), + ("https://github.com/ROCm/TheRock.git", "ROCm", "TheRock"), + ("https://github.com/llvm/llvm-project.git", "llvm", "llvm-project"), + ]) + def test_valid_urls(self, url, owner, repo): + assert extract_owner_repo(url) == (owner, repo) + + def test_invalid_url_raises_value_error(self): + with pytest.raises(ValueError, match="Cannot extract owner/repo"): + extract_owner_repo("not-a-url") + + def test_gitlab_url_raises_value_error(self): + with pytest.raises(ValueError): + extract_owner_repo("https://gitlab.com/ROCm/hip.git") + + def test_bare_hostname_raises_value_error(self): + with pytest.raises(ValueError): + extract_owner_repo("https://github.com/ROCm") + + +# --------------------------------------------------------------------------- +# get_gh_token +# --------------------------------------------------------------------------- + +class TestGetGhToken: + def test_returns_token_on_success(self): + mock_result = MagicMock() + mock_result.stdout = "ghp_faketoken\n" + with patch("subprocess.run", return_value=mock_result): + token = get_gh_token() + assert token == "ghp_faketoken" + + def test_gh_not_found_raises_system_exit(self): + with patch("subprocess.run", side_effect=FileNotFoundError): + with pytest.raises(SystemExit, match="gh CLI not found"): + get_gh_token() + + def test_not_authenticated_raises_system_exit(self): + with patch( + "subprocess.run", + side_effect=subprocess.CalledProcessError(1, "gh"), + ): + with pytest.raises(SystemExit, match="Not authenticated"): + get_gh_token() + + def test_empty_token_raises_system_exit(self): + mock_result = MagicMock() + mock_result.stdout = " " + with patch("subprocess.run", return_value=mock_result): + with pytest.raises(SystemExit, match="empty token"): + get_gh_token() + + +# --------------------------------------------------------------------------- +# fetch_lightweight_plan +# --------------------------------------------------------------------------- + +def _gitmodules_content(entries: list[tuple[str, str]]) -> str: + """Build a .gitmodules-style string from (path, url) pairs.""" + blocks = [] + for path, url in entries: + name = Path(path).name + blocks.append( + f'[submodule "{path}"]\n' + f"\tpath = {path}\n" + f"\turl = {url}\n" + ) + return "\n".join(blocks) + + +def _make_gitmodules_response(entries: list[tuple[str, str]]) -> MagicMock: + raw = _gitmodules_content(entries) + content_b64 = base64.b64encode(raw.encode()).decode() + return _make_mock_response({"content": content_b64}) + + +class TestFetchLightweightPlan: + def test_rocm_repos_included(self): + entries = [ + ("external/hip", "https://github.com/ROCm/hip.git"), + ("external/clr", "https://github.com/ROCm/clr.git"), + ] + mock_resp = _make_gitmodules_response(entries) + with patch("urllib.request.urlopen", return_value=mock_resp): + result = fetch_lightweight_plan("token", _FAKE_COMMIT, set()) + + assert "hip" in result + assert "clr" in result + assert result["hip"] == "https://github.com/ROCm/hip.git" + + def test_therock_always_included(self): + mock_resp = _make_gitmodules_response([]) + with patch("urllib.request.urlopen", return_value=mock_resp): + result = fetch_lightweight_plan("token", _FAKE_COMMIT, set()) + + assert "TheRock" in result + assert result["TheRock"] == ROCK_URL + + def test_non_rocm_repos_excluded(self): + entries = [ + ("external/llvm", "https://github.com/llvm/llvm-project.git"), + ("external/hip", "https://github.com/ROCm/hip.git"), + ] + mock_resp = _make_gitmodules_response(entries) + with patch("urllib.request.urlopen", return_value=mock_resp): + result = fetch_lightweight_plan("token", _FAKE_COMMIT, set()) + + assert "llvm-project" not in result + assert "hip" in result + + def test_exclude_list_respected(self): + entries = [ + ("external/hip", "https://github.com/ROCm/hip.git"), + ("external/clr", "https://github.com/ROCm/clr.git"), + ] + mock_resp = _make_gitmodules_response(entries) + with patch("urllib.request.urlopen", return_value=mock_resp): + result = fetch_lightweight_plan("token", _FAKE_COMMIT, {"hip"}) + + assert "hip" not in result + assert "clr" in result + + def test_http_error_raises_system_exit(self): + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.HTTPError( + url=None, code=404, msg="Not Found", hdrs=None, fp=None + ), + ): + with pytest.raises(SystemExit, match="HTTP 404"): + fetch_lightweight_plan("token", _FAKE_COMMIT, set()) + + def test_network_error_raises_system_exit(self): + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.URLError("name resolution failed"), + ): + with pytest.raises(SystemExit, match="Network error"): + fetch_lightweight_plan("token", _FAKE_COMMIT, set()) + + def test_ssh_urls_in_gitmodules_included(self): + entries = [ + ("external/hip", "git@github.com:ROCm/hip.git"), + ] + mock_resp = _make_gitmodules_response(entries) + with patch("urllib.request.urlopen", return_value=mock_resp): + result = fetch_lightweight_plan("token", _FAKE_COMMIT, set()) + + assert "hip" in result + + def test_therock_excluded_when_in_exclude_list(self): + mock_resp = _make_gitmodules_response([]) + with patch("urllib.request.urlopen", return_value=mock_resp): + result = fetch_lightweight_plan("token", _FAKE_COMMIT, {"TheRock"}) + + assert "TheRock" not in result + + +# --------------------------------------------------------------------------- +# check_permissions +# --------------------------------------------------------------------------- + +class TestCheckPermissions: + def _logger(self): + import logging + return logging.getLogger("test") + + def test_push_access_passes(self): + repo_map = {"hip": "https://github.com/ROCm/hip.git"} + mock_resp = _make_mock_response({"permissions": {"push": True, "admin": False}}) + with patch("urllib.request.urlopen", return_value=mock_resp): + check_permissions("token", repo_map, self._logger()) # must not raise + + def test_admin_access_passes(self): + repo_map = {"hip": "https://github.com/ROCm/hip.git"} + mock_resp = _make_mock_response({"permissions": {"push": False, "admin": True}}) + with patch("urllib.request.urlopen", return_value=mock_resp): + check_permissions("token", repo_map, self._logger()) # must not raise + + def test_no_access_raises_system_exit(self): + repo_map = {"hip": "https://github.com/ROCm/hip.git"} + mock_resp = _make_mock_response({"permissions": {"push": False, "admin": False}}) + with patch("urllib.request.urlopen", return_value=mock_resp): + with pytest.raises(SystemExit) as exc_info: + check_permissions("token", repo_map, self._logger()) + assert "hip" in str(exc_info.value) + + def test_http_403_recorded_as_failure(self): + repo_map = {"hip": "https://github.com/ROCm/hip.git"} + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.HTTPError( + url=None, code=403, msg="Forbidden", hdrs=None, fp=None + ), + ): + with pytest.raises(SystemExit) as exc_info: + check_permissions("token", repo_map, self._logger()) + assert "hip" in str(exc_info.value) + + def test_http_404_recorded_as_failure(self): + repo_map = {"hip": "https://github.com/ROCm/hip.git"} + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.HTTPError( + url=None, code=404, msg="Not Found", hdrs=None, fp=None + ), + ): + with pytest.raises(SystemExit) as exc_info: + check_permissions("token", repo_map, self._logger()) + assert "hip" in str(exc_info.value) + + def test_other_http_error_recorded_as_failure(self): + repo_map = {"hip": "https://github.com/ROCm/hip.git"} + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.HTTPError( + url=None, code=500, msg="Server Error", hdrs=None, fp=None + ), + ): + with pytest.raises(SystemExit) as exc_info: + check_permissions("token", repo_map, self._logger()) + assert "hip" in str(exc_info.value) + + def test_network_error_recorded_as_failure(self): + repo_map = {"hip": "https://github.com/ROCm/hip.git"} + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.URLError("connection refused"), + ): + with pytest.raises(SystemExit) as exc_info: + check_permissions("token", repo_map, self._logger()) + assert "hip" in str(exc_info.value) + + def test_all_repos_checked_before_abort(self): + """All repos are checked even if the first one fails.""" + repo_map = { + "hip": "https://github.com/ROCm/hip.git", + "clr": "https://github.com/ROCm/clr.git", + } + mock_resp = _make_mock_response({"permissions": {"push": False, "admin": False}}) + with patch("urllib.request.urlopen", return_value=mock_resp): + with pytest.raises(SystemExit) as exc_info: + check_permissions("token", repo_map, self._logger()) + msg = str(exc_info.value) + assert "hip" in msg + assert "clr" in msg + + def test_invalid_url_recorded_as_failure(self): + repo_map = {"bad": "not-a-url"} + with pytest.raises(SystemExit) as exc_info: + check_permissions("token", repo_map, self._logger()) + assert "bad" in str(exc_info.value) + + def test_action_label_appears_in_abort_message(self): + repo_map = {"hip": "https://github.com/ROCm/hip.git"} + mock_resp = _make_mock_response({"permissions": {"push": False, "admin": False}}) + with patch("urllib.request.urlopen", return_value=mock_resp): + with pytest.raises(SystemExit) as exc_info: + check_permissions("token", repo_map, self._logger(), action="tags") + assert "tags" in str(exc_info.value) + + +# --------------------------------------------------------------------------- +# RockBase.convert_to_ssh +# --------------------------------------------------------------------------- + +class TestConvertToSsh: + def test_https_converted(self): + base = make_base() + assert base.convert_to_ssh("https://github.com/ROCm/hip.git") == \ + "git@github.com:ROCm/hip.git" + + def test_https_without_dot_git(self): + base = make_base() + assert base.convert_to_ssh("https://github.com/ROCm/clr") == \ + "git@github.com:ROCm/clr" + + def test_ssh_passthrough(self): + base = make_base() + url = "git@github.com:ROCm/hip.git" + assert base.convert_to_ssh(url) == url + + def test_non_github_passthrough(self): + base = make_base() + url = "https://gitlab.com/org/repo.git" + assert base.convert_to_ssh(url) == url + + +# --------------------------------------------------------------------------- +# RockBase.get_submodule_url_map +# --------------------------------------------------------------------------- + +class TestGetSubmoduleUrlMap: + def test_no_gitmodules_returns_empty(self, tmp_path): + base = make_base() + assert base.get_submodule_url_map(tmp_path) == {} + + def test_parses_paths_and_urls(self, tmp_path): + (tmp_path / ".gitmodules").write_text(textwrap.dedent("""\ + [submodule "external/hip"] + path = external/hip + url = https://github.com/ROCm/hip.git + [submodule "external/clr"] + path = external/clr + url = https://github.com/ROCm/clr.git + """)) + base = make_base() + url_map = base.get_submodule_url_map(tmp_path) + assert url_map["external/hip"] == "https://github.com/ROCm/hip.git" + assert url_map["external/clr"] == "https://github.com/ROCm/clr.git" + + def test_missing_url_entry_skipped(self, tmp_path): + (tmp_path / ".gitmodules").write_text(textwrap.dedent("""\ + [submodule "external/hip"] + path = external/hip + """)) + base = make_base() + assert "external/hip" not in base.get_submodule_url_map(tmp_path) + + +# --------------------------------------------------------------------------- +# RockBase.build_plan — submodule parsing logic +# --------------------------------------------------------------------------- + +class TestBuildPlanSubmoduleParsing: + """Test the submodule-filtering logic inside build_plan without a real clone.""" + + def _run_build_plan(self, tmp_path, submodule_status, url_map, exclude_list=()): + """Drive build_plan with fully mocked git operations.""" + base = make_base(exclude_list=list(exclude_list)) + base.cache_root = tmp_path + + clone_dir = tmp_path / "TheRock" + clone_dir.mkdir() + + with patch.object(base, "_prepare_clone", return_value=clone_dir), \ + patch.object(base, "_checkout_and_update_submodules"), \ + patch.object( + base, "run_command_output", + return_value="\n".join(submodule_status), + ), \ + patch.object(base, "get_submodule_url_map", return_value=url_map): + return base.build_plan() + + def test_rocm_submodule_included(self, tmp_path): + plan = self._run_build_plan( + tmp_path, + submodule_status=[f" {'b'*40} external/hip (v6.0)"], + url_map={"external/hip": "https://github.com/ROCm/hip.git"}, + ) + assert "hip" in plan + assert "TheRock" in plan + + def test_non_rocm_submodule_excluded(self, tmp_path): + plan = self._run_build_plan( + tmp_path, + submodule_status=[f" {'b'*40} external/llvm (v17)"], + url_map={"external/llvm": "https://github.com/llvm/llvm-project.git"}, + ) + assert "llvm-project" not in plan + + def test_excluded_submodule_skipped(self, tmp_path): + plan = self._run_build_plan( + tmp_path, + submodule_status=[f" {'b'*40} external/hip (v6.0)"], + url_map={"external/hip": "https://github.com/ROCm/hip.git"}, + exclude_list=["hip"], + ) + assert "hip" not in plan + + def test_missing_url_in_map_skipped(self, tmp_path): + plan = self._run_build_plan( + tmp_path, + submodule_status=[f" {'b'*40} external/hip (v6.0)"], + url_map={}, # no entry for external/hip + ) + assert "hip" not in plan + + def test_therock_always_in_plan(self, tmp_path): + plan = self._run_build_plan( + tmp_path, + submodule_status=[], + url_map={}, + ) + assert "TheRock" in plan + assert plan["TheRock"].commit == _FAKE_COMMIT + + def test_therock_excluded_from_plan(self, tmp_path): + plan = self._run_build_plan( + tmp_path, + submodule_status=[], + url_map={}, + exclude_list=["TheRock"], + ) + assert "TheRock" not in plan + + def test_sha_prefix_chars_stripped(self, tmp_path): + """Leading -, + in git submodule status output are stripped from the SHA.""" + plan = self._run_build_plan( + tmp_path, + submodule_status=[f"-{'b'*40} external/hip"], + url_map={"external/hip": "https://github.com/ROCm/hip.git"}, + ) + assert plan["hip"].commit == "b" * 40 + + +# --------------------------------------------------------------------------- +# RockBase.run_command — streaming and buffered modes +# --------------------------------------------------------------------------- + +class TestRunCommand: + def test_buffered_success_logs_stdout(self, tmp_path): + base = make_base() + logged = [] + base._logger.info = lambda msg, *a: logged.append(msg % a if a else msg) + + result = MagicMock() + result.stdout = b"hello stdout" + result.stderr = b"" + with patch("subprocess.run", return_value=result): + base.run_command(["echo", "hi"], cwd=tmp_path) + + assert any("hello stdout" in m for m in logged) + + def test_buffered_failure_raises(self, tmp_path): + base = make_base() + with patch( + "subprocess.run", + side_effect=subprocess.CalledProcessError(1, "git"), + ): + with pytest.raises(subprocess.CalledProcessError): + base.run_command(["git", "fail"], cwd=tmp_path) + + def test_stream_success(self, tmp_path): + base = make_base() + proc = MagicMock() + proc.stdout = iter(["line1\n", "line2\n"]) + proc.wait.return_value = 0 + with patch("subprocess.Popen", return_value=proc): + base.run_command(["git", "fetch"], cwd=tmp_path, stream=True) + + def test_stream_nonzero_exit_raises(self, tmp_path): + base = make_base() + proc = MagicMock() + proc.stdout = iter([]) + proc.wait.return_value = 1 + with patch("subprocess.Popen", return_value=proc): + with pytest.raises(subprocess.CalledProcessError): + base.run_command(["git", "fetch"], cwd=tmp_path, stream=True) + + def test_stream_timeout_kills_process(self, tmp_path): + base = make_base() + proc = MagicMock() + proc.stdout = iter([]) + proc.wait.side_effect = subprocess.TimeoutExpired("git", 10) + with patch("subprocess.Popen", return_value=proc): + with pytest.raises(subprocess.TimeoutExpired): + base.run_command(["git", "fetch"], cwd=tmp_path, stream=True, timeout=10) + proc.kill.assert_called_once() diff --git a/scripts/tests/test_rock_tagging.py b/scripts/tests/test_rock_tagging.py index 56ca7e7c..ec04a495 100644 --- a/scripts/tests/test_rock_tagging.py +++ b/scripts/tests/test_rock_tagging.py @@ -20,7 +20,8 @@ import sys sys.path.insert(0, str(Path(__file__).parent.parent.parent)) -from scripts.rock_tagging import RepoInfo, RockTagging +from scripts.rock_tagging import RockTagging +from scripts.release_utils import RepoInfo _FAKE_COMMIT = "a" * 40