"""Registers for notifications to lean that process output is ready to read, and dispatches lines to

fdmap = {process.stdout.fileno(): (stdout_handler, [b'']),

process.stderr.fileno(): (stderr_handler, [b''])}

def _parse_lines_from_buffer(buf):

line = b''

bi = 0

lb = len(buf)

while bi < lb:

char = _bchr(buf[bi])

bi += 1

if char in (b'\r', b'\n') and line:

yield bi, line

line = b''

else:

line += char

# END process parsed line

# END while file is not done reading

# end

def _read_lines_from_fno(fno, last_buf_list):

buf = os.read(fno, mmap.PAGESIZE)

buf = last_buf_list[0] + buf

bi = 0

for bi, line in _parse_lines_from_buffer(buf):

yield line

# for each line to parse from the buffer

# keep remainder

last_buf_list[0] = buf[bi:]

def _dispatch_single_line(line, handler):

line = line.decode(defenc)

if line and handler:

handler(line)

# end dispatch helper

# end single line helper

def _dispatch_lines(fno, handler, buf_list):

lc = 0

for line in _read_lines_from_fno(fno, buf_list):

_dispatch_single_line(line, handler)

lc += 1

# for each line

return lc

# end

def _deplete_buffer(fno, handler, buf_list, wg=None):

lc = 0

while True:

line_count = _dispatch_lines(fno, handler, buf_list)

lc += line_count

if line_count == 0:

break

# end deplete buffer

if buf_list[0]:

_dispatch_single_line(buf_list[0], handler)

lc += 1

# end

if wg:

wg.done()

return lc

# end

if hasattr(select, 'poll'):

# poll is preferred, as select is limited to file handles up to 1024 ... . This could otherwise be

# an issue for us, as it matters how many handles our own process has

poll = select.poll()

READ_ONLY = select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR

CLOSED = select.POLLHUP | select.POLLERR

poll.register(process.stdout, READ_ONLY)

poll.register(process.stderr, READ_ONLY)

closed_streams = set()

while True:

# no timeout

try:

poll_result = poll.poll()

except select.error as e:

if e.args[0] == errno.EINTR:

continue

raise

# end handle poll exception

for fd, result in poll_result:

if result & CLOSED:

closed_streams.add(fd)

else:

_dispatch_lines(fd, *fdmap[fd])

# end handle closed stream

# end for each poll-result tuple

if len(closed_streams) == len(fdmap):

break

# end its all done

# end endless loop

# Depelete all remaining buffers

for fno, (handler, buf_list) in fdmap.items():

_deplete_buffer(fno, handler, buf_list)

# end for each file handle

for fno in fdmap.keys():

poll.unregister(fno)

# end don't forget to unregister !

else:

# Oh ... probably we are on windows. select.select() can only handle sockets, we have files

# The only reliable way to do this now is to use threads and wait for both to finish

# Since the finalizer is expected to wait, we don't have to introduce our own wait primitive

# NO: It's not enough unfortunately, and we will have to sync the threads

wg = WaitGroup()

for fno, (handler, buf_list) in fdmap.items():

wg.add(1)

t = threading.Thread(target=lambda: _deplete_buffer(fno, handler, buf_list, wg))

t.start()

# end

# NOTE: Just joining threads can possibly fail as there is a gap between .start() and when it's

# actually started, which could make the wait() call to just return because the thread is not yet

# active

wg.wait()

# end

for k, v in d.items():

setattr(self, k, v)

for k in excluded:

"""

__slots__ = ("_working_dir", "cat_file_all", "cat_file_header", "_version_info",

"_git_options", "_environment")

_excluded_ = ('cat_file_all', 'cat_file_header', '_version_info')

def __getstate__(self):

return slots_to_dict(self, exclude=self._excluded_)

def __setstate__(self, d):

dict_to_slots_and__excluded_are_none(self, d, excluded=self._excluded_)

# CONFIGURATION

# The size in bytes read from stdout when copying git's output to another stream

max_chunk_size = 1024 * 64

git_exec_name = "git" # default that should work on linux and windows

git_exec_name_win = "git.cmd" # alternate command name, windows only

