gitblit/src/main/java/pt.py

#!/usr/bin/env python3
#
# Barnum, a Patchset Tool (pt)
#
# This Git wrapper script is designed to reduce the ceremony of working with Gitblit patchsets.
#
# Copyright 2014 gitblit.com.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
#
# Usage:
#
#    pt fetch <id> [-p,--patchset <n>]
#    pt checkout <id> [-p,--patchset <n>] [-f,--force]
#    pt pull <id> [-p,--patchset <n>]
#    pt push [<id>] [-i,--ignore] [-f,--force] [-m,--milestone <milestone>] [-t,--topic <topic>] [-cc <user> <user>]
#    pt start <topic> | <id>
#    pt propose [new | <branch> | <id>] [-i,--ignore] [-m,--milestone <milestone>] [-t,--topic <topic>] [-cc <user> <user>]
#    pt cleanup [<id>]
#

__author__ = 'James Moger'
__version__ = '1.0.6'

import subprocess
import argparse
import errno
import sys


def fetch(args):
    """
    fetch(args)

    Fetches the specified patchset for the ticket from the specified remote.
    """

    __resolve_remote(args)

    # fetch the patchset from the remote repository

    if args.patchset is None:
        # fetch all current ticket patchsets
        print("Fetching ticket patchsets from the '{}' repository".format(args.remote))
        if args.quiet:
            __call(['git', 'fetch', '-p', args.remote, '--quiet'])
        else:
            __call(['git', 'fetch', '-p', args.remote])
    else:
        # fetch specific patchset
        __resolve_patchset(args)
        print("Fetching ticket {} patchset {} from the '{}' repository".format(args.id, args.patchset, args.remote))
        patchset_ref = 'refs/tickets/{:02d}/{:d}/{:d}'.format(args.id % 100, args.id, args.patchset)
        if args.quiet:
            __call(['git', 'fetch', args.remote, patchset_ref, '--quiet'])
        else:
            __call(['git', 'fetch', args.remote, patchset_ref])

    return


def checkout(args):
    """
    checkout(args)

    Checkout the patchset on a named branch.
    """

    __resolve_uncommitted_changes_checkout(args)
    fetch(args)

    # collect local branch names
    branches = []
    for branch in __call(['git', 'branch']):
        if branch[0] == '*':
            branches.append(branch[1:].strip())
        else:
            branches.append(branch.strip())

    if args.patchset is None or args.patchset is 0:
        branch = 'ticket/{:d}'.format(args.id)
        illegals = set(branches) & {'ticket'}
    else:
        branch = 'patchset/{:d}/{:d}'.format(args.id, args.patchset)
        illegals = set(branches) & {'patchset', 'patchset/{:d}'.format(args.id)}

    # ensure there are no local branch names that will interfere with branch creation
    if len(illegals) > 0:
        print('')
        print('Sorry, can not complete the checkout for ticket {}.'.format(args.id))
        print("The following branches are blocking '{}' branch creation:".format(branch))
        for illegal in illegals:
            print('  ' + illegal)
        exit(errno.EINVAL)

    if args.patchset is None or args.patchset is 0:
        # checkout the current ticket patchset
        if args.force:
            __call(['git', 'checkout', '-B', branch, '{}/{}'.format(args.remote, branch)])
        else:
            __call(['git', 'checkout', branch])
    else:
        # checkout a specific patchset
        __checkout(args.remote, args.id, args.patchset, branch, args.force)

    return


def pull(args):
    """
    pull(args)

    Pull (fetch & merge) a ticket patchset into the current branch.
    """

    __resolve_uncommitted_changes_checkout(args)
    __resolve_remote(args)

    # reset the checkout before pulling
    __call(['git', 'reset', '--hard'])

    # pull the patchset from the remote repository
    if args.patchset is None or args.patchset is 0:
        print("Pulling ticket {} from the '{}' repository".format(args.id, args.remote))
        patchset_ref = 'ticket/{:d}'.format(args.id)
    else:
        __resolve_patchset(args)
        print("Pulling ticket {} patchset {} from the '{}' repository".format(args.id, args.patchset, args.remote))
        patchset_ref = 'refs/tickets/{:02d}/{:d}/{:d}'.format(args.id % 100, args.id, args.patchset)

    if args.squash:
        __call(['git', 'pull', '--squash', '--no-log', '--no-rebase', args.remote, patchset_ref], echo=True)
    else:
        __call(['git', 'pull', '--commit', '--no-ff', '--no-log', '--no-rebase', args.remote, patchset_ref], echo=True)

    return


