push-rewrite

#! /usr/bin/env python3

# push-rewrite -- Force push to a PR after rewriting history, with benefits.
#
# This tool force pushes your local changes to origin but only if that
# doesn't change any files.  If that is the case, the tool will also
# copy the test results over.
#
# The idea is that you use this tool after squashing fixups in a pull
# request or after other similar last minute rewriting activity before
# merging a pull request.  Then you can merge the PR using the GitHub
# UI and get a nice "merged" label for it, and the tests will still be
# green.

import subprocess
import argparse
import sys
import time

from task import github, label, labels_of_pull

def execute(*args):
    try:
        sys.stderr.write("+ " + " ".join(args) + "\n")
        output = subprocess.check_output(args, universal_newlines=True)
    except subprocess.CalledProcessError as ex:
        sys.exit(ex.returncode)
    return output

def git(*args):
    return execute("git", *args).strip()

def find_pr_with_sha(api, sha):
    pulls = api.pulls()
    for pull in pulls:
        if pull["head"]["sha"] == sha:
            return pull
    sys.stderr.write("Could not find pull with revision {0}.\n".format(sha))
    sys.exit(1)

def main():
    parser = argparse.ArgumentParser(description='Force push after a rewrite')
    parser.add_argument('--repo', help="The GitHub repository to work with", default=None)
    parser.add_argument('remote', help='The remote to push to')
    parser.add_argument('branch', nargs='?',  help='The branch to push')
    opts = parser.parse_args()

    if not opts.branch:
        opts.branch = git("rev-parse", "--abbrev-ref", "HEAD")

    local = git("rev-parse", opts.branch)
    remote = git("rev-parse", opts.remote + "/" + opts.branch)

    if local == remote:
        sys.stderr.write("Nothing to push.\n")
        return 0

    if git("diff", "--stat", local, remote) != "":
        sys.stderr.write("You have local changes, aborting.\n")
        return 1

    api = github.GitHub(repo=opts.repo)
    old_statuses = api.statuses(remote)

    pull = find_pr_with_sha(api, remote)
    labels = labels_of_pull(pull)
    tests_disabled = "no-test" in labels or "[no-test]" in pull["title"]
    # add no-test label to prevent tests bring triggered
    if not tests_disabled:
        label(pull, { "labels": labels + ["no-test"] })

    git("push", "--force-with-lease=" + opts.branch + ":" + remote, opts.remote, opts.branch + ":" + opts.branch)

    for n in range(100,0,-1):
        try:
            if api.get("commits/%s" % local):
                break
        except RuntimeError as e:
            if not "Unprocessable Entity" in str(e) or n <= 1:
                raise
            print("(new commits not yet in the GiHub API...please stand by. *beep*)")
        time.sleep(1)

    for key in old_statuses:
        if old_statuses[key]["state"] != "pending":
            print("Copying results for %s" % old_statuses[key]["context"])
            api.post("statuses/" + local, old_statuses[key])

    # remove no-test label
    if not tests_disabled:
        api.delete("issues/%i/labels/no-test" % pull["number"])

if __name__ == '__main__':
    sys.exit(main())