# -*- coding: utf-8 -*-

from __future__ import (absolute_import, division, print_function,

unicode_literals)

#

# This file is part of `gitflow`.

# Copyright (c) 2010-2011 Vincent Driessen

# Copyright (c) 2012-2013 Hartmut Goebel

# Copyright (c) 2015 Christian Assing

# Distributed under a BSD-like license. For full terms see the file LICENSE.txt

#

import sys

import time

try:

from ConfigParser import NoSectionError, NoOptionError

except ImportError:

from configparser import NoSectionError, NoOptionError

from functools import wraps

import git

from git import (Git, Repo, InvalidGitRepositoryError, RemoteReference,

GitCommandError)

from gitflow.branches import BranchManager

from gitflow.util import itersubclasses

from gitflow.exceptions import (NotInitialized, BranchExistsError,

MergeConflict,

NoSuchRemoteError, NoSuchBranchError,

Usage, BadObjectError)

__copyright__ = "2010-2011 Vincent Driessen; 2012-2013 Hartmut Goebel; 2015 Christian Assing"

__license__ = "BSD"

def datetime_to_timestamp(d):

return time.mktime(d.timetuple()) + d.microsecond / 1e6

def requires_repo(f):

@wraps(f)

def _inner(self, *args, **kwargs):

if self.repo is None:

msg = 'This repo has not yet been initialized for git-flow.'

raise NotInitialized(msg)

return f(self, *args, **kwargs)

return _inner

def requires_initialized(f):

@wraps(f)

def _inner(self, *args, **kwargs):

if not self.is_initialized() or not self.master_name() in self.repo.branches or not self.develop_name() in self.repo.branches:

msg = 'This repo has not yet been initialized for git-flow.'

raise NotInitialized(msg)

return f(self, *args, **kwargs)

return _inner

def info(*texts):

for txt in texts:

print(txt)

def warn(*texts):

for txt in texts:

print(txt, file=sys.stderr)

class _NONE:

pass

class GitFlow(object):

"""

Creates a :class:`GitFlow` instance.

:param working_dir:

The directory where the Git repo is located. If not specified, the

current working directory is used.

When a :class:`GitFlow` class is instantiated, it auto-discovers all

subclasses of :class:`gitflow.branches.BranchManager`, so there is no

explicit registration required.

"""

def _discover_branch_managers(self):

managers = {}

for cls in itersubclasses(BranchManager):

# TODO: Initialize managers with the gitflow branch prefixes

managers[cls.identifier] = cls(self)

return managers

def __init__(self, working_dir='.'):

# Allow Repos to be passed in instead of strings

self.repo = None

if isinstance(working_dir, Repo):

self.working_dir = working_dir.working_dir

else:

self.working_dir = working_dir

self.git = Git(self.working_dir)

try:

self.repo = Repo(self.working_dir)

except InvalidGitRepositoryError:

pass

self.managers = self._discover_branch_managers()

self.defaults = {

'gitflow.branch.master': 'master',

'gitflow.branch.develop': 'develop',

'gitflow.prefix.versiontag': '',

'gitflow.origin': 'origin',

}

for identifier, manager in self.managers.items():

self.defaults['gitflow.prefix.%s' % identifier] = manager.DEFAULT_PREFIX

def _init_config(self, master=None, develop=None, prefixes={}, names={},

force_defaults=False):

for setting, default in self.defaults.items():

if force_defaults:

value = default

elif setting == 'gitflow.branch.master':

value = master

elif setting == 'gitflow.branch.develop':

value = develop

elif setting.startswith('gitflow.prefix.'):

name = setting[len('gitflow.prefix.'):]

value = prefixes.get(name, None)

else:

name = setting[len('gitflow.'):]

value = names.get(name, None)

if value is None:

value = self.get(setting, default)

self.set(setting, value)

def _init_initial_commit(self):

master = self.master_name()

if master in self.repo.branches:

# local `master` branch exists

return

elif self.origin_name(master) in self.repo.refs:

# the origin branch counterpart exists

origin = self.repo.refs[self.origin_name(master)]

branch = self.repo.create_head(master, origin)

branch.set_tracking_branch(origin)

elif self.repo.heads:

raise NotImplementedError(

'Local and remote branches exist, but neither %s nor %s' % (

master, self.origin_name(master)

))

else:

# Create 'master' branch

info('Creating branch %r' % master)

c = self.repo.index.commit('Initial commit', head=False)

