"""Class providing detailed progress information to the caller who should

CLONE, FETCH, UPDWKTREE = [1 << x for x in range(RemoteProgress._num_op_codes, RemoteProgress._num_op_codes + 3)]

_num_op_codes = RemoteProgress._num_op_codes + 3

"""Implements access to a git submodule. They are special in that their sha

_id_attribute_ = "name"

k_modules_file = '.gitmodules'

k_head_option = 'branch'

k_head_default = 'master'

k_default_mode = stat.S_IFDIR | stat.S_IFLNK # submodules are directories with link-status

# this is a bogus type for base class compatibility

type = 'submodule'

__slots__ = ('_parent_commit', '_url', '_branch_path', '_name', '__weakref__')

_cache_attrs = ('path', '_url', '_branch_path')

def __init__(self, repo, binsha, mode=None, path=None, name=None, parent_commit=None, url=None, branch_path=None):

"""Initialize this instance with its attributes. We only document the ones

super(Submodule, self).__init__(repo, binsha, mode, path)

self.size = 0

self._parent_commit = parent_commit

if url is not None:

self._url = url

if branch_path is not None:

assert isinstance(branch_path, string_types)

self._branch_path = branch_path

if name is not None:

self._name = name

def _set_cache_(self, attr):

if attr in ('path', '_url', '_branch_path'):

reader = self.config_reader()

# default submodule values

try:

self.path = reader.get('path')

except cp.NoSectionError:

raise ValueError("This submodule instance does not exist anymore in '%s' file"

% osp.join(self.repo.working_tree_dir, '.gitmodules'))

# end

self._url = reader.get('url')

# git-python extension values - optional

self._branch_path = reader.get_value(self.k_head_option, git.Head.to_full_path(self.k_head_default))

elif attr == '_name':

raise AttributeError("Cannot retrieve the name of a submodule if it was not set initially")

else:

super(Submodule, self)._set_cache_(attr)

# END handle attribute name

def _get_intermediate_items(self, item):

""":return: all the submodules of our module repository"""

try:

return type(self).list_items(item.module())

except InvalidGitRepositoryError:

return []

# END handle intermediate items

@classmethod

def _need_gitfile_submodules(cls, git):

return git.version_info[:3] >= (1, 7, 5)

def __eq__(self, other):

"""Compare with another submodule"""

# we may only compare by name as this should be the ID they are hashed with

# Otherwise this type wouldn't be hashable

# return self.path == other.path and self.url == other.url and super(Submodule, self).__eq__(other)

return self._name == other._name

def __ne__(self, other):

"""Compare with another submodule for inequality"""

return not (self == other)

def __hash__(self):

"""Hash this instance using its logical id, not the sha"""

return hash(self._name)

def __str__(self):

return self._name

def __repr__(self):

return "git.%s(name=%s, path=%s, url=%s, branch_path=%s)"\

% (type(self).__name__, self._name, self.path, self.url, self.branch_path)

@classmethod

def _config_parser(cls, repo, parent_commit, read_only):

""":return: Config Parser constrained to our submodule in read or write mode

parent_matches_head = True

if parent_commit is not None:

try:

parent_matches_head = repo.head.commit == parent_commit

except ValueError:

# We are most likely in an empty repository, so the HEAD doesn't point to a valid ref

pass

# end handle parent_commit

if not repo.bare and parent_matches_head:

fp_module = osp.join(repo.working_tree_dir, cls.k_modules_file)

else:

assert parent_commit is not None, "need valid parent_commit in bare repositories"

try:

fp_module = cls._sio_modules(parent_commit)

except KeyError:

raise IOError("Could not find %s file in the tree of parent commit %s" %

(cls.k_modules_file, parent_commit))

# END handle exceptions

# END handle non-bare working tree

if not read_only and (repo.bare or not parent_matches_head):

raise ValueError("Cannot write blobs of 'historical' submodule configurations")

# END handle writes of historical submodules

return SubmoduleConfigParser(fp_module, read_only=read_only)

def _clear_cache(self):

# clear the possibly changed values

for name in self._cache_attrs:

try:

delattr(self, name)

except AttributeError:

pass

# END try attr deletion

# END for each name to delete

@classmethod

def _sio_modules(cls, parent_commit):