# Enables debugging of GitPython's git commands

GIT_PYTHON_TRACE = os.environ.get("GIT_PYTHON_TRACE", False)

# value of Windows process creation flag taken from MSDN

CREATE_NO_WINDOW = 0x08000000

# Provide the full path to the git executable. Otherwise it assumes git is in the path

_git_exec_env_var = "GIT_PYTHON_GIT_EXECUTABLE"

GIT_PYTHON_GIT_EXECUTABLE = os.environ.get(_git_exec_env_var, git_exec_name)

# If True, a shell will be used when executing git commands.

# This should only be desirable on windows, see https://github.com/gitpython-developers/GitPython/pull/126

# for more information

# Override this value using `Git.USE_SHELL = True`

USE_SHELL = False

class AutoInterrupt(object):

"""Kill/Interrupt the stored process instance once this instance goes out of scope. It is

__slots__ = ("proc", "args")

def __init__(self, proc, args):

self.proc = proc

self.args = args

def __del__(self):

if self.proc is None:

return

proc = self.proc

self.proc = None

if proc.stdin:

proc.stdin.close()

if proc.stdout:

proc.stdout.close()

if proc.stderr:

proc.stderr.close()

# did the process finish already so we have a return code ?

if proc.poll() is not None:

return

# can be that nothing really exists anymore ...

if os is None or os.kill is None:

return

# try to kill it

try:

os.kill(proc.pid, 2) # interrupt signal

proc.wait() # ensure process goes away

except (OSError, WindowsError):

pass # ignore error when process already died

except AttributeError:

# try windows

# for some reason, providing None for stdout/stderr still prints something. This is why

# we simply use the shell and redirect to nul. Its slower than CreateProcess, question

# is whether we really want to see all these messages. Its annoying no matter what.

call(("TASKKILL /F /T /PID %s 2>nul 1>nul" % str(proc.pid)), shell=True)

# END exception handling

def __getattr__(self, attr):

return getattr(self.proc, attr)

def wait(self, stderr=b''):

"""Wait for the process and return its status code.

if stderr is None:

stderr = b''

stderr = force_bytes(stderr)

status = self.proc.wait()

def read_all_from_possibly_closed_stream(stream):

try:

return stderr + force_bytes(stream.read())

except ValueError:

return stderr or b''

if status != 0:

errstr = read_all_from_possibly_closed_stream(self.proc.stderr)

log.debug('AutoInterrupt wait stderr: %r' % (errstr,))

raise GitCommandError(self.args, status, errstr)

# END status handling

return status

# END auto interrupt

class CatFileContentStream(object):

"""Object representing a sized read-only stream returning the contents of

__slots__ = ('_stream', '_nbr', '_size')

def __init__(self, size, stream):

self._stream = stream

self._size = size

self._nbr = 0 # num bytes read

# special case: if the object is empty, has null bytes, get the

# final newline right away.

if size == 0:

stream.read(1)

# END handle empty streams

def read(self, size=-1):

bytes_left = self._size - self._nbr

if bytes_left == 0:

return b''

if size > -1:

# assure we don't try to read past our limit

size = min(bytes_left, size)

else:

# they try to read all, make sure its not more than what remains

size = bytes_left

# END check early depletion

data = self._stream.read(size)

self._nbr += len(data)

# check for depletion, read our final byte to make the stream usable by others

if self._size - self._nbr == 0:

self._stream.read(1) # final newline

# END finish reading

return data

def readline(self, size=-1):

if self._nbr == self._size:

return b''

# clamp size to lowest allowed value

bytes_left = self._size - self._nbr

if size > -1:

size = min(bytes_left, size)

else:

size = bytes_left

# END handle size

data = self._stream.readline(size)

self._nbr += len(data)

# handle final byte

if self._size - self._nbr == 0:

self._stream.read(1)

# END finish reading

return data

def readlines(self, size=-1):

if self._nbr == self._size:

return list()

# leave all additional logic to our readline method, we just check the size

out = list()

nbr = 0

while True:

line = self.readline()

if not line:

break

out.append(line)

if size > -1:

nbr += len(line)

if nbr > size:

break

# END handle size constraint

# END readline loop

return out

def __iter__(self):

return self

def next(self):

line = self.readline()

if not line:

raise StopIteration

return line

def __del__(self):

bytes_left = self._size - self._nbr

if bytes_left:

# read and discard - seeking is impossible within a stream

# includes terminating newline

self._stream.read(bytes_left + 1)

# END handle incomplete read

def __init__(self, working_dir=None):

"""Initialize this instance with:

super(Git, self).__init__()

self._working_dir = working_dir

self._git_options = ()

# Extra environment variables to pass to git commands

self._environment = {}

# cached command slots

self.cat_file_header = None

self.cat_file_all = None

def __getattr__(self, name):

"""A convenience method as it allows to call the command as if it was

if name[0] == '_':

return LazyMixin.__getattr__(self, name)

return lambda *args, **kwargs: self._call_process(name, *args, **kwargs)

def _set_cache_(self, attr):

if attr == '_version_info':

# We only use the first 4 numbers, as everthing else could be strings in fact (on windows)

version_numbers = self._call_process('version').split(' ')[2]

self._version_info = tuple(int(n) for n in version_numbers.split('.')[:4] if n.isdigit())

else:

super(Git, self)._set_cache_(attr)

# END handle version info

@property

def working_dir(self):

""":return: Git directory we are working on"""

return self._working_dir

@property

def version_info(self):

"""

return self._version_info

def execute(self, command,

istream=None,

with_keep_cwd=False,

with_extended_output=False,

with_exceptions=True,

as_process=False,

output_stream=None,

stdout_as_string=True,

kill_after_timeout=None,

with_stdout=True,

universal_newlines=False,

**subprocess_kwargs

):

"""Handles executing the command on the shell and consumes and returns

if self.GIT_PYTHON_TRACE and (self.GIT_PYTHON_TRACE != 'full' or as_process):

log.info(' '.join(command))

# Allow the user to have the command executed in their working dir.

if with_keep_cwd or self._working_dir is None:

cwd = os.getcwd()

else:

cwd = self._working_dir

# Start the process

env = os.environ.copy()

# Attempt to force all output to plain ascii english, which is what some parsing code

# may expect.

# According to stackoverflow (http://goo.gl/l74GC8), we are setting LANGUAGE as well

# just to be sure.

env["LANGUAGE"] = "C"

env["LC_ALL"] = "C"

env.update(self._environment)

if sys.platform == 'win32':

cmd_not_found_exception = WindowsError

if kill_after_timeout:

raise GitCommandError('"kill_after_timeout" feature is not supported on Windows.')

else:

if sys.version_info[0] > 2:

cmd_not_found_exception = FileNotFoundError # NOQA # this is defined, but flake8 doesn't know

else:

cmd_not_found_exception = OSError

# end handle

creationflags = self.CREATE_NO_WINDOW if sys.platform == 'win32' else 0

try:

proc = Popen(command,

env=env,

cwd=cwd,

bufsize=-1,

stdin=istream,

stderr=PIPE,

stdout=PIPE if with_stdout else open(os.devnull, 'wb'),

shell=self.USE_SHELL,

close_fds=(os.name == 'posix'), # unsupported on windows

universal_newlines=universal_newlines,

creationflags=creationflags,

**subprocess_kwargs

)

except cmd_not_found_exception as err:

raise GitCommandNotFound(str(err))

if as_process:

return self.AutoInterrupt(proc, command)

def _kill_process(pid):

""" Callback method to kill a process. """

p = Popen(['ps', '--ppid', str(pid)], stdout=PIPE, creationflags=creationflags)

child_pids = []

for line in p.stdout:

if len(line.split()) > 0:

local_pid = (line.split())[0]

if local_pid.isdigit():

child_pids.append(int(local_pid))

try:

# Windows does not have SIGKILL, so use SIGTERM instead

sig = getattr(signal, 'SIGKILL', signal.SIGTERM)

os.kill(pid, sig)

for child_pid in child_pids:

try:

os.kill(child_pid, sig)

except OSError:

pass

kill_check.set() # tell the main routine that the process was killed

except OSError:

# It is possible that the process gets completed in the duration after timeout

# happens and before we try to kill the process.

pass

return

# end

if kill_after_timeout:

kill_check = threading.Event()

watchdog = threading.Timer(kill_after_timeout, _kill_process, args=(proc.pid, ))

# Wait for the process to return

status = 0

stdout_value = b''

stderr_value = b''

try:

if output_stream is None:

if kill_after_timeout:

watchdog.start()

stdout_value, stderr_value = proc.communicate()

if kill_after_timeout:

watchdog.cancel()

if kill_check.isSet():

stderr_value = 'Timeout: the command "%s" did not complete in %d ' \

'secs.' % (" ".join(command), kill_after_timeout)

# strip trailing "\n"

if stdout_value.endswith(b"\n"):

stdout_value = stdout_value[:-1]

if stderr_value.endswith(b"\n"):

stderr_value = stderr_value[:-1]

status = proc.returncode

else:

stream_copy(proc.stdout, output_stream, self.max_chunk_size)

stdout_value = output_stream

stderr_value = proc.stderr.read()

# strip trailing "\n"

if stderr_value.endswith(b"\n"):

stderr_value = stderr_value[:-1]

status = proc.wait()

# END stdout handling

finally:

proc.stdout.close()

proc.stderr.close()

if self.GIT_PYTHON_TRACE == 'full':

cmdstr = " ".join(command)

def as_text(stdout_value):

return not output_stream and safe_decode(stdout_value) or '<OUTPUT_STREAM>'

# end

if stderr_value:

log.info("%s -> %d; stdout: '%s'; stderr: '%s'",

cmdstr, status, as_text(stdout_value), safe_decode(stderr_value))

elif stdout_value:

log.info("%s -> %d; stdout: '%s'", cmdstr, status, as_text(stdout_value))

else:

log.info("%s -> %d", cmdstr, status)

# END handle debug printing

if with_exceptions and status != 0:

if with_extended_output:

raise GitCommandError(command, status, stderr_value, stdout_value)

else:

raise GitCommandError(command, status, stderr_value)

if isinstance(stdout_value, bytes) and stdout_as_string: # could also be output_stream

stdout_value = safe_decode(stdout_value)

# Allow access to the command's status code

if with_extended_output:

return (status, stdout_value, safe_decode(stderr_value))

else:

return stdout_value

def environment(self):

return self._environment

def update_environment(self, **kwargs):

"""