def push(args):
    """
    push(args)

    Push your patchset update or a patchset rewrite.
    """

    if args.id is None:
        # try to determine ticket and patchset from current branch name
        for line in __call(['git', 'status', '-b', '-s']):
            if line[0:2] == '##':
                branch = line[2:].strip()
                segments = branch.split('/')
                if len(segments) >= 2:
                    if segments[0] == 'ticket' or segments[0] == 'patchset':
                        if '...' in segments[1]:
                            args.id = int(segments[1][:segments[1].index('...')])
                        else:
                            args.id = int(segments[1])
                        args.patchset = None

    if args.id is None:
        print('Please specify a ticket id for the push command.')
        exit(errno.EINVAL)

    __resolve_uncommitted_changes_push(args)
    __resolve_remote(args)

    if args.force:
       # rewrite a patchset for an existing ticket
        push_ref = 'refs/for/' + str(args.id)
    else:
        # fast-forward update to an existing patchset
        push_ref = 'refs/heads/ticket/{:d}'.format(args.id)

    ref_params = __get_pushref_params(args)
    ref_spec = 'HEAD:' + push_ref + ref_params

    print("Pushing your patchset to the '{}' repository".format(args.remote))
    __call(['git', 'push', args.remote, ref_spec], echo=True)
    return


def start(args):
    """
    start(args)

    Start development of a topic on a new branch.
    """

    # collect local branch names
    branches = []
    for branch in __call(['git', 'branch']):
        if branch[0] == '*':
            branches.append(branch[1:].strip())
        else:
            branches.append(branch.strip())

    branch = 'topic/' + args.topic
    try:
        int(args.topic)
        branch = 'ticket/' + args.topic
    except ValueError:
        pass

    illegals = set(branches) & {'topic', branch}

    # ensure there are no local branch names that will interfere with branch creation
    if len(illegals) > 0:
        print('Sorry, can not complete the creation of the topic branch.')
        print("The following branches are blocking '{}' branch creation:".format(branch))
        for illegal in illegals:
            print('  ' + illegal)
        exit(errno.EINVAL)

    __call(['git', 'checkout', '-b', branch])

    return


def propose(args):
    """
    propose_patchset(args)

    Push a patchset to create a new proposal ticket or to attach a proposal patchset to an existing ticket.
    """

    __resolve_uncommitted_changes_push(args)
    __resolve_remote(args)

    curr_branch = None
    push_ref = None
    if args.target is None:
        # see if the topic is a ticket id
        # else default to new
        for branch in __call(['git', 'branch']):
            if branch[0] == '*':
                curr_branch = branch[1:].strip()
                if curr_branch.startswith('topic/'):
                    topic = curr_branch[6:].strip()
                    try:
                        int(topic)
                        push_ref = topic
                    except ValueError:
                        pass
                if curr_branch.startswith('ticket/'):
                    topic = curr_branch[7:].strip()
                    try:
                        int(topic)
                        push_ref = topic
                    except ValueError:
                        pass
        if push_ref is None:
            push_ref = 'new'
    else:
        push_ref = args.target

    try:
        # check for current patchset and current branch
        args.id = int(push_ref)
        args.patchset = __get_current_patchset(args.remote, args.id)
        if args.patchset > 0:
            print('You can not propose a patchset for ticket {} because it already has one.'.format(args.id))

            # check current branch for accidental propose instead of push
            for line in __call(['git', 'status', '-b', '-s']):
                if line[0:2] == '##':
                    branch = line[2:].strip()
                    segments = branch.split('/')
                    if len(segments) >= 2:
                        if segments[0] == 'ticket':
                            if '...' in segments[1]:
                                args.id = int(segments[1][:segments[1].index('...')])
                            else:
                                args.id = int(segments[1])
                            args.patchset = None
                            print("You are on the '{}' branch, perhaps you meant to push instead?".format(branch))
                        elif segments[0] == 'patchset':
                            args.id = int(segments[1])
                            args.patchset = int(segments[2])
                            print("You are on the '{}' branch, perhaps you meant to push instead?".format(branch))
            exit(errno.EINVAL)
    except ValueError:
        pass

    ref_params = __get_pushref_params(args)
    ref_spec = 'HEAD:refs/for/{}{}'.format(push_ref, ref_params)

    print("Pushing your proposal to the '{}' repository".format(args.remote))
    for line in __call(['git', 'push', args.remote, ref_spec, '-q'], echo=True, err=subprocess.STDOUT):
        fields = line.split(':')
        if fields[0] == 'remote' and fields[1].strip().startswith('--> #'):
            # set the upstream branch configuration
            args.id = int(fields[1].strip()[len('--> #'):])
            __call(['git', 'fetch', '-p', args.remote])
            __call(['git', 'branch', '-u', '{}/ticket/{:d}'.format(args.remote, args.id)])
            break

    return