""":return: Configuration file as BytesIO - we only access it through the respective blob's data"""

sio = BytesIO(parent_commit.tree[cls.k_modules_file].data_stream.read())

sio.name = cls.k_modules_file

return sio

def _config_parser_constrained(self, read_only):

""":return: Config Parser constrained to our submodule in read or write mode"""

try:

pc = self.parent_commit

except ValueError:

pc = None

# end handle empty parent repository

parser = self._config_parser(self.repo, pc, read_only)

parser.set_submodule(self)

return SectionConstraint(parser, sm_section(self.name))

@classmethod

def _module_abspath(cls, parent_repo, path, name):

if cls._need_gitfile_submodules(parent_repo.git):

return osp.join(parent_repo.git_dir, 'modules', name)

else:

return osp.join(parent_repo.working_tree_dir, path)

# end

@classmethod

def _clone_repo(cls, repo, url, path, name, **kwargs):

""":return: Repo instance of newly cloned repository

module_abspath = cls._module_abspath(repo, path, name)

module_checkout_path = module_abspath

if cls._need_gitfile_submodules(repo.git):

kwargs['separate_git_dir'] = module_abspath

module_abspath_dir = osp.dirname(module_abspath)

if not osp.isdir(module_abspath_dir):

os.makedirs(module_abspath_dir)

module_checkout_path = osp.join(repo.working_tree_dir, path)

# end

clone = git.Repo.clone_from(url, module_checkout_path, **kwargs)

if cls._need_gitfile_submodules(repo.git):

cls._write_git_file_and_module_config(module_checkout_path, module_abspath)

# end

return clone

@classmethod

def _to_relative_path(cls, parent_repo, path):

""":return: a path guaranteed to be relative to the given parent-repository

path = to_native_path_linux(path)

if path.endswith('/'):

path = path[:-1]

# END handle trailing slash

if osp.isabs(path):

working_tree_linux = to_native_path_linux(parent_repo.working_tree_dir)

if not path.startswith(working_tree_linux):

raise ValueError("Submodule checkout path '%s' needs to be within the parents repository at '%s'"

% (working_tree_linux, path))

path = path[len(working_tree_linux.rstrip('/')) + 1:]

if not path:

raise ValueError("Absolute submodule path '%s' didn't yield a valid relative path" % path)

# end verify converted relative path makes sense

# end convert to a relative path

return path

@classmethod

def _write_git_file_and_module_config(cls, working_tree_dir, module_abspath):

"""Writes a .git file containing a (preferably) relative path to the actual git module repository.

git_file = osp.join(working_tree_dir, '.git')

rela_path = osp.relpath(module_abspath, start=working_tree_dir)

if is_win:

if osp.isfile(git_file):

os.remove(git_file)

with open(git_file, 'wb') as fp:

fp.write(("gitdir: %s" % rela_path).encode(defenc))

with GitConfigParser(osp.join(module_abspath, 'config'),

read_only=False, merge_includes=False) as writer:

writer.set_value('core', 'worktree',

to_native_path_linux(osp.relpath(working_tree_dir, start=module_abspath)))

#{ Edit Interface

@classmethod

def add(cls, repo, name, path, url=None, branch=None, no_checkout=False):

"""Add a new submodule to the given repository. This will alter the index

if repo.bare:

raise InvalidGitRepositoryError("Cannot add submodules to bare repositories")

# END handle bare repos

path = cls._to_relative_path(repo, path)

# assure we never put backslashes into the url, as some operating systems

# like it ...

if url is not None:

url = to_native_path_linux(url)

# END assure url correctness

# INSTANTIATE INTERMEDIATE SM

sm = cls(repo, cls.NULL_BIN_SHA, cls.k_default_mode, path, name, url='invalid-temporary')

if sm.exists():

# reretrieve submodule from tree

try:

sm = repo.head.commit.tree[path]

sm._name = name

return sm

except KeyError:

# could only be in index

index = repo.index

entry = index.entries[index.entry_key(path, 0)]

sm.binsha = entry.binsha

return sm

# END handle exceptions

# END handle existing

# fake-repo - we only need the functionality on the branch instance

br = git.Head(repo, git.Head.to_full_path(str(branch) or cls.k_head_default))

has_module = sm.module_exists()

branch_is_default = branch is None

if has_module and url is not None:

if url not in [r.url for r in sm.module().remotes]:

raise ValueError(

"Specified URL '%s' does not match any remote url of the repository at '%s'" % (url, sm.abspath))

# END check url

# END verify urls match

mrepo = None

if url is None:

if not has_module:

raise ValueError("A URL was not given and existing repository did not exsit at %s" % path)

# END check url

mrepo = sm.module()

urls = [r.url for r in mrepo.remotes]

if not urls:

raise ValueError("Didn't find any remote url in repository at %s" % sm.abspath)

# END verify we have url

url = urls[0]

else:

# clone new repo

kwargs = {'n': no_checkout}

if not branch_is_default:

kwargs['b'] = br.name

# END setup checkout-branch

# _clone_repo(cls, repo, url, path, name, **kwargs):

mrepo = cls._clone_repo(repo, url, path, name, **kwargs)

# END verify url

## See #525 for ensuring git urls in config-files valid under Windows.

url = Git.polish_url(url)

# It's important to add the URL to the parent config, to let `git submodule` know.

# otherwise there is a '-' character in front of the submodule listing

# a38efa84daef914e4de58d1905a500d8d14aaf45 mymodule (v0.9.0-1-ga38efa8)

# -a38efa84daef914e4de58d1905a500d8d14aaf45 submodules/intermediate/one

with sm.repo.config_writer() as writer:

writer.set_value(sm_section(name), 'url', url)

# update configuration and index

index = sm.repo.index

with sm.config_writer(index=index, write=False) as writer:

writer.set_value('url', url)

writer.set_value('path', path)

sm._url = url

if not branch_is_default:

# store full path

writer.set_value(cls.k_head_option, br.path)

sm._branch_path = br.path

# we deliberately assume that our head matches our index !

sm.binsha = mrepo.head.commit.binsha

index.add([sm], write=True)

return sm

def update(self, recursive=False, init=True, to_latest_revision=False, progress=None, dry_run=False,

force=False, keep_going=False):

"""Update the repository of this submodule to point to the checkout

if self.repo.bare:

return self

# END pass in bare mode

if progress is None:

progress = UpdateProgress()

# END handle progress

prefix = ''

if dry_run:

prefix = "DRY-RUN: "

# END handle prefix

# to keep things plausible in dry-run mode

if dry_run:

mrepo = None

# END init mrepo

try:

# ASSURE REPO IS PRESENT AND UPTODATE

#####################################

try:

mrepo = self.module()

rmts = mrepo.remotes

len_rmts = len(rmts)

for i, remote in enumerate(rmts):

op = FETCH

if i == 0:

op |= BEGIN

# END handle start

progress.update(op, i, len_rmts, prefix + "Fetching remote %s of submodule %r"

% (remote, self.name))

#===============================

if not dry_run:

remote.fetch(progress=progress)

# END handle dry-run

#===============================

if i == len_rmts - 1:

op |= END

# END handle end

progress.update(op, i, len_rmts, prefix + "Done fetching remote of submodule %r" % self.name)

# END fetch new data

except InvalidGitRepositoryError:

if not init:

return self

# END early abort if init is not allowed

# there is no git-repository yet - but delete empty paths

checkout_module_abspath = self.abspath

if not dry_run and osp.isdir(checkout_module_abspath):

try:

os.rmdir(checkout_module_abspath)

except OSError:

raise OSError("Module directory at %r does already exist and is non-empty"

% checkout_module_abspath)

# END handle OSError

# END handle directory removal

# don't check it out at first - nonetheless it will create a local

# branch according to the remote-HEAD if possible

progress.update(BEGIN | CLONE, 0, 1, prefix + "Cloning url '%s' to '%s' in submodule %r" %

(self.url, checkout_module_abspath, self.name))

if not dry_run:

mrepo = self._clone_repo(self.repo, self.url, self.path, self.name, n=True)

# END handle dry-run

progress.update(END | CLONE, 0, 1, prefix + "Done cloning to %s" % checkout_module_abspath)

if not dry_run:

# see whether we have a valid branch to checkout

try:

# find a remote which has our branch - we try to be flexible

remote_branch = find_first_remote_branch(mrepo.remotes, self.branch_name)

local_branch = mkhead(mrepo, self.branch_path)

# have a valid branch, but no checkout - make sure we can figure

# that out by marking the commit with a null_sha

local_branch.set_object(Object(mrepo, self.NULL_BIN_SHA))

# END initial checkout + branch creation

# make sure HEAD is not detached

mrepo.head.set_reference(local_branch, logmsg="submodule: attaching head to %s" % local_branch)

mrepo.head.ref.set_tracking_branch(remote_branch)

except (IndexError, InvalidGitRepositoryError):

log.warn("Failed to checkout tracking branch %s", self.branch_path)

# END handle tracking branch

# NOTE: Have to write the repo config file as well, otherwise

# the default implementation will be offended and not update the repository

# Maybe this is a good way to assure it doesn't get into our way, but

# we want to stay backwards compatible too ... . Its so redundant !

with self.repo.config_writer() as writer:

writer.set_value(sm_section(self.name), 'url', self.url)

# END handle dry_run

# END handle initialization

# DETERMINE SHAS TO CHECKOUT

############################

binsha = self.binsha

hexsha = self.hexsha

if mrepo is not None:

# mrepo is only set if we are not in dry-run mode or if the module existed

is_detached = mrepo.head.is_detached

# END handle dry_run

if mrepo is not None and to_latest_revision:

msg_base = "Cannot update to latest revision in repository at %r as " % mrepo.working_dir

if not is_detached:

rref = mrepo.head.ref.tracking_branch()

if rref is not None:

rcommit = rref.commit

binsha = rcommit.binsha

hexsha = rcommit.hexsha

else:

log.error("%s a tracking branch was not set for local branch '%s'", msg_base, mrepo.head.ref)

# END handle remote ref

else:

log.error("%s there was no local tracking branch", msg_base)

# END handle detached head

# END handle to_latest_revision option

# update the working tree

# handles dry_run

if mrepo is not None and mrepo.head.commit.binsha != binsha:

# We must assure that our destination sha (the one to point to) is in the future of our current head.

# Otherwise, we will reset changes that might have been done on the submodule, but were not yet pushed

# We also handle the case that history has been rewritten, leaving no merge-base. In that case

# we behave conservatively, protecting possible changes the user had done

may_reset = True

if mrepo.head.commit.binsha != self.NULL_BIN_SHA:

base_commit = mrepo.merge_base(mrepo.head.commit, hexsha)

if len(base_commit) == 0 or base_commit[0].hexsha == hexsha:

if force:

msg = "Will force checkout or reset on local branch that is possibly in the future of"

msg += "the commit it will be checked out to, effectively 'forgetting' new commits"

log.debug(msg)

else:

msg = "Skipping %s on branch '%s' of submodule repo '%s' as it contains un-pushed commits"

msg %= (is_detached and "checkout" or "reset", mrepo.head, mrepo)

log.info(msg)

may_reset = False

# end handle force

# end handle if we are in the future

if may_reset and not force and mrepo.is_dirty(index=True, working_tree=True, untracked_files=True):

raise RepositoryDirtyError(mrepo, "Cannot reset a dirty repository")

# end handle force and dirty state

# end handle empty repo

# end verify future/past

progress.update(BEGIN | UPDWKTREE, 0, 1, prefix +

"Updating working tree at %s for submodule %r to revision %s"

% (self.path, self.name, hexsha))

if not dry_run and may_reset:

if is_detached:

# NOTE: for now we force, the user is no supposed to change detached

# submodules anyway. Maybe at some point this becomes an option, to

# properly handle user modifications - see below for future options

# regarding rebase and merge.

mrepo.git.checkout(hexsha, force=force)

else:

mrepo.head.reset(hexsha, index=True, working_tree=True)

# END handle checkout

# if we may reset/checkout

progress.update(END | UPDWKTREE, 0, 1, prefix + "Done updating working tree for submodule %r"

% self.name)

# END update to new commit only if needed

except Exception as err:

if not keep_going:

raise

log.error(str(err))

# end handle keep_going

# HANDLE RECURSION

##################

if recursive:

# in dry_run mode, the module might not exist

if mrepo is not None:

for submodule in self.iter_items(self.module()):

submodule.update(recursive, init, to_latest_revision, progress=progress, dry_run=dry_run,

force=force, keep_going=keep_going)

# END handle recursive update

# END handle dry run

# END for each submodule

return self

@unbare_repo

def move(self, module_path, configuration=True, module=True):

"""Move the submodule to a another module path. This involves physically moving

if module + configuration < 1:

raise ValueError("You must specify to move at least the module or the configuration of the submodule")

# END handle input

module_checkout_path = self._to_relative_path(self.repo, module_path)

# VERIFY DESTINATION

if module_checkout_path == self.path:

return self

# END handle no change

module_checkout_abspath = join_path_native(self.repo.working_tree_dir, module_checkout_path)

if osp.isfile(module_checkout_abspath):

raise ValueError("Cannot move repository onto a file: %s" % module_checkout_abspath)

# END handle target files

index = self.repo.index

tekey = index.entry_key(module_checkout_path, 0)

# if the target item already exists, fail

if configuration and tekey in index.entries:

raise ValueError("Index entry for target path did already exist")

# END handle index key already there

# remove existing destination

if module:

if osp.exists(module_checkout_abspath):

if len(os.listdir(module_checkout_abspath)):

raise ValueError("Destination module directory was not empty")

# END handle non-emptiness

if osp.islink(module_checkout_abspath):

os.remove(module_checkout_abspath)

else:

os.rmdir(module_checkout_abspath)

# END handle link

else:

# recreate parent directories

# NOTE: renames() does that now

pass

# END handle existence

# END handle module

# move the module into place if possible

cur_path = self.abspath

renamed_module = False

if module and osp.exists(cur_path):

os.renames(cur_path, module_checkout_abspath)

renamed_module = True

if osp.isfile(osp.join(module_checkout_abspath, '.git')):

module_abspath = self._module_abspath(self.repo, self.path, self.name)

self._write_git_file_and_module_config(module_checkout_abspath, module_abspath)

# end handle git file rewrite

# END move physical module

# rename the index entry - have to manipulate the index directly as

# git-mv cannot be used on submodules ... yeah

previous_sm_path = self.path

try:

if configuration:

try:

ekey = index.entry_key(self.path, 0)

entry = index.entries[ekey]

del(index.entries[ekey])

nentry = git.IndexEntry(entry[:3] + (module_checkout_path,) + entry[4:])

index.entries[tekey] = nentry

except KeyError:

raise InvalidGitRepositoryError("Submodule's entry at %r did not exist" % (self.path))

# END handle submodule doesn't exist

# update configuration

with self.config_writer(index=index) as writer: # auto-write

writer.set_value('path', module_checkout_path)

self.path = module_checkout_path

# END handle configuration flag

except Exception:

if renamed_module:

os.renames(module_checkout_abspath, cur_path)

# END undo module renaming

raise

# END handle undo rename

# Auto-rename submodule if it's name was 'default', that is, the checkout directory

if previous_sm_path == self.name:

self.rename(module_checkout_path)

# end

return self

@unbare_repo

def remove(self, module=True, force=False, configuration=True, dry_run=False):

"""Remove this submodule from the repository. This will remove our entry

if not (module or configuration):

raise ValueError("Need to specify to delete at least the module, or the configuration")

# END handle parameters

# Recursively remove children of this submodule

nc = 0

for csm in self.children():

nc += 1

csm.remove(module, force, configuration, dry_run)

del(csm)

# end

if configuration and not dry_run and nc > 0:

# Assure we don't leave the parent repository in a dirty state, and commit our changes

# It's important for recursive, unforced, deletions to work as expected

self.module().index.commit("Removed at least one of child-modules of '%s'" % self.name)

# end handle recursion

# DELETE REPOSITORY WORKING TREE

################################

if module and self.module_exists():

mod = self.module()

git_dir = mod.git_dir

if force:

# take the fast lane and just delete everything in our module path

# TODO: If we run into permission problems, we have a highly inconsistent

# state. Delete the .git folders last, start with the submodules first

mp = self.abspath

method = None

if osp.islink(mp):

method = os.remove

elif osp.isdir(mp):

method = rmtree

elif osp.exists(mp):

raise AssertionError("Cannot forcibly delete repository as it was neither a link, nor a directory")

# END handle brutal deletion

if not dry_run:

assert method

method(mp)

# END apply deletion method

else:

# verify we may delete our module

if mod.is_dirty(index=True, working_tree=True, untracked_files=True):

raise InvalidGitRepositoryError(

"Cannot delete module at %s with any modifications, unless force is specified"

% mod.working_tree_dir)

# END check for dirt

# figure out whether we have new commits compared to the remotes

# NOTE: If the user pulled all the time, the remote heads might

# not have been updated, so commits coming from the remote look

# as if they come from us. But we stay strictly read-only and

# don't fetch beforehand.

for remote in mod.remotes:

num_branches_with_new_commits = 0

rrefs = remote.refs

for rref in rrefs:

num_branches_with_new_commits += len(mod.git.cherry(rref)) != 0

# END for each remote ref

# not a single remote branch contained all our commits

if len(rrefs) and num_branches_with_new_commits == len(rrefs):

raise InvalidGitRepositoryError(

"Cannot delete module at %s as there are new commits" % mod.working_tree_dir)

# END handle new commits

# have to manually delete references as python's scoping is

# not existing, they could keep handles open ( on windows this is a problem )

if len(rrefs):

del(rref)

# END handle remotes

del(rrefs)

del(remote)

# END for each remote

# finally delete our own submodule

if not dry_run:

self._clear_cache()

wtd = mod.working_tree_dir

del(mod) # release file-handles (windows)

import gc

gc.collect()

try:

rmtree(wtd)

except Exception as ex:

if HIDE_WINDOWS_KNOWN_ERRORS:

raise SkipTest("FIXME: fails with: PermissionError\n %s", ex)

else:

raise

# END delete tree if possible

# END handle force

if not dry_run and osp.isdir(git_dir):

self._clear_cache()

try:

rmtree(git_dir)

except Exception as ex:

if HIDE_WINDOWS_KNOWN_ERRORS:

raise SkipTest("FIXME: fails with: PermissionError\n %s", ex)

else:

raise

# end handle separate bare repository

# END handle module deletion

# void our data not to delay invalid access

if not dry_run:

self._clear_cache()

# DELETE CONFIGURATION

######################

if configuration and not dry_run:

# first the index-entry

parent_index = self.repo.index

try:

del(parent_index.entries[parent_index.entry_key(self.path, 0)])

except KeyError:

pass

# END delete entry

parent_index.write()

# now git config - need the config intact, otherwise we can't query

# information anymore

with self.repo.config_writer() as writer:

writer.remove_section(sm_section(self.name))

with self.config_writer() as writer:

writer.remove_section()

# END delete configuration

return self

def set_parent_commit(self, commit, check=True):

"""Set this instance to use the given commit whose tree is supposed to

if commit is None:

self._parent_commit = None

return self

# end handle None

pcommit = self.repo.commit(commit)

pctree = pcommit.tree

if self.k_modules_file not in pctree:

raise ValueError("Tree of commit %s did not contain the %s file" % (commit, self.k_modules_file))

# END handle exceptions

prev_pc = self._parent_commit

self._parent_commit = pcommit

if check:

parser = self._config_parser(self.repo, self._parent_commit, read_only=True)

if not parser.has_section(sm_section(self.name)):

self._parent_commit = prev_pc

raise ValueError("Submodule at path %r did not exist in parent commit %s" % (self.path, commit))

# END handle submodule did not exist

# END handle checking mode

# update our sha, it could have changed

# If check is False, we might see a parent-commit that doesn't even contain the submodule anymore.

# in that case, mark our sha as being NULL

try:

self.binsha = pctree[self.path].binsha

except KeyError:

self.binsha = self.NULL_BIN_SHA

# end

self._clear_cache()

return self

@unbare_repo

def config_writer(self, index=None, write=True):

""":return: a config writer instance allowing you to read and write the data

writer = self._config_parser_constrained(read_only=False)

if index is not None:

writer.config._index = index

writer.config._auto_write = write

return writer

@unbare_repo

def rename(self, new_name):

"""Rename this submodule

if self.name == new_name:

return self

# .git/config

with self.repo.config_writer() as pw:

# As we ourselves didn't write anything about submodules into the parent .git/config,

# we will not require it to exist, and just ignore missing entries.