self.repo.create_head(master, c)

def _init_develop_branch(self):

# assert master already exists

assert self.master_name() in self.repo.refs

develop = self.develop_name()

if develop in self.repo.branches:

# local `develop` branch exists, but do not switch there

return

if self.origin_name(develop) in self.repo.refs:

# the origin branch counterpart exists

origin = self.repo.refs[self.origin_name(develop)]

branch = self.repo.create_head(develop, origin)

branch.set_tracking_branch(origin)

else:

# Create 'develop' branch

info('Creating branch %r' % develop)

branch = self.repo.create_head(develop, self.master())

# switch to develop branch if its newly created

info('Switching to branch %s' % branch)

branch.checkout()

def _enforce_git_repo(self):

"""

Ensure a (maybe empty) repository exists we can work on.

This is to be used by the `init` sub-command.

"""

if self.repo is None:

self.git.init(self.working_dir)

self.repo = Repo(self.working_dir)

def init(self, master=None, develop=None, prefixes={}, names={},

force_defaults=False):

self._enforce_git_repo()

self._init_config(master, develop, prefixes, names, force_defaults)

self._init_initial_commit()

self._init_develop_branch()

return self

def is_initialized(self):

return (self.repo and

self.is_set('gitflow.branch.master') and

self.is_set('gitflow.branch.develop') and

self.is_set('gitflow.prefix.feature') and

self.is_set('gitflow.prefix.release') and

self.is_set('gitflow.prefix.hotfix') and

self.is_set('gitflow.prefix.support') and

self.is_set('gitflow.prefix.versiontag'))

def _parse_setting(self, setting):

groups = setting.split('.', 2)

if len(groups) == 2:

section, option = groups

elif len(groups) == 3:

section, subsection, option = groups

section = '%s "%s"' % (section, subsection)

else:

raise ValueError('Invalid setting name: %s' % setting)

return (section, option)

@requires_repo

def get(self, setting, default=_NONE):

section, option = self._parse_setting(setting)

try:

return self.repo.config_reader().get_value(section, option)

except (NoSectionError, NoOptionError):

if default is not _NONE:

return default

raise

def get_prefix(self, identifier):

return self._safe_get('gitflow.prefix.%s' % (identifier,))

@requires_repo

def set(self, setting, value):

section, option = self._parse_setting(setting)

writer = self.repo.config_writer()

writer.set_value(section, option, value)

writer.release()

del writer

def is_set(self, setting):

return self.get(setting, None) is not None

@requires_repo

def _safe_get(self, setting_name):

try:

return self.get(setting_name)

except (NoSectionError, NoOptionError):

raise NotInitialized('This repo has not yet been initialized.')

def master_name(self):

return self._safe_get('gitflow.branch.master')

def develop_name(self):

return self._safe_get('gitflow.branch.develop')

def origin_name(self, name=None):

origin = self.get('gitflow.origin', self.defaults['gitflow.origin'])

if name is not None:

return origin + '/' + name

else:

return origin

@requires_repo

def require_remote(self, name):

try:

return self.repo.remotes[name]

except IndexError:

raise NoSuchRemoteError(name)

def origin(self):

return self.require_remote(self.origin_name())

@requires_repo

def develop(self):

return self.repo.branches[self.develop_name()]

@requires_repo

def master(self):

return self.repo.branches[self.master_name()]

@requires_repo

def branch_names(self, remote=False):

if remote:

return [r.name

for r in self.repo.refs

if isinstance(r, RemoteReference)]

else:

return [r.name for r in self.repo.branches]

@requires_repo

def nameprefix_or_current(self, identifier, prefix):

"""

:param identifier:

The identifier for the type of branch to work on.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param prefix: If the empty, see if the current branch is of

type `identifier`. If so, returns the current branches

short name, otherwise raises :exc:`NoSuchBranchError`.

:returns:

If exactly one branch of type `identifier` starts with the

given name `prefix`, returns that branches short name.

Raises :exc:`NoSuchBranchError` in case no branch exists

with the given prefix, or :exc:`PrefixNotUniqueError` in

case multiple matches are found.

"""

repo = self.repo

manager = self.managers[identifier]

if not prefix:

if repo.active_branch.name.startswith(manager.prefix):

return manager.shorten(repo.active_branch.name)

else:

raise NoSuchBranchError(

'The current branch is no %s branch. '

'Please specify one explicitly.' % identifier)