old_env = {}

for key, value in kwargs.items():

# set value if it is None

if value is not None:

if key in self._environment:

old_env[key] = self._environment[key]

else:

old_env[key] = None

self._environment[key] = value

# remove key from environment if its value is None

elif key in self._environment:

old_env[key] = self._environment[key]

del self._environment[key]

return old_env

@contextmanager

def custom_environment(self, **kwargs):

"""

old_env = self.update_environment(**kwargs)

try:

yield

finally:

self.update_environment(**old_env)

def transform_kwarg(self, name, value, split_single_char_options):

if len(name) == 1:

if value is True:

return ["-%s" % name]

elif type(value) is not bool:

if split_single_char_options:

return ["-%s" % name, "%s" % value]

else:

return ["-%s%s" % (name, value)]

else:

if value is True:

return ["--%s" % dashify(name)]

elif type(value) is not bool:

return ["--%s=%s" % (dashify(name), value)]

return []

def transform_kwargs(self, split_single_char_options=True, **kwargs):

"""Transforms Python style kwargs into git command line options."""

args = list()

kwargs = OrderedDict(sorted(kwargs.items(), key=lambda x: x[0]))

for k, v in kwargs.items():

if isinstance(v, (list, tuple)):

for value in v:

args += self.transform_kwarg(k, value, split_single_char_options)

else:

args += self.transform_kwarg(k, v, split_single_char_options)

return args

@classmethod

def __unpack_args(cls, arg_list):

if not isinstance(arg_list, (list, tuple)):

# This is just required for unicode conversion, as subprocess can't handle it

# However, in any other case, passing strings (usually utf-8 encoded) is totally fine

if not PY3 and isinstance(arg_list, unicode):

return [arg_list.encode(defenc)]

return [str(arg_list)]

outlist = list()

for arg in arg_list:

if isinstance(arg_list, (list, tuple)):

outlist.extend(cls.__unpack_args(arg))

elif not PY3 and isinstance(arg_list, unicode):

outlist.append(arg_list.encode(defenc))

# END recursion

else:

outlist.append(str(arg))

# END for each arg

return outlist

def __call__(self, **kwargs):

"""Specify command line options to the git executable

