mirror of
https://github.com/RIOT-OS/RIOT.git
synced 2024-12-29 04:50:03 +01:00
291 lines
11 KiB
Python
Executable File
291 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
# Copyright (C) 2018 Freie Universitat Berlin
|
|
# Copyright (C) 2018 Inria
|
|
#
|
|
# This file is subject to the terms and conditions of the GNU Lesser
|
|
# General Public License v2.1. See the file LICENSE in the top level
|
|
# directory for more details.
|
|
#
|
|
# @author Koen Zandberg <koen@bergzand.net>
|
|
|
|
"""Script used to backport PRs."""
|
|
|
|
import os
|
|
import os.path
|
|
import sys
|
|
import shutil
|
|
import argparse
|
|
|
|
import git
|
|
from agithub.GitHub import GitHub
|
|
|
|
ORG = "RIOT-OS"
|
|
REPO = "RIOT"
|
|
GITHUBTOKEN_FILE = ".riotgithubtoken"
|
|
WORKTREE_SUBDIR = "backport_temp"
|
|
|
|
RELEASE_PREFIX = ""
|
|
RELEASE_SUFFIX = "-branch"
|
|
|
|
LABELS_REMOVE = ['Process: needs backport', 'Reviewed: ']
|
|
LABELS_ADD = ['Process: release backport']
|
|
|
|
BACKPORT_BRANCH = 'backport/{release}/{origbranch}'
|
|
|
|
|
|
def _get_labels(pull_request):
|
|
"""
|
|
>>> _get_labels({'labels': [{'name': 'test'}, {'name': 'abcd'}]})
|
|
['Process: release backport', 'abcd', 'test']
|
|
>>> _get_labels({'labels': [{'name': 'Reviewed: what'}, \
|
|
{'name': 'Reviewed: 3-testing'}]})
|
|
['Process: release backport']
|
|
>>> _get_labels({'labels': [{'name': 'Process: release backport'}]})
|
|
['Process: release backport']
|
|
>>> _get_labels({'labels': [{'name': 'Process: needs backport'}]})
|
|
['Process: release backport']
|
|
"""
|
|
labels = set(label['name'] for label in pull_request['labels']
|
|
if all(not label['name'].startswith(remove)
|
|
for remove in LABELS_REMOVE))
|
|
labels.update(LABELS_ADD)
|
|
return sorted(list(labels))
|
|
|
|
|
|
def _branch_name_strip(branch_name, prefix=RELEASE_PREFIX,
|
|
suffix=RELEASE_SUFFIX):
|
|
"""Strip suffix and prefix.
|
|
|
|
>>> _branch_name_strip('2018.10-branch')
|
|
'2018.10'
|
|
"""
|
|
if (branch_name.startswith(prefix) and
|
|
branch_name.endswith(suffix)):
|
|
if prefix:
|
|
branch_name = branch_name.split(prefix, maxsplit=1)[0]
|
|
if suffix:
|
|
branch_name = branch_name.rsplit(suffix, maxsplit=1)[0]
|
|
return branch_name
|
|
|
|
|
|
def _get_latest_release(branches):
|
|
"""Get latest release from a list of branches.
|
|
|
|
>>> _get_latest_release([{'name': '2018.10-branch'}, \
|
|
{'name': '2020.10-branch'}])
|
|
('2020.10', '2020.10-branch')
|
|
>>> _get_latest_release([{'name': '2020.01-branch'}, \
|
|
{'name': '2020.04-branch'}])
|
|
('2020.04', '2020.04-branch')
|
|
>>> _get_latest_release([{'name': 'non-release-branch'}, \
|
|
{'name': '2020.04-branch'}])
|
|
('2020.04', '2020.04-branch')
|
|
"""
|
|
version_latest = 0
|
|
release_fullname = ''
|
|
release_short = ''
|
|
for branch in branches:
|
|
branch_name = _branch_name_strip(branch['name'])
|
|
branch_num = 0
|
|
try:
|
|
branch_num = int(''.join(branch_name.split('.')))
|
|
except ValueError:
|
|
pass
|
|
if branch_num > version_latest:
|
|
version_latest = branch_num
|
|
release_short = branch_name
|
|
release_fullname = branch['name']
|
|
return (release_short, release_fullname)
|
|
|
|
|
|
def _find_remote(repo, user, repo_name):
|
|
for remote in repo.remotes:
|
|
if (remote.url.endswith("{}/{}.git".format(user, repo_name)) or
|
|
remote.url.endswith("{}/{}".format(user, repo_name))):
|
|
return remote
|
|
raise ValueError("Could not find remote with URL ending in {}/{}.git"
|
|
.format(user, repo_name))
|
|
|
|
|
|
def _get_upstream(repo):
|
|
return _find_remote(repo, ORG, REPO)
|
|
|
|
|
|
def _delete_worktree(repo, workdir):
|
|
shutil.rmtree(workdir)
|
|
repo.git.worktree('prune')
|
|
|
|
|
|
def main():
|
|
# pylint:disable=too-many-locals,too-many-branches,too-many-statements
|
|
"""Main function of this script."""
|
|
keyfile = os.path.join(os.environ['HOME'], GITHUBTOKEN_FILE)
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("-k", "--keyfile", type=argparse.FileType('r'),
|
|
default=keyfile,
|
|
help="File containing github token")
|
|
parser.add_argument("-c", "--comment", action="store_true",
|
|
help="Put a comment with a reference under"
|
|
"the original PR")
|
|
parser.add_argument("-n", "--noop", action="store_true",
|
|
help="Limited noop mode, creates branch, but doesn't"
|
|
"push and create the PR")
|
|
parser.add_argument("-r", "--release-branch", type=str,
|
|
help="Base the backport on this branch, "
|
|
"default is the latest")
|
|
parser.add_argument("--backport-branch-fmt", type=str,
|
|
default=BACKPORT_BRANCH,
|
|
help="Backport branch format. "
|
|
"Fields '{release}' and '{origbranch} will be "
|
|
"replaced by the release name and remote branch "
|
|
"name.")
|
|
parser.add_argument('-d', '--gitdir', type=str, default=os.getcwd(),
|
|
help="Base git repo to work from")
|
|
parser.add_argument("PR", type=int, help="Pull request number to backport")
|
|
args = parser.parse_args()
|
|
|
|
gittoken = args.keyfile.read().strip()
|
|
github_api = GitHub(token=gittoken)
|
|
# TODO: exception handling
|
|
status, user = github_api.user.get()
|
|
if status != 200:
|
|
print("Could not retrieve user: {}".format(user['message']))
|
|
sys.exit(1)
|
|
# Token-scope-check: Is the token is powerful enough to complete
|
|
# the Backport?
|
|
response_headers = dict(github_api.getheaders())
|
|
# agithub documentation says it's lower case header field-names but
|
|
# at this moment it's not
|
|
if 'X-OAuth-Scopes' in response_headers:
|
|
scopes = response_headers['X-OAuth-Scopes']
|
|
else:
|
|
scopes = response_headers['x-oauth-scopes']
|
|
scopes_list = [x.strip() for x in scopes.split(',')]
|
|
if not ('public_repo' in scopes_list or 'repo' in scopes_list):
|
|
print("missing public_repo scope from token settings."
|
|
" Please add it on the GitHub webinterface")
|
|
sys.exit(1)
|
|
username = user['login']
|
|
status, pulldata = github_api.repos[ORG][REPO].pulls[args.PR].get()
|
|
if status != 200:
|
|
print("Commit #{} not found: {}".format(args.PR, pulldata['message']))
|
|
sys.exit(2)
|
|
if not pulldata['merged']:
|
|
print("Original PR not yet merged")
|
|
sys.exit(0)
|
|
print("Fetching for commit: #{}: {}".format(args.PR, pulldata['title']))
|
|
orig_branch = pulldata['head']['ref']
|
|
status, commits = github_api.repos[ORG][REPO].pulls[args.PR].commits.get()
|
|
if status != 200:
|
|
print("No commits found for #{}: {}".format(args.PR,
|
|
commits['message']))
|
|
sys.exit(3)
|
|
for commit in commits:
|
|
print("found {} : {}".format(commit['sha'],
|
|
commit['commit']['message']))
|
|
|
|
# Find latest release branch
|
|
if args.release_branch:
|
|
release_fullname = args.release_branch
|
|
release_shortname = _branch_name_strip(args.release_branch)
|
|
else:
|
|
status, branches = github_api.repos[ORG][REPO].branches.get()
|
|
if status != 200:
|
|
print("Could not retrieve branches for {}/{}: {}"
|
|
.format(ORG,
|
|
REPO,
|
|
branches['message']))
|
|
sys.exit(4)
|
|
release_shortname, release_fullname = _get_latest_release(branches)
|
|
if not release_fullname:
|
|
print("No release branch found, exiting")
|
|
sys.exit(5)
|
|
print("Backport based on branch {}".format(release_fullname))
|
|
|
|
repo = git.Repo(args.gitdir)
|
|
# Fetch current upstream
|
|
upstream_remote = _get_upstream(repo)
|
|
if not upstream_remote:
|
|
print("No upstream remote found, can't fetch")
|
|
sys.exit(6)
|
|
print("Fetching {} remote".format(upstream_remote))
|
|
|
|
upstream_remote.fetch()
|
|
# Build topic branch in temp dir
|
|
new_branch = args.backport_branch_fmt.format(release=release_shortname,
|
|
origbranch=orig_branch)
|
|
if new_branch in repo.branches:
|
|
print("ERROR: Branch {} already exists".format(new_branch))
|
|
sys.exit(1)
|
|
worktree_dir = os.path.join(args.gitdir, WORKTREE_SUBDIR)
|
|
repo.git.worktree("add", "-b",
|
|
new_branch,
|
|
WORKTREE_SUBDIR,
|
|
"{}/{}".format(upstream_remote, release_fullname))
|
|
# transform branch name into Head object for later configuring
|
|
new_branch = repo.branches[new_branch]
|
|
try:
|
|
bp_repo = git.Repo(worktree_dir)
|
|
# Apply commits
|
|
for commit in commits:
|
|
bp_repo.git.cherry_pick('-x', commit['sha'])
|
|
# Push to github
|
|
origin = _find_remote(repo, username, REPO)
|
|
print("Pushing branch {} to {}".format(new_branch, origin))
|
|
if not args.noop:
|
|
push_info = origin.push('{0}:{0}'.format(new_branch))
|
|
new_branch.set_tracking_branch(push_info[0].remote_ref)
|
|
except Exception as exc:
|
|
# Delete worktree
|
|
print("Pruning temporary workdir at {}".format(worktree_dir))
|
|
_delete_worktree(repo, worktree_dir)
|
|
# also delete branch created by worktree; this is only possible after
|
|
# the worktree was deleted
|
|
repo.delete_head(new_branch)
|
|
raise exc
|
|
else:
|
|
# Delete worktree
|
|
print("Pruning temporary workdir at {}".format(worktree_dir))
|
|
_delete_worktree(repo, worktree_dir)
|
|
|
|
labels = _get_labels(pulldata)
|
|
merger = pulldata['merged_by']['login']
|
|
if not args.noop:
|
|
# Open new PR on github
|
|
pull_request = {
|
|
'title': "{} [backport {}]".format(pulldata['title'],
|
|
release_shortname),
|
|
'head': '{}:{}'.format(username, new_branch),
|
|
'base': release_fullname,
|
|
'body': "# Backport of #{}\n\n{}".format(args.PR,
|
|
pulldata['body']),
|
|
'maintainer_can_modify': True,
|
|
}
|
|
status, new_pr = github_api.repos[ORG][REPO].pulls.post(
|
|
body=pull_request)
|
|
if status != 201:
|
|
print("Error creating the new pr: \"{}\". Is \"Public Repo\""
|
|
" access enabled for the token"
|
|
.format(new_pr['message']))
|
|
pr_number = new_pr['number']
|
|
print("Create PR number #{} for backport".format(pr_number))
|
|
github_api.repos[ORG][REPO].issues[pr_number].labels.post(body=labels)
|
|
review_request = {"reviewers": [merger]}
|
|
github_api.repos[ORG][REPO].pulls[pr_number].\
|
|
requested_reviewers.post(body=review_request)
|
|
|
|
# Put commit under old PR
|
|
if args.comment and not args.noop:
|
|
comment = {"body": "Backport provided in #{}".format(pr_number)}
|
|
status, res = github_api.repos[ORG][REPO].\
|
|
issues[args.PR].comments.post(body=comment)
|
|
if status != 201:
|
|
print("Something went wrong adding the comment: {}"
|
|
.format(res['message']))
|
|
print("Added comment to #{}".format(args.PR))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|