return manager.shorten(manager.by_name_prefix(prefix).name)

@requires_repo

def name_or_current(self, identifier, name, must_exist=True):

"""

:param identifier:

The identifier for the type of branch to work on.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param name:

If the `name` is empty, see if the current branch is of

type `identifier`. If so, returns the current branches

short name, otherwise raises :exc:`NoSuchBranchError`.

:param must_exist: If `True` (the default), raises

:exc:`NoSuchBranchError` in case no branch exists with the

given `name`.

Otherwise return the `name` unchanged.

"""

repo = self.repo

manager = self.managers[identifier]

if not name:

if repo.active_branch.name.startswith(manager.prefix):

return manager.shorten(repo.active_branch.name)

else:

raise NoSuchBranchError(

'The current branch is no %s branch. '

'Please specify one explicitly.' % identifier)

elif must_exist and not manager.full_name(name) in (b.name for b in manager.list()):

raise NoSuchBranchError('There is no %s branch named %s.'

% (identifier, name))

return name

@requires_repo

def status(self):

result = []

for b in self.repo.branches:

tup = self.branch_info(b.name)

result.append(tup)

return result

@requires_repo

def branch_info(self, name):

active_branch = self.repo.active_branch

b = self.repo.heads[name]

return (name, b.commit.hexsha, b == active_branch)

@requires_repo

def is_dirty(self):

"""

Returns whether or not the current working directory contains

uncommitted changes.

"""

return self.repo.is_dirty()

@requires_repo

def has_staged_commits(self):

"""

Returns whether or not the current repo contains local changes

checked into the index but not committed.

"""

return len(self.repo.index.diff(self.repo.head.commit)) > 0

@requires_repo

def require_no_merge_conflict(self):

"""

Raises :exc:`MergeConflict` if the current working directory

contains a merge conflict.

"""

try:

git.Reference(self.repo, 'MERGE_HEAD', check_path=False).commit

# reference exists, so there is a merge conflict

raise MergeConflict()

except ValueError:

# no such reference, so there is no merge conflict

pass

def is_merged_into(self, commit, target_branch):

"""

Checks whether `commit` is successfully merged into branch

`target_branch`.

:param commit:

The commit or branch that ought to be merged. This may be

a full branch-name, a commit-hexsha or any of branch-,

head-, reference- or commit-object.

:param target_branch:

The branch which should contain the commit. This may be a

full branch-name, or any of branch-, head- or

reference-object.

"""

try:

commit = self.repo.rev_parse(str(commit))

except (git.BadObject, git.BadName):

raise BadObjectError(commit)

if isinstance(target_branch, git.RemoteReference):

target_branch = 'remotes/' + target_branch.name

elif isinstance(target_branch, git.SymbolicReference):

target_branch = target_branch.name

# :todo: implement this more efficiently

return target_branch in [

b.lstrip('* ')

for b in self.git.branch('-a', '--contains', commit).splitlines()]

def must_be_uptodate(self, branch, fetch):

remote_branch = self.origin_name(branch)

if remote_branch in self.branch_names(remote=True):

if fetch:

self.origin().fetch(branch)

self.require_branches_equal(branch, remote_branch)

@requires_repo

def _compare_branches(self, branch1, branch2):

"""

Tests whether branches and their 'origin' counterparts have

diverged and need merging first. It returns error codes to

provide more detail, like so:

0 Branch heads point to the same commit

1 First given branch needs fast-forwarding

2 Second given branch needs fast-forwarding

3 Branch needs a real merge

4 There is no merge base, i.e. the branches have no common ancestors

"""

try:

commit1 = self.repo.rev_parse(branch1)

commit2 = self.repo.rev_parse(branch2)

except (git.BadObject, git.BadName) as e:

raise NoSuchBranchError(e.args[0])

if commit1 == commit2:

return 0

try:

# merge_base() returns a list of Commit objects

# this list will have at max one Commit

# or it will be empty if no common merge base exists

base = self.repo.merge_base(commit1, commit2)[0]

except (GitCommandError, IndexError):

return 4

if base == commit1:

return 1

elif base == commit2:

return 2

else:

return 3

@requires_repo

def require_branches_equal(self, branch1, branch2):

status = self._compare_branches(branch1, branch2)

if status == 0:

# branches are equal

return

else:

warn("Branches '%s' and '%s' have diverged." % (branch1, branch2))

if status == 1:

raise SystemExit("And branch '%s' may be fast-forwarded." % branch1)

elif status == 2:

# Warn here, since there is no harm in being ahead

warn("And local branch '%s' is ahead of '%s'." % (branch1, branch2))

else:

raise SystemExit("Branches need merging first.")

@requires_repo

def start_transaction(self, message=None):

if message:

info(message)

@requires_initialized

def tag(self, tagname, commit, message=None, sign=False, signingkey=None):

kwargs = {}

if sign:

kwargs['s'] = True

if signingkey:

kwargs['u'] = signingkey

self.repo.create_tag(tagname, commit, message=message or None, **kwargs)

#

# ====== sub commands =====

#

@requires_repo

def list(self, identifier, arg0_name, verbose, use_tagname):

"""

List the all branches of the given type. If there are not

branches of this type, raises :exc:`Usage` with an

explanation on how to start a branch of this type.

:param identifier:

The identifier for the type of branch to work on.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param arg0_name:

Name of the first argument for the command line to be put

into the explanation on how to start a branch of this

type. This typically is `name` or `version`.

:param verbose:

If True, give more information about the state of the

branch: Whether it's ahead or behind it's default base,

may be rebased, etc.

:param use_tagname:

If True, try to describe the state based on the next tag.

"""

repo = self.repo

manager = self.managers[identifier]

branches = manager.list()

if not branches:

raise Usage(

'No %s branches exist.' % identifier,

'You can start a new %s branch with the command:' % identifier,

' git flow %s start <%s> [<base>]' % (identifier, arg0_name)

)

# determine the longest branch name

width = max(len(b.name) for b in branches) - len(manager.prefix) + 1

basebranch_sha = repo.branches[manager.default_base()].commit.hexsha

for branch in branches:

if repo.active_branch == branch:

prefix = '* '

else:

prefix = ' '

name = manager.shorten(branch.name)

extra_info = ''

if verbose:

name = name.ljust(width)

branch_sha = branch.commit.hexsha

base_sha = repo.git.merge_base(basebranch_sha, branch_sha)

if branch_sha == basebranch_sha:

extra_info = '(no commits yet)'

elif use_tagname:

try:

extra_info = self.git.name_rev('--tags', '--name-only',

'--no-undefined', base_sha)

extra_info = '(based on %s)' % extra_info

except GitCommandError:

pass

if not extra_info:

if base_sha == branch_sha:

extra_info = '(is behind %s, may ff)' % manager.default_base()

elif base_sha == basebranch_sha:

extra_info = '(based on latest %s)' % manager.default_base()

else:

extra_info = '(may be rebased)'

info(prefix + name + extra_info)

@requires_initialized

def create(self, identifier, name, base, fetch):

"""

Creates a branch of the given type, with the given short name.

:param identifier:

The identifier for the type of branch to create.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param name:

The friendly (short) name to create.

:param base:

The alternative base to branch off from. If not given, the default

base for the given branch type is used.

:returns:

The newly created :class:`git.refs.Head` branch.

"""

return self.managers[identifier].create(name, base, fetch=fetch)

@requires_initialized

def finish(self, identifier, name, fetch, rebase, keep, force_delete,

tagging_info):

"""

Finishes a branch of the given type, with the given short name.

:param identifier:

The identifier for the type of branch to finish.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param name:

The friendly (short) name to finish.

"""

mgr = self.managers[identifier]

branch = mgr.by_name_prefix(name)

try:

self.require_no_merge_conflict()

except MergeConflict as e:

raise Usage(e,

"You can then complete the finish by running it again:",

" git flow %s finish %s" % (identifier, name)

)

return mgr.finish(mgr.shorten(branch.name), fetch=fetch, rebase=rebase,

keep=keep, force_delete=force_delete,

tagging_info=tagging_info)

@requires_initialized

def checkout(self, identifier, name):

"""

Checkout a branch of the given type, with the given short name.

:param identifier:

The identifier for the type of branch to checkout.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param name:

The friendly (short) name to checkout.

:returns:

The checked out :class:`git.refs.Head` branch.

"""

mgr = self.managers[identifier]

branch = mgr.by_name_prefix(name)

return branch.checkout()

@requires_initialized

def diff(self, identifier, name):

"""

Print the diff of changes since this branch branched off.

:param identifier:

The identifier for the type of branch to work on.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param name:

The friendly (short) name to work on.

"""

mgr = self.managers[identifier]