def cleanup(args):
    """
    cleanup(args)

    Removes local branches for the ticket.
    """

    if args.id is None:
        branches = __call(['git', 'branch', '--list', 'ticket/*'])
        branches += __call(['git', 'branch', '--list', 'patchset/*'])
    else:
        branches = __call(['git', 'branch', '--list', 'ticket/{:d}'.format(args.id)])
        branches += __call(['git', 'branch', '--list', 'patchset/{:d}/*'.format(args.id)])

    if len(branches) == 0:
        print("No local branches found for ticket {}, cleanup skipped.".format(args.id))
        return

    if not args.force:
        print('Cleanup would remove the following local branches for ticket {}.'.format(args.id))
        for branch in branches:
            if branch[0] == '*':
                print('  ' + branch[1:].strip() + ' (skip)')
            else:
                print('  ' + branch)
        print("To discard these local branches, repeat this command with '--force'.")
        exit(errno.EINVAL)

    for branch in branches:
        if branch[0] == '*':
            print('Skipped {} because it is the current branch.'.format(branch[1:].strip()))
            continue
        __call(['git', 'branch', '-D', branch.strip()], echo=True)

    return


def __resolve_uncommitted_changes_checkout(args):
    """
    __resolve_uncommitted_changes_checkout(args)

    Ensures the current checkout has no uncommitted changes that would be discarded by a checkout or pull.
    """

    status = __call(['git', 'status', '--porcelain'])
    for line in status:
        if not args.force and line[0] != '?':
            print('Your local changes to the following files would be overwritten by {}:'.format(args.command))
            print('')
            for state in status:
                print(state)
            print('')
            print("To discard your local changes, repeat the {} with '--force'.".format(args.command))
            print('NOTE: forcing a {} will HARD RESET your working directory!'.format(args.command))
            exit(errno.EINVAL)


def __resolve_uncommitted_changes_push(args):
    """
    __resolve_uncommitted_changes_push(args)

    Ensures the current checkout has no uncommitted changes that should be part of a propose or push.
    """

    status = __call(['git', 'status', '--porcelain'])
    for line in status:
        if not args.ignore and line[0] != '?':
            print('You have local changes that have not been committed:')
            print('')
            for state in status:
                print(state)
            print('')
            print("To ignore these uncommitted changes, repeat the {} with '--ignore'.".format(args.command))
            exit(errno.EINVAL)


def __resolve_remote(args):
    """
    __resolve_remote(args)

    Identifies the git remote to use for fetching and pushing patchsets by parsing .git/config.
    """

    remotes = __call(['git', 'remote'])

    if len(remotes) == 0:
        # no remotes defined
        print("Please define a Git remote")
        exit(errno.EINVAL)
    elif len(remotes) == 1:
        # only one remote, use it
        args.remote = remotes[0]
        return
    else:
        # multiple remotes, read .git/config
        output = __call(['git', 'config', '--local', 'patchsets.remote'], fail=False)
        preferred = output[0] if len(output) > 0 else ''

        if len(preferred) == 0:
            print("You have multiple remote repositories and you have not configured 'patchsets.remote'.")
            print("")
            print("Available remote repositories:")
            for remote in remotes:
                print('  ' + remote)
            print("")
            print("Please set the remote repository to use for patchsets.")
            print("  git config --local patchsets.remote <remote>")
            exit(errno.EINVAL)
        else:
            try:
                remotes.index(preferred)
            except ValueError:
                print("The '{}' repository specified in 'patchsets.remote' is not configured!".format(preferred))
                print("")
                print("Available remotes:")
                for remote in remotes:
                    print('  ' + remote)
                print("")
                print("Please set the remote repository to use for patchsets.")
                print("  git config --local patchsets.remote <remote>")
                exit(errno.EINVAL)

            args.remote = preferred
    return


