hg extension for pelo branching model

hgpelo.py

# coding: utf8
"""pelo

pelo branch management wrapper commands
"""
from datetime import datetime
from mercurial import commands


HELP_DOC = """COMMAND OPERATION [label]

  pelo release start uXXXX .... start new release
  pelo release finish ......... finish current release

  pelo hotfix start uXXXX ..... start new hotfix
  pelo hotfix finish .......... finish current hotfix
"""


def pelo_cmd(ui, repo, cmd, op, label=None, **opts):
    if cmd == 'release':
        return release(ui, repo, op, label, **opts)
    elif cmd == 'hotfix':
        return hotfix(ui, repo, op, label, **opts)
    else:
        _write(ui, HELP_DOC)

def release(ui, repo, op, label, **opts):
    prefix = 'release'
    start_at = 'default'
    return operate_branch(ui, repo, op, label, prefix, start_at, **opts)

def hotfix(ui, repo, op, label, **opts):
    prefix = 'hotfix'
    start_at = 'master'
    return operate_branch(ui, repo, op, label, prefix, start_at, **opts)

def operate_branch(ui, repo, op, label, prefix, start_at, **opts):
    if op == 'start':
        return start_branch(ui, repo, label, prefix, start_at, **opts)
    elif op == 'finish':
        return finish_branch(ui, repo, label, prefix)
    else:
        _write(ui, HELP_DOC)

def start_branch(ui, repo, label, prefix, start_at, **opts):
    """start prefix branch"""
    # exit if label is not provided
    if not label:
        _write(ui,
               'abort: You must specify ticket to start %s branch.' % prefix)
        return
    # exit if not in default branch
    if _get_current_branch(repo) != start_at:
        _write(ui,
               'abort: Run "hg update default" before starting %s branch.' % start_at)
        return
    # branch <prefix>/<label>
    commands.branch(ui, repo, '%s/%s' % (prefix, label))
    # commit to mark start
    commands.commit(ui, repo, message='started %s/%s' % (prefix, label))

def finish_branch(ui, repo, label, prefix='hotfix', **opts):
    """finish hotfix branch"""
    # 変更が残っていたらエラー
    status = repo.status()
    for l in status:
        if len(l) > 0:
            _write(ui, 'abort: Your workspace has uncommit content.')
            return
    # label が指定されていたらエラー
    if label:
        ui.write('abort: You cannot specify branch name\n')
        return
    # <prefix> ブランチじゃなかったらエラー
    if not _get_current_branch(repo).startswith('%s/' % prefix):
        _write(ui,
               'abort: you must be in %s branch' % prefix)
        return
    # determine current branch
    current_branch = _get_current_branch(repo)
    # merge into default branch
    error = _merge(ui, repo, current_branch, 'default')
    if error:
        return
    # merge into master branch
    error = _merge(ui, repo, current_branch, 'master')
    if error:
        return
    # notify
    _write(ui, '%s was successfully finished.' % current_branch)

def _get_current_branch(repo):
    ctx = repo[None]
    current_branch = str(ctx.branch())
    return current_branch

def _write(ui, message):
    ui.status(message + '\n')

def _merge(ui, repo, from_, to):
    commands.update(ui, repo, to)
    error = commands.merge(ui, repo, from_)
    if error:
        return True
    commands.commit(ui, repo, message='merged %s -> %s' % (from_, to))

cmdtable = {
    'pelo': (pelo_cmd, [], HELP_DOC),
}