full_name = mgr.full_name(name)

base = self.git.merge_base(mgr.default_base(), full_name)

print(self.git.diff('%s..%s' % (base, full_name)))

@requires_initialized

def rebase(self, identifier, name, interactive):

"""

Rebase a branch of the given type, with the given short name,

on top of it's default base.

:param identifier:

The identifier for the type of branch to rebase.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param name:

The friendly (short) name to rebase.

:param interactive:

If True, do an interactive rebase.

"""

warn("Will try to rebase %s branch '%s' ..." % (identifier, name))

mgr = self.managers[identifier]

mgr.full_name(name)

# :todo: require_clean_working_tree

self.checkout(identifier, name)

args = []

if interactive:

args.append('-i')

args.append(mgr.default_base())

self.git.rebase(*args)

@requires_initialized

def publish(self, identifier, name):

"""

Publish a branch of the given type, with the given short name,

to `origin` (or whatever is configured as `remote` for gitflow.)

:param identifier:

The identifier for the type of branch to publish.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param name:

The friendly (short) name to publish.

:returns:

The full name of the published branch.

"""

repo = self.repo

mgr = self.managers[identifier]

# sanity checks

# :todo: require_clean_working_tree

full_name = mgr.full_name(name)

remote_name = self.origin_name(full_name)

if full_name not in repo.branches:

raise NoSuchBranchError(full_name)

if remote_name in repo.refs:

raise BranchExistsError(remote_name)

# :todo: check if full_name already has a tracking branch

# :todo: check if full_name already has the same tracking branch

# create remote branch

origin = self.origin()

info = origin.push('%s:refs/heads/%s' % (full_name, full_name))[0]

origin.fetch()

# configure remote tracking

repo.branches[full_name].set_tracking_branch(info.remote_ref)

return full_name

@requires_initialized

def pull(self, identifier, remote, name):

"""

Pull a branch of the given type, with the given short name,

from the given remote peer.

:param identifier:

The identifier for the type of branch to pull.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param remote:

The remote to pull from. This must have been configured by

`git remote add ...`.

:param name:

The friendly (short) name to pull.

"""

def avoid_accidental_cross_branch_action(branch_name):

current_branch = repo.active_branch

if branch_name != current_branch.name:

warn("Trying to pull from '%s' while currently on branch '%s'."

% (branch_name, current_branch))

raise SystemExit("To avoid unintended merges, git-flow aborted.")

repo = self.repo

mgr = self.managers[identifier]

full_name = mgr.full_name(name)

# To avoid accidentally merging different feature branches

# into each other, die if the current feature branch differs

# from the requested $NAME argument.

if repo.active_branch.name.startswith(self.get_prefix(identifier)):

# We are on a local `identifier` branch already, so `full_name`

# must be equal to the current branch.

avoid_accidental_cross_branch_action(full_name)

# :todo: require_clean_working_tree

if full_name in self.repo.branches:

# Again, avoid accidental merges

avoid_accidental_cross_branch_action(full_name)

# We already have a local branch called like this, so

# simply pull the remote changes in

self.require_remote(remote).pull(full_name)

# :fixme: why is the branch not checked out here?

info("Pulled %s's changes into %s." % (remote, full_name))

else:

# Setup the non-tracking local branch clone for the first time

self.require_remote(remote).fetch(full_name + ':' + full_name)

repo.heads[full_name].checkout()

info("Created local branch %s based on %s's %s."

% (full_name, remote, full_name))

@requires_initialized

def track(self, identifier, name):

"""

Track a branch of the given type, with the given short name,

from `origin` (or whatever is configured as `remote` for

gitflow.)

:param identifier:

The identifier for the type of branch to track.

A :class:`BranchManager <git.branches.BranchManager>` for the given

identifier must exist in the :attr:`self.managers`.

:param name:

The friendly (short) name to track.

:param base:

The alternative base to branch off from. If not given, the default

base for the given branch type is used.

:returns:

The newly created :class:`git.refs.Head` branch.

"""

repo = self.repo

mgr = self.managers[identifier]

# sanity checks

# :todo: require_clean_working_tree

full_name = mgr.full_name(name)

if full_name in repo.branches:

raise BranchExistsError(full_name)

self.origin().fetch(full_name)

remote_branch = self.origin().refs[full_name]

branch = repo.create_head(full_name, remote_branch)

branch.set_tracking_branch(remote_branch)

return branch.checkout()