def __resolve_patchset(args):
    """
    __resolve_patchset(args)

    Resolves the current patchset or validates the the specified patchset exists.
    """
    if args.patchset is None:
        # resolve current patchset
        args.patchset = __get_current_patchset(args.remote, args.id)

        if args.patchset == 0:
            # there are no patchsets for the ticket or the ticket does not exist
            print("There are no patchsets for ticket {} in the '{}' repository".format(args.id, args.remote))
            exit(errno.EINVAL)
    else:
        # validate specified patchset
        args.patchset = __validate_patchset(args.remote, args.id, args.patchset)

        if args.patchset == 0:
            # there are no patchsets for the ticket or the ticket does not exist
            print("Patchset {} for ticket {} can not be found in the '{}' repository".format(args.patchset, args.id, args.remote))
            exit(errno.EINVAL)

    return


def __validate_patchset(remote, ticket, patchset):
    """
    __validate_patchset(remote, ticket, patchset)

    Validates that the specified ticket patchset exists.
    """

    nps = 0
    patchset_ref = 'refs/tickets/{:02d}/{:d}/{:d}'.format(ticket % 100, ticket, patchset)
    for line in __call(['git', 'ls-remote', remote, patchset_ref]):
        ps = int(line.split('/')[4])
        if ps > nps:
            nps = ps

    if nps == patchset:
        return patchset
    return 0


def __get_current_patchset(remote, ticket):
    """
    __get_current_patchset(remote, ticket)

    Determines the most recent patchset for the ticket by listing the remote patchset refs
    for the ticket and parsing the patchset numbers from the resulting set.
    """

    nps = 0
    patchset_refs = 'refs/tickets/{:02d}/{:d}/*'.format(ticket % 100, ticket)
    for line in __call(['git', 'ls-remote', remote, patchset_refs]):
        ps = int(line.split('/')[4])
        if ps > nps:
            nps = ps

    return nps


def __checkout(remote, ticket, patchset, branch, force=False):
    """
    __checkout(remote, ticket, patchset, branch)
    __checkout(remote, ticket, patchset, branch, force)

    Checkout the patchset on a detached head or on a named branch.
    """

    has_branch = False
    on_branch = False

    if branch is None or len(branch) == 0:
        # checkout the patchset on a detached head
        print('Checking out ticket {} patchset {} on a detached HEAD'.format(ticket, patchset))
        __call(['git', 'checkout', 'FETCH_HEAD'], echo=True)
        return
    else:
        # checkout on named branch

        # determine if we are already on the target branch
        for line in __call(['git', 'branch', '--list', branch]):
            has_branch = True
            if line[0] == '*':
                # current branch (* name)
                on_branch = True

        if not has_branch:
            if force:
                # force the checkout the patchset to the new named branch
                # used when there are local changes to discard
                print("Forcing checkout of ticket {} patchset {} on named branch '{}'".format(ticket, patchset, branch))
                __call(['git', 'checkout', '-b', branch, 'FETCH_HEAD', '--force'], echo=True)
            else:
                # checkout the patchset to the new named branch
                __call(['git', 'checkout', '-b', branch, 'FETCH_HEAD'], echo=True)
            return

        if not on_branch:
            # switch to existing local branch
            __call(['git', 'checkout', branch], echo=True)

        #
        # now we are on the local branch for the patchset
        #

        if force:
            # reset HEAD to FETCH_HEAD, this drops any local changes
            print("Forcing checkout of ticket {} patchset {} on named branch '{}'".format(ticket, patchset, branch))
            __call(['git', 'reset', '--hard', 'FETCH_HEAD'], echo=True)
            return
        else:
            # try to merge the existing ref with the FETCH_HEAD
            merge = __call(['git', 'merge', '--ff-only', branch, 'FETCH_HEAD'], echo=True, fail=False)
            if len(merge) is 1:
                up_to_date = merge[0].lower().index('up-to-date') > 0
                if up_to_date:
                    return
            elif len(merge) is 0:
                print('')
                print("Your '{}' branch has diverged from patchset {} on the '{}' repository.".format(branch, patchset, remote))
                print('')
                print("To discard your local changes, repeat the checkout with '--force'.")
                print('NOTE: forcing a checkout will HARD RESET your working directory!')
                exit(errno.EINVAL)
    return


def __get_pushref_params(args):
    """
    __get_pushref_params(args)

    Returns the push ref parameters for ticket field assignments.
    """

    params = []

    if args.milestone is not None:
        params.append('m=' + args.milestone)

    if args.topic is not None:
        params.append('t=' + args.topic)
    else:
        for branch in __call(['git', 'branch']):
            if branch[0] == '*':
                b = branch[1:].strip()
                if b.startswith('topic/'):
                    topic = b[len('topic/'):]
                    try:
                        # ignore ticket id topics
                        int(topic)
                    except:
                        # topic is a string
                        params.append('t=' + topic)

    if args.responsible is not None:
        params.append('r=' + args.responsible)

    if args.cc is not None:
        for cc in args.cc:
            params.append('cc=' + cc)

    if len(params) > 0:
        return '%' + ','.join(params)

    return ''


