Skip to content

Commit

Permalink
[Utils] Add new merge-release-pr.py script. (#101630)
Browse files Browse the repository at this point in the history
This script helps the release managers merge backport PR's.

It does the following things:

* Validate the PR, checks approval, target branch and many other things.
* Rebases the PR
* Checkout the PR locally
* Pushes the PR to the release branch
* Deletes the local branch

I have found the script very helpful to merge the PR's.
  • Loading branch information
tru authored Aug 10, 2024
1 parent 80eea01 commit f3e950a
Showing 1 changed file with 254 additions and 0 deletions.
254 changes: 254 additions & 0 deletions llvm/utils/release/merge-release-pr.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,254 @@
#!/usr/bin/env python3
# ===-- merge-release-pr.py ------------------------------------------------===#
#
# Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
# See https://llvm.org/LICENSE.txt for license information.
# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
#
# ===------------------------------------------------------------------------===#

"""
Helper script that will merge a Pull Request into a release branch. It will first
do some validations of the PR then rebase and finally push the changes to the
release branch.
Usage: merge-release-pr.py <PR id>
By default it will push to the 'upstream' origin, but you can pass
--upstream-origin/-o <origin> if you want to change it.
If you want to skip a specific validation, like the status checks you can
pass -s status_checks, this argument can be passed multiple times.
"""

import argparse
import json
import subprocess
import sys
import time
from typing import List


class PRMerger:
def __init__(self, args):
self.args = args

def run_gh(self, gh_cmd: str, args: List[str]) -> str:
cmd = ["gh", gh_cmd, "-Rllvm/llvm-project"] + args
p = subprocess.run(cmd, capture_output=True)
if p.returncode != 0:
print(p.stderr)
raise RuntimeError("Failed to run gh")
return p.stdout

def validate_state(self, data):
"""Validate the state of the PR, this means making sure that it is OPEN and not already merged or closed."""
state = data["state"]
if state != "OPEN":
return False, f"state is {state.lower()}, not open"
return True

def validate_target_branch(self, data):
"""
Validate that the PR is targetting a release/ branch. We could
validate the exact branch here, but I am not sure how to figure
out what we want except an argument and that might be a bit to
to much overhead.
"""
baseRefName: str = data["baseRefName"]
if not baseRefName.startswith("release/"):
return False, f"target branch is {baseRefName}, not a release branch"
return True

def validate_approval(self, data):
"""
Validate the approval decision. This checks that the PR has been
approved.
"""
if data["reviewDecision"] != "APPROVED":
return False, "PR is not approved"
return True

def validate_status_checks(self, data):
"""
Check that all the actions / status checks succeeded. Will also
fail if we have status checks in progress.
"""
failures = []
pending = []
for status in data["statusCheckRollup"]:
if "conclusion" in status and status["conclusion"] == "FAILURE":
failures.append(status)
if "status" in status and status["status"] == "IN_PROGRESS":
pending.append(status)

if failures or pending:
errstr = "\n"
if failures:
errstr += " FAILED: "
errstr += ", ".join([d["name"] for d in failures])
if pending:
if failures:
errstr += "\n"
errstr += " PENDING: "
errstr += ", ".join([d["name"] for d in pending])

return False, errstr

return True

def validate_commits(self, data):
"""
Validate that the PR contains just one commit. If it has more
we might want to squash. Which is something we could add to
this script in the future.
"""
if len(data["commits"]) > 1:
return False, f"More than 1 commit! {len(data['commits'])}"
return True

def validate_pr(self):
fields_to_fetch = [
"baseRefName",
"reviewDecision",
"title",
"statusCheckRollup",
"url",
"state",
"commits",
]
o = self.run_gh(
"pr",
["view", self.args.pr, "--json", ",".join(fields_to_fetch)],
)
prdata = json.loads(o)

# save the baseRefName (target branch) so that we know where to push
self.target_branch = prdata["baseRefName"]

print(f"> Handling PR {self.args.pr} - {prdata['title']}")
print(f"> {prdata['url']}")

VALIDATIONS = {
"state": self.validate_state,
"target_branch": self.validate_target_branch,
"approval": self.validate_approval,
"commits": self.validate_commits,
"status_checks": self.validate_status_checks,
}

print()
print("> Validations:")
total_ok = True
for val_name, val_func in VALIDATIONS.items():
try:
validation_data = val_func(prdata)
except:
validation_data = False
ok = None
skipped = (
True
if (self.args.skip_validation and val_name in self.args.skip_validation)
else False
)
if isinstance(validation_data, bool) and validation_data:
ok = "OK"
elif isinstance(validation_data, tuple) and not validation_data[0]:
failstr = validation_data[1]
if skipped:
ok = "SKIPPED: "
else:
total_ok = False
ok = "FAIL: "
ok += failstr
else:
ok = "FAIL! (Unknown)"
print(f" * {val_name}: {ok}")
return total_ok

def rebase_pr(self):
print("> Rebasing")
self.run_gh("pr", ["update-branch", "--rebase", self.args.pr])
print("> Waiting for GitHub to update PR")
time.sleep(4)

def checkout_pr(self):
print("> Fetching PR changes...")
self.run_gh(
"pr",
[
"checkout",
self.args.pr,
"--force",
"--branch",
"llvm_merger_" + self.args.pr,
],
)

def push_upstream(self):
print("> Pushing changes...")
subprocess.run(
["git", "push", self.args.upstream, "HEAD:" + self.target_branch],
check=True,
)

def delete_local_branch(self):
print("> Deleting the old branch...")
subprocess.run(["git", "switch", "main"])
subprocess.run(["git", "branch", "-D", f"llvm_merger_{self.args.pr}"])


if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument(
"pr",
help="The Pull Request ID that should be merged into a release.",
)
parser.add_argument(
"--skip-validation",
"-s",
action="append",
help="Skip a specific validation, can be passed multiple times. I.e. -s status_checks -s approval",
)
parser.add_argument(
"--upstream-origin",
"-o",
default="upstream",
dest="upstream",
help="The name of the origin that we should push to. (default: upstream)",
)
parser.add_argument(
"--no-push",
action="store_true",
help="Run validations, rebase and fetch, but don't push.",
)
parser.add_argument(
"--validate-only", action="store_true", help="Only run the validations."
)
args = parser.parse_args()

merger = PRMerger(args)
if not merger.validate_pr():
print()
print(
"! Validations failed! Pass --skip-validation/-s <validation name> to pass this, can be passed multiple times"
)
sys.exit(1)

if args.validate_only:
print()
print("! --validate-only passed, will exit here")
sys.exit(0)

merger.rebase_pr()
merger.checkout_pr()

if args.no_push:
print()
print("! --no-push passed, will exit here")
sys.exit(0)

merger.push_upstream()
merger.delete_local_branch()

print()
print("> Done! Have a nice day!")

0 comments on commit f3e950a

Please sign in to comment.