fastai-make-pr-branch-py

#!/usr/bin/env python3
"""
This program will checkout a forked version of the original repository, sync it with the original, create a new branch and set it up for a PR.

usage:

{me} auth user_name repo_name new_branch_name

parameters:

  auth:   ssh or https (use ssh if your github account is setup to use ssh
  user:   your github username
  repo:   repository name to fork/use
  branch: name of the branch to create

example:

  {me} ssh my-github-username fastai new-feature-branch

notes:

- if the original repository has been already forked, it'll be done by the program (it will ask you for your github password)
- if the program is run from a directory that already contains a checked out git repository that matches the parameters, it will re-use it instead of making a new checkout
- if the requested branch already exists it'll reuse it
- if the master is out of sync with the original master it'll sync it first

"""
import argparse
import os
from os.path import basename, exists
from textwrap import dedent
import sys
from subprocess import call, PIPE, Popen
from pathlib import Path

PY3 = sys.version_info >= (3, 0)
ME = basename(sys.argv[0])
ORIG_USER_NAME = 'fastai'


def main():
    args = parse_args()
    was_already_forked = fork_if_needed(args.url, args.user_name, args.repo_name)
    need_clone = clone_is_needed(args.url)
    if need_clone:
        clone(args.url, args.path)
    run_after_git_clone()
    track_upstream(args.orig_repo_url)
    if was_already_forked:
        sync_with_master(args)
    create_branch(args.new_branch_name)
    set_upstream(args.new_branch_name)
    print_instructions(args, need_clone)


def parse_args():
    parser = SysExitArgumentParser(usage=__doc__.format(me=ME))
    parser.add_argument(
        'auth',
        metavar='AUTH', choices={'ssh', 'https'},
        help='ssh or https (use ssh if your github account is setup to use ssh)'
    )
    parser.add_argument(
        'user_name',
        metavar='USER',
        help='your github username'
    )
    parser.add_argument(
        'repo_name',
        metavar='REPO',
        help='repository name to fork/use'
    )
    parser.add_argument(
        'new_branch_name',
        metavar='BR',
        help='name of the branch to create'
    )

    args = parser.parse_args()
    args.prefix = 'git@github.com:' if args.auth == 'ssh' else 'https://github.com/'
    args.path = f'{args.repo_name}-{args.new_branch_name}'
    args.url = f'{args.prefix}{args.user_name}/{args.repo_name}.git'
    args.orig_repo_url = f'{args.prefix}{ORIG_USER_NAME}/{args.repo_name}.git'

    return args


def check_curl_is_available():
    "Check if curl is available and terminate the script if it isn't."
    try:
        call(['curl'], stdout=PIPE, stderr=PIPE)
    except FileNotFoundError:
        print(f"{ME} requires 'curl' but it's not installed. Aborting.")
        sys.exit(1)


def fork_if_needed(url, user, repo):
    "Fork the repo if it hasn't been forked yet."
    print('\n\n*** Checking if we need to fork')
    if run_and_check_success(f'git ls-remote {url} >/dev/null 2>&1'):
        print(f'{url} is already forked')
        return True
    else:
        print(f"{url} doesn't exist - need to fork")
        check_curl_is_available()
        run(f"curl -u {user} https://api.github.com/repos/fastai/{repo}/forks -d ''")
        return False


def clone_is_needed(url):
    "Support an already existing checkout if it matches the desired parameters, but only if it's not dirty."
    if not exists('.git'):
        return True
    print('detected current dir to be a git repository')
    output = run_and_read_all('git status -s')
    if output:
        print('git status is not clean, not going to re-use the current checkout')
        return True
    cur_url = run_and_read_all('git config --get remote.origin.url')
    if cur_url != url:
        return True
    print('already on the right repo, going to re-use the current checkout')
    return False


def clone(url, path):
    "Clone `url` into `path`."
    print(f'\n\n*** Cloning {url} to {path}')
    run(f'git clone {url} {path}')
    os.chdir(path)
    print(f'cd {path}')


def run_after_git_clone():
    "Run fastai's tools/run-after-git-clone if exists."
    tool = Path("tools")/"run-after-git-clone"
    if tool.exists():
        print(f'\n\n*** run {tool}')
        run(f'{sys.executable} {tool}')


