"join_path_native", "Stats", "IndexFileSHA1Writer", "Iterable", "IterableList",

"BlockingLockFile", "LockFile", 'Actor', 'get_user_id', 'assure_directory_exists',

'RemoteProgress', 'CallableRemoteProgress', 'rmtree', 'unbare_repo',

"""Methods with this decorator raise InvalidGitRepositoryError if they

@wraps(func)

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

if self.repo.bare:

raise InvalidGitRepositoryError("Method '%s' cannot operate on bare repositories" % func.__name__)

# END bare method

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

# END wrapper

old_dir = os.getcwd()

os.chdir(new_dir)

try:

yield new_dir

finally:

"""Remove the given recursively.

def onerror(func, path, exc_info):

# Is the error an access error ?

os.chmod(path, stat.S_IWUSR)

try:

func(path) # Will scream if still not possible to delete.

except Exception as ex:

if HIDE_WINDOWS_KNOWN_ERRORS:

raise SkipTest("FIXME: fails with: PermissionError\n {}".format(ex)) from ex

raise

"""Ensure file deleted also on *Windows* where read-only files need special treatment."""

if osp.isfile(path):

if is_win:

os.chmod(path, 0o777)

"""Copy all data from the source stream into the destination stream in chunks

br = 0

while True:

chunk = source.read(chunk_size)

destination.write(chunk)

br += len(chunk)

if len(chunk) < chunk_size:

break

# END reading output stream

"""Join path tokens together similar to osp.join, but always use

path = a

for b in p:

if not b:

continue

if b.startswith('/'):

path += b[1:]

elif path == '' or path.endswith('/'):

path += b

else:

path += '/' + b

# END for each path token to add

"""

"""Assure that the directory pointed to by path exists.

if is_file:

path = osp.dirname(path)

# END handle file

if not osp.isdir(path):

os.makedirs(path, exist_ok=True)

return True

PATHEXT = os.environ.get('PATHEXT', None)

return tuple(p.upper() for p in PATHEXT.split(os.pathsep)) \

if PATHEXT \

# From: http://stackoverflow.com/a/377028/548792

winprog_exts = _get_exe_extensions()

def is_exec(fpath):

return osp.isfile(fpath) and os.access(fpath, os.X_OK) and (

os.name != 'nt' or not winprog_exts or any(fpath.upper().endswith(ext)

for ext in winprog_exts))

progs = []

if not path:

path = os.environ["PATH"]

for folder in path.split(os.pathsep):

folder = folder.strip('"')

if folder:

exe_path = osp.join(folder, program)

for f in [exe_path] + ['%s%s' % (exe_path, e) for e in winprog_exts]:

if is_exec(f):

progs.append(f)

if osp.isabs(path) and not drive:

## Invoked from `cygpath()` directly with `D:Apps\123`?

# It's an error, leave it alone just slashes)

p = path

else:

p = path and osp.normpath(osp.expandvars(osp.expanduser(path)))

if osp.isabs(p):

if drive:

# Confusing, maybe a remote system should expand vars.

p = path

else:

p = cygpath(p)

elif drive:

p = '/cygdrive/%s/%s' % (drive.lower(), p)

## See: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx

## and: https://www.cygwin.com/cygwin-ug-net/using.html#unc-paths

(re.compile(r"\\\\\?\\UNC\\([^\\]+)\\([^\\]+)(?:\\(.*))?"),

(lambda server, share, rest_path: '//%s/%s/%s' % (server, share, rest_path.replace('\\', '/'))),

False

),

(re.compile(r"\\\\\?\\(\w):[/\\](.*)"),

_cygexpath,

False

),

(re.compile(r"(\w):[/\\](.*)"),

_cygexpath,

False

),

(re.compile(r"file:(.*)", re.I),

(lambda rest_path: rest_path),

True),

(re.compile(r"(\w{2,}:.*)"), # remote URL, do nothing

(lambda url: url),

False),

"""Use :meth:`git.cmd.Git.polish_url()` instead, that works on any environment."""

if not path.startswith(('/cygdrive', '//')):

for regex, parser, recurse in _cygpath_parsers:

match = regex.match(path)

if match:

path = parser(*match.groups())

if recurse:

path = cygpath(path)

break

else:

path = _cygexpath(None, path)

m = _decygpath_regex.match(path)

if m:

drive, rest_path = m.groups()

path = '%s:%s' % (drive.upper(), rest_path or '')

if not is_win:

return False

#from subprocess import check_output

is_cygwin = _is_cygwin_cache.get(git_executable)

if is_cygwin is None:

is_cygwin = False

try:

git_dir = osp.dirname(git_executable)

if not git_dir:

res = py_where(git_executable)

git_dir = osp.dirname(res[0]) if res else None

## Just a name given, not a real path.

uname_cmd = osp.join(git_dir, 'uname')

process = subprocess.Popen([uname_cmd], stdout=subprocess.PIPE,

universal_newlines=True)

uname_out, _ = process.communicate()

#retcode = process.poll()

is_cygwin = 'CYGWIN' in uname_out

except Exception as ex:

log.debug('Failed checking if running in CYGWIN due to: %r', ex)

_is_cygwin_cache[git_executable] = is_cygwin

try:

p = osp.expanduser(p)

if expand_vars:

p = osp.expandvars(p)

return osp.normpath(osp.abspath(p))

except Exception:

"""

_num_op_codes = 9

BEGIN, END, COUNTING, COMPRESSING, WRITING, RECEIVING, RESOLVING, FINDING_SOURCES, CHECKING_OUT = \

[1 << x for x in range(_num_op_codes)]

STAGE_MASK = BEGIN | END

OP_MASK = ~STAGE_MASK

DONE_TOKEN = 'done.'

TOKEN_SEPARATOR = ', '

__slots__ = ('_cur_line',

'_seen_ops',

'error_lines', # Lines that started with 'error:' or 'fatal:'.

'other_lines') # Lines not denoting progress (i.e.g. push-infos).

re_op_absolute = re.compile(r"(remote: )?([\w\s]+):\s+()(\d+)()(.*)")

re_op_relative = re.compile(r"(remote: )?([\w\s]+):\s+(\d+)% \((\d+)/(\d+)\)(.*)")

def __init__(self):

self._seen_ops = []

self._cur_line = None

self.error_lines = []

self.other_lines = []

def _parse_progress_line(self, line):

"""Parse progress information from the given line as retrieved by git-push

# handle

# Counting objects: 4, done.

# Compressing objects: 50% (1/2)

# Compressing objects: 100% (2/2)

# Compressing objects: 100% (2/2), done.

self._cur_line = line = line.decode('utf-8') if isinstance(line, bytes) else line

if self.error_lines or self._cur_line.startswith(('error:', 'fatal:')):

self.error_lines.append(self._cur_line)

return

# find escape characters and cut them away - regex will not work with

# them as they are non-ascii. As git might expect a tty, it will send them

last_valid_index = None

for i, c in enumerate(reversed(line)):

if ord(c) < 32:

# its a slice index

last_valid_index = -i - 1

# END character was non-ascii

# END for each character in line

if last_valid_index is not None:

line = line[:last_valid_index]

# END cut away invalid part

line = line.rstrip()

cur_count, max_count = None, None

match = self.re_op_relative.match(line)

if match is None:

match = self.re_op_absolute.match(line)

if not match:

self.line_dropped(line)

self.other_lines.append(line)

return

# END could not get match

op_code = 0

_remote, op_name, _percent, cur_count, max_count, message = match.groups()

# get operation id

if op_name == "Counting objects":

op_code |= self.COUNTING

elif op_name == "Compressing objects":

op_code |= self.COMPRESSING

elif op_name == "Writing objects":

op_code |= self.WRITING

elif op_name == 'Receiving objects':

op_code |= self.RECEIVING

elif op_name == 'Resolving deltas':

op_code |= self.RESOLVING

elif op_name == 'Finding sources':

op_code |= self.FINDING_SOURCES

elif op_name == 'Checking out files':

op_code |= self.CHECKING_OUT

else:

# Note: On windows it can happen that partial lines are sent

# Hence we get something like "CompreReceiving objects", which is

# a blend of "Compressing objects" and "Receiving objects".

# This can't really be prevented, so we drop the line verbosely

# to make sure we get informed in case the process spits out new

# commands at some point.

self.line_dropped(line)

# Note: Don't add this line to the other lines, as we have to silently

# drop it

return

# END handle op code

# figure out stage

if op_code not in self._seen_ops:

self._seen_ops.append(op_code)

op_code |= self.BEGIN

# END begin opcode

if message is None:

message = ''

# END message handling

message = message.strip()

if message.endswith(self.DONE_TOKEN):

op_code |= self.END

message = message[:-len(self.DONE_TOKEN)]

# END end message handling

message = message.strip(self.TOKEN_SEPARATOR)

self.update(op_code,

cur_count and float(cur_count),

max_count and float(max_count),

message)

def new_message_handler(self):

"""

def handler(line):

return self._parse_progress_line(line.rstrip())

# end

return handler

def line_dropped(self, line):

"""Called whenever a line could not be understood and was therefore dropped."""

pass

def update(self, op_code, cur_count, max_count=None, message=''):

"""Called whenever the progress changes

"""An implementation forwarding updates to any callable"""

__slots__ = ('_callable')

def __init__(self, fn):

self._callable = fn

super(CallableRemoteProgress, self).__init__()

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

"""Actors hold information about a person acting on the repository. They

# PRECOMPILED REGEX

name_only_regex = re.compile(r'<(.*)>')

name_email_regex = re.compile(r'(.*) <(.*?)>')

# ENVIRONMENT VARIABLES

# read when creating new commits

env_author_name = "GIT_AUTHOR_NAME"

env_author_email = "GIT_AUTHOR_EMAIL"

env_committer_name = "GIT_COMMITTER_NAME"

env_committer_email = "GIT_COMMITTER_EMAIL"

# CONFIGURATION KEYS

conf_name = 'name'

conf_email = 'email'

__slots__ = ('name', 'email')

def __init__(self, name, email):

self.name = name

self.email = email

def __eq__(self, other):

return self.name == other.name and self.email == other.email

def __ne__(self, other):

return not (self == other)

def __hash__(self):

return hash((self.name, self.email))

def __str__(self):

return self.name

def __repr__(self):

return '<git.Actor "%s <%s>">' % (self.name, self.email)

@classmethod

def _from_string(cls, string):

"""Create an Actor from a string.

m = cls.name_email_regex.search(string)

if m:

name, email = m.groups()

return Actor(name, email)

else:

m = cls.name_only_regex.search(string)

if m:

return Actor(m.group(1), None)

# assume best and use the whole string as name

return Actor(string, None)

# END special case name

# END handle name/email matching

@classmethod

def _main_actor(cls, env_name, env_email, config_reader=None):

actor = Actor('', '')

user_id = None # We use this to avoid multiple calls to getpass.getuser()

def default_email():

nonlocal user_id

if not user_id:

user_id = get_user_id()

return user_id

def default_name():

return default_email().split('@')[0]

for attr, evar, cvar, default in (('name', env_name, cls.conf_name, default_name),

('email', env_email, cls.conf_email, default_email)):

try:

val = os.environ[evar]

setattr(actor, attr, val)

except KeyError:

if config_reader is not None:

setattr(actor, attr, config_reader.get_value('user', cvar, default()))

# END config-reader handling

if not getattr(actor, attr):

setattr(actor, attr, default())

# END handle name

# END for each item to retrieve

return actor

@classmethod

def committer(cls, config_reader=None):

"""

return cls._main_actor(cls.env_committer_name, cls.env_committer_email, config_reader)

@classmethod

def author(cls, config_reader=None):

"""Same as committer(), but defines the main author. It may be specified in the environment,

"""

__slots__ = ("total", "files")

def __init__(self, total, files):

self.total = total

self.files = files

@classmethod

def _list_from_string(cls, repo, text):

"""Create a Stat object from output retrieved by git-diff.

hsh = {'total': {'insertions': 0, 'deletions': 0, 'lines': 0, 'files': 0}, 'files': {}}

for line in text.splitlines():

(raw_insertions, raw_deletions, filename) = line.split("\t")

insertions = raw_insertions != '-' and int(raw_insertions) or 0

deletions = raw_deletions != '-' and int(raw_deletions) or 0

hsh['total']['insertions'] += insertions

hsh['total']['deletions'] += deletions

hsh['total']['lines'] += insertions + deletions

hsh['total']['files'] += 1

hsh['files'][filename.strip()] = {'insertions': insertions,

'deletions': deletions,

'lines': insertions + deletions}

"""Wrapper around a file-like object that remembers the SHA1 of

__slots__ = ("f", "sha1")

def __init__(self, f):

self.f = f

self.sha1 = make_sha(b"")

def write(self, data):

self.sha1.update(data)

return self.f.write(data)

def write_sha(self):

sha = self.sha1.digest()

self.f.write(sha)

return sha

def close(self):

sha = self.write_sha()

self.f.close()

return sha

def tell(self):

"""Provides methods to obtain, check for, and release a file based lock which

__slots__ = ("_file_path", "_owns_lock")

def __init__(self, file_path):

self._file_path = file_path

self._owns_lock = False

def __del__(self):

self._release_lock()

def _lock_file_path(self):

""":return: Path to lockfile"""

return "%s.lock" % (self._file_path)

def _has_lock(self):

""":return: True if we have a lock and if the lockfile still exists

return self._owns_lock

def _obtain_lock_or_raise(self):

"""Create a lock file as flag for other instances, mark our instance as lock-holder

if self._has_lock():

return

lock_file = self._lock_file_path()

if osp.isfile(lock_file):

raise IOError("Lock for file %r did already exist, delete %r in case the lock is illegal" %

(self._file_path, lock_file))

try:

flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL

if is_win:

flags |= os.O_SHORT_LIVED

fd = os.open(lock_file, flags, 0)

os.close(fd)

except OSError as e:

raise IOError(str(e)) from e

self._owns_lock = True

def _obtain_lock(self):

"""The default implementation will raise if a lock cannot be obtained.

return self._obtain_lock_or_raise()

def _release_lock(self):

"""Release our lock if we have one"""

if not self._has_lock():

return

# if someone removed our file beforhand, lets just flag this issue

# instead of failing, to make it more usable.

lfp = self._lock_file_path()

try:

rmfile(lfp)

except OSError:

pass

"""The lock file will block until a lock could be obtained, or fail after

__slots__ = ("_check_interval", "_max_block_time")

def __init__(self, file_path, check_interval_s=0.3, max_block_time_s=maxsize):

"""Configure the instance

super(BlockingLockFile, self).__init__(file_path)

self._check_interval = check_interval_s

self._max_block_time = max_block_time_s

def _obtain_lock(self):

"""This method blocks until it obtained the lock, or raises IOError if

starttime = time.time()

maxtime = starttime + float(self._max_block_time)

while True:

try:

super(BlockingLockFile, self)._obtain_lock()

except IOError as e:

# synity check: if the directory leading to the lockfile is not

# readable anymore, raise an exception

curtime = time.time()

if not osp.isdir(osp.dirname(self._lock_file_path())):

msg = "Directory containing the lockfile %r was not readable anymore after waiting %g seconds" % (

self._lock_file_path(), curtime - starttime)

raise IOError(msg) from e

# END handle missing directory

if curtime >= maxtime:

msg = "Waited %g seconds for lock at %r" % (maxtime - starttime, self._lock_file_path())

raise IOError(msg) from e

# END abort if we wait too long

time.sleep(self._check_interval)

else:

break

"""

__slots__ = ('_id_attr', '_prefix')

def __new__(cls, id_attr, prefix=''):

return super(IterableList, cls).__new__(cls)

def __init__(self, id_attr, prefix=''):

self._id_attr = id_attr

self._prefix = prefix

def __contains__(self, attr):

# first try identity match for performance

try:

rval = list.__contains__(self, attr)

if rval:

return rval

except (AttributeError, TypeError):

pass

# END handle match

# otherwise make a full name search

try:

getattr(self, attr)

return True

except (AttributeError, TypeError):

return False

# END handle membership

def __getattr__(self, attr):

attr = self._prefix + attr

for item in self:

if getattr(item, self._id_attr) == attr:

return item

# END for each item

return list.__getattribute__(self, attr)

def __getitem__(self, index):

if isinstance(index, int):

return list.__getitem__(self, index)

try:

return getattr(self, index)

except AttributeError as e:

raise IndexError("No item found with id %r" % (self._prefix + index)) from e

# END handle getattr

def __delitem__(self, index):

delindex = index

if not isinstance(index, int):

delindex = -1

name = self._prefix + index

for i, item in enumerate(self):

if getattr(item, self._id_attr) == name:

delindex = i

break

# END search index

# END for each item

if delindex == -1:

raise IndexError("Item with name %s not found" % name)

# END handle error

# END get index to delete

"""Defines an interface for iterable items which is to assure a uniform

__slots__ = ()

_id_attribute_ = "attribute that most suitably identifies your instance"

@classmethod

def list_items(cls, repo, *args, **kwargs):

"""

out_list = IterableList(cls._id_attribute_)

out_list.extend(cls.iter_items(repo, *args, **kwargs))

return out_list

@classmethod

def iter_items(cls, repo, *args, **kwargs):

"""For more information about the arguments, see list_items