def __call(cmd_args, echo=False, fail=True, err=None):
    """
    __call(cmd_args)

    Executes the specified command as a subprocess.  The output is parsed and returned as a list
    of strings.  If the process returns a non-zero exit code, the script terminates with that
    exit code.  Std err of the subprocess is passed-through to the std err of the parent process.
    """

    p = subprocess.Popen(cmd_args, stdout=subprocess.PIPE, stderr=err, universal_newlines=True)
    lines = []
    for line in iter(p.stdout.readline, b''):
        line_str = str(line).strip()
        if len(line_str) is 0:
            break
        lines.append(line_str)
        if echo:
            print(line_str)
    p.wait()
    if fail and p.returncode is not 0:
        exit(p.returncode)

    return lines

#
# define the acceptable arguments and their usage/descriptions
#

# force argument
force_arg = argparse.ArgumentParser(add_help=False)
force_arg.add_argument('-f', '--force', default=False, help='force the command to complete', action='store_true')

# quiet argument
quiet_arg = argparse.ArgumentParser(add_help=False)
quiet_arg.add_argument('-q', '--quiet', default=False, help='suppress git stderr output', action='store_true')

# ticket & patchset arguments
ticket_args = argparse.ArgumentParser(add_help=False)
ticket_args.add_argument('id', help='the ticket id', type=int)
ticket_args.add_argument('-p', '--patchset', help='the patchset number', type=int)

# push refspec arguments
push_args = argparse.ArgumentParser(add_help=False)
push_args.add_argument('-i', '--ignore', default=False, help='ignore uncommitted changes', action='store_true')
push_args.add_argument('-m', '--milestone', help='set the milestone')
push_args.add_argument('-r', '--responsible', help='set the responsible user')
push_args.add_argument('-t', '--topic', help='set the topic')
push_args.add_argument('-cc', nargs='+', help='specify accounts to add to the watch list')

# the commands
parser = argparse.ArgumentParser(description='a Patchset Tool for Gitblit Tickets')
parser.add_argument('--version', action='version', version='%(prog)s {}'.format(__version__))
commands = parser.add_subparsers(dest='command', title='commands')

fetch_parser = commands.add_parser('fetch', help='fetch a patchset', parents=[ticket_args, quiet_arg])
fetch_parser.set_defaults(func=fetch)

checkout_parser = commands.add_parser('checkout', aliases=['co'],
                                      help='fetch & checkout a patchset to a branch',
                                      parents=[ticket_args, force_arg, quiet_arg])
checkout_parser.set_defaults(func=checkout)

pull_parser = commands.add_parser('pull',
                                  help='fetch & merge a patchset into the current branch',
                                  parents=[ticket_args, force_arg])
pull_parser.add_argument('-s', '--squash',
                         help='squash the pulled patchset into your working directory',
                         default=False,
                         action='store_true')
pull_parser.set_defaults(func=pull)

push_parser = commands.add_parser('push', aliases=['up'],
                                  help='upload your patchset changes',
                                  parents=[push_args, force_arg])
push_parser.add_argument('id', help='the ticket id', nargs='?', type=int)
push_parser.set_defaults(func=push)

propose_parser = commands.add_parser('propose', help='propose a new ticket or the first patchset', parents=[push_args])
propose_parser.add_argument('target', help="the ticket id, 'new', or the integration branch", nargs='?')
propose_parser.set_defaults(func=propose)

cleanup_parser = commands.add_parser('cleanup', aliases=['rm'],
                                     help='remove local ticket branches',
                                     parents=[force_arg])
cleanup_parser.add_argument('id', help='the ticket id', nargs='?', type=int)
cleanup_parser.set_defaults(func=cleanup)

start_parser = commands.add_parser('start', help='start a new branch for the topic or ticket')
start_parser.add_argument('topic', help="the topic or ticket id")
start_parser.set_defaults(func=start)

if len(sys.argv) < 2:
    parser.parse_args(['--help'])
else:
    # parse the command-line arguments
    script_args = parser.parse_args()

    # exec the specified command
    script_args.func(script_args)