def track_upstream(orig_repo_url):
    "Setup this fork to track the upstream."
    print('\n\n*** Setting up this fork to track the upstream')
    run(f"git remote add upstream {orig_repo_url} || echo 'already exists'")
    run('git remote -v')


def sync_with_master(args):
    "Sync the forked repo master with the upstream master if it hasn't been just forked."
    user, repo = args.user_name, args.repo_name
    print(f'\n\n*** Syncing {user}/master with {ORIG_USER_NAME}/master')
    run('git fetch upstream')
    run('git checkout master')
    # XXX: this will fail if the user was committing directly to their forked `master` branch is
    ok = run_and_check_success('git merge --no-edit upstream/master')
    if not ok:
        run('git merge --abort')
        print(dedent(f'''
            Error: {user}/{repo}/master can't be merged with {ORIG_USER_NAME}/{repo}/master.
            You must merge manually, or reset {user}/{repo}/master.
            See:  https://docs-dev.fast.ai/git.html#how-to-reset-your-forked-master-branch
            Aborting."
        '''), file=sys.stderr)
        sys.exit(1)
    run('git push')
    print('Done')


def create_branch(new_branch_name):
    print(f"\n\n*** Creating '{new_branch_name}' branch and setting its upstream for an easy push")
    if run_and_check_success(f'git show-branch {new_branch_name} >/dev/null 2>&1'):
        print(f'{new_branch_name} already exists, switching to it')
        run(f'git checkout {new_branch_name}')
    else:
        print(f"{new_branch_name} doesn't yet exist, creating")
        run(f'git checkout -b {new_branch_name}')
        print(f'{new_branch_name} branch was successfully created')


def set_upstream(branch):
    print(f"\n\n*** Set this {branch}'s upstream to simplify push and alikes")
    run(f'git push --set-upstream origin {branch}')


def print_instructions(args, need_clone):
    switch_command = f'cd {args.path}' if need_clone else 'cd .'
    print(dedent(f'''

        Next do:

        1. Switch to the branch dir, unless you're already there:

        {switch_command}

        2. Validate you're on the right repo/branch by pasting the following into your bash:

        echo on repo: $(git config --get remote.origin.url | sed 's|^.*//||; s/.*@//; s/[^:/]\+[:/]//; s/.git$//') branch: $(git branch | sed -n '/\* /s///p')

        3. And do your work.

        4. When you're done:

        git add <newfiles>  # if you added new files
        git commit -a       # if you want to commit all modified/added files
        git push            # push files into {args.new_branch_name}

        5. Then go to:
          https://github.com/{args.user_name}/{args.repo_name}/tree/{args.new_branch_name}

        review your changes and when you're happy with what you see, submit your PR.

        While it's very good to review your PR on github before submitting it, if you want to use a CLI solution,
        there are several solutions to choose from:
        * https://github.com/github/hub"
        * https://github.com/jd/git-pull-request"
        * https://github.com/jehiah/git-open-pull"
        * https://www.npmjs.com/package/pullr"
    '''))


# --------------
# Shell wrappers
# --------------

# some ideas are taken from:
#
#   https://github.com/pytorch/pytorch/blob/master/torch/utils/collect_env.py
#

class SysExitArgumentParser(argparse.ArgumentParser):

    def error(self, message):
        print(self.usage)
        sys.exit(1)


def run(command):
    print(f'running: {command}')
    rc, out, err = run_and_return_output(command)
    if rc != 0:
        print(err)
        sys.exit(1)
    if out.strip():
        print(out)


def run_and_return_output(command):
    "Returns (return-code, stdout, stderr)."
    p = Popen(command, stdout=PIPE, stderr=PIPE, shell=True)
    output, err = p.communicate()
    rc = p.returncode
    if PY3:
        output = output.decode("ascii")
        err = err.decode("ascii")
    return rc, output.strip(), err.strip()


def run_and_read_all(command):
    "Runs command using run_lambda; reads and returns entire output if rc is 0."
    rc, out, _ = run_and_return_output(command)
    if rc is not 0:
        return None
    return out


def run_and_check_success(command):
    print(f'running: {command}')
    rc, _, _ = run_and_return_output(command)
    return rc == 0


if __name__ == '__main__':
    main()