self._git_options = self.transform_kwargs(

split_single_char_options=True, **kwargs)

return self

def _call_process(self, method, *args, **kwargs):

"""Run the given git command with the specified arguments and return

# Handle optional arguments prior to calling transform_kwargs

# otherwise these'll end up in args, which is bad.

_kwargs = dict()

for kwarg in execute_kwargs:

try:

_kwargs[kwarg] = kwargs.pop(kwarg)

except KeyError:

pass

insert_after_this_arg = kwargs.pop('insert_kwargs_after', None)

# Prepare the argument list

opt_args = self.transform_kwargs(**kwargs)

ext_args = self.__unpack_args([a for a in args if a is not None])

if insert_after_this_arg is None:

args = opt_args + ext_args

else:

try:

index = ext_args.index(insert_after_this_arg)

except ValueError:

raise ValueError("Couldn't find argument '%s' in args %s to insert kwargs after"

% (insert_after_this_arg, str(ext_args)))

# end handle error

args = ext_args[:index + 1] + opt_args + ext_args[index + 1:]

# end handle kwargs

def make_call():

call = [self.GIT_PYTHON_GIT_EXECUTABLE]

# add the git options, the reset to empty

# to avoid side_effects

call.extend(self._git_options)

self._git_options = ()

call.extend([dashify(method)])

call.extend(args)

return call

# END utility to recreate call after changes

if sys.platform == 'win32':

try:

try:

return self.execute(make_call(), **_kwargs)

except WindowsError:

# did we switch to git.cmd already, or was it changed from default ? permanently fail

if self.GIT_PYTHON_GIT_EXECUTABLE != self.git_exec_name:

raise

# END handle overridden variable

type(self).GIT_PYTHON_GIT_EXECUTABLE = self.git_exec_name_win

try:

return self.execute(make_call(), **_kwargs)

finally:

import warnings

msg = "WARNING: Automatically switched to use git.cmd as git executable"

msg += ", which reduces performance by ~70%."

msg += "It is recommended to put git.exe into the PATH or to "

msg += "set the %s " % self._git_exec_env_var

msg += "environment variable to the executable's location"

warnings.warn(msg)

# END print of warning

# END catch first failure

except WindowsError:

raise WindowsError("The system cannot find or execute the file at %r" % self.GIT_PYTHON_GIT_EXECUTABLE)

# END provide better error message

else:

return self.execute(make_call(), **_kwargs)

# END handle windows default installation

def _parse_object_header(self, header_line):

"""

tokens = header_line.split()

if len(tokens) != 3:

if not tokens:

raise ValueError("SHA could not be resolved, git returned: %r" % (header_line.strip()))

else:

raise ValueError("SHA %s could not be resolved, git returned: %r" % (tokens[0], header_line.strip()))

# END handle actual return value

# END error handling

if len(tokens[0]) != 40:

raise ValueError("Failed to parse header: %r" % header_line)

return (tokens[0], tokens[1], int(tokens[2]))

def _prepare_ref(self, ref):

# required for command to separate refs on stdin, as bytes

refstr = ref

if isinstance(ref, bytes):

# Assume 40 bytes hexsha - bin-to-ascii for some reason returns bytes, not text

refstr = ref.decode('ascii')

elif not isinstance(ref, string_types):

refstr = str(ref) # could be ref-object

if not refstr.endswith("\n"):

refstr += "\n"

return refstr.encode(defenc)

def _get_persistent_cmd(self, attr_name, cmd_name, *args, **kwargs):

cur_val = getattr(self, attr_name)

if cur_val is not None:

return cur_val

options = {"istream": PIPE, "as_process": True}

options.update(kwargs)

cmd = self._call_process(cmd_name, *args, **options)

setattr(self, attr_name, cmd)