11195840649

Committed 05 Oct 2024 07:42PM UTC coverage: 93.035% (+0.6%) from 92.424%

Build # 11195840649

Build Type

push

github

Committed by

msiemens

Commit Message

chore: drop Python 3.7 support

Run Details

374 of 402 relevant lines covered (93.03%)

5.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.21

/PyGitUp/git_wrapper.py

"""
A wrapper extending GitPython's repo.git.

This wrapper class provides support for stdout messages in Git Exceptions
and (nearly) realtime stdout output. In addition, some methods of the
original repo.git are shadowed by custom methods providing functionality
needed for `git up`.
"""


__all__ = ['GitWrapper', 'GitError']

###############################################################################
# IMPORTS
###############################################################################

# Python libs
import sys
import re
import subprocess
import platform
from contextlib import contextmanager
from io import BufferedReader
from threading import Thread
from typing import IO, Optional, List

# 3rd party libs
from termcolor import colored  # Assume, colorama is already initialized
from git import GitCommandError, CheckoutError as OrigCheckoutError, Git
from git.cmd import Git as GitCmd

# PyGitUp libs
from PyGitUp.utils import find


###############################################################################
# GitWrapper
###############################################################################

class GitWrapper:
    """
    A wrapper for repo.git providing better stdout handling + better exceptions.

    It is preferred to repo.git because it doesn't print to stdout
    in real time. In addition, this wrapper provides better error
    handling (it provides stdout messages inside the exception, too).
    """

    def __init__(self, repo):
        if repo:
            #: :type: git.Repo
            self.repo = repo
            #: :type: git.Git
            self.git = self.repo.git
        else:
            #: :type: git.Git
            self.git = Git()

    def __del__(self):
        # Is the following true?

        # GitPython runs persistent git processes in  the working directory.
        # Therefore, when we use 'git up' in something like a test environment,
        # this might cause troubles because of the open file handlers (like
        # trying to remove the directory right after the test has finished).
        # 'clear_cache' kills the processes...

        if platform.system() == 'Windows':  # pragma: no cover
            pass
            # ... or rather "should kill", because but somehow it recently
            # started to not kill cat_file_header out of the blue (I even
            # tried running old code, but the once working code failed).
            # Thus, we kill it  manually here.
            if self.git.cat_file_header is not None:
                subprocess.call(("TASKKILL /F /T /PID {} 2>nul 1>nul".format(
                    str(self.git.cat_file_header.proc.pid)
                )), shell=True)
            if self.git.cat_file_all is not None:
                subprocess.call(("TASKKILL /F /T /PID {} 2>nul 1>nul".format(
                    str(self.git.cat_file_all.proc.pid)
                )), shell=True)

        self.git.clear_cache()

    def _run(self, name, *args, **kwargs):

        """ Run a git command specified by name and args/kwargs. """

        stdout = b''
        cmd = getattr(self.git, name)

        # Ask cmd(...) to return a (status, stdout, stderr) tuple
        kwargs['with_extended_output'] = True

        # Execute command
        try:
            (_, stdout, _) = cmd(*args, **kwargs)
        except GitCommandError as error:
            # Add more meta-information to errors
            message = "'{}' returned exit status {}".format(
                ' '.join(str(c) for c in error.command),
                error.status
            )

            raise GitError(message, stderr=error.stderr, stdout=stdout)

        return stdout.strip()

    def __getattr__(self, name):
        return lambda *args, **kwargs: self._run(name, *args, **kwargs)

    ###########################################################################
    # Overwrite some methods and add new ones
    ###########################################################################

    @contextmanager
    def stasher(self):
        """
        A stashing contextmanager.
        """
        # nonlocal for python2
        stashed = [False]
        clean = [False]

        def stash():
            if clean[0] or not self.repo.is_dirty(submodules=False):
                clean[0] = True
                return
            if stashed[0]:
                return

            if self.change_count > 1:
                message = 'stashing {0} changes'
            else:
                message = 'stashing {0} change'
            print(colored(
                message.format(self.change_count),
                'magenta'
            ))
            try:
                self._run('stash')
            except GitError as git_error:
                raise StashError(stderr=git_error.stderr, stdout=git_error.stdout)

            stashed[0] = True

        yield stash

        if stashed[0]:
            print(colored('unstashing', 'magenta'))
            try:
                self._run('stash', 'pop')
            except GitError as e:
                raise UnstashError(stderr=e.stderr, stdout=e.stdout)

    def checkout(self, branch_name):
        """ Checkout a branch by name. """
        try:
            find(
                self.repo.branches, lambda b: b.name == branch_name
            ).checkout()
        except OrigCheckoutError as e:
            raise CheckoutError(branch_name, details=e)

    def rebase(self, target_branch):
        """ Rebase to target branch. """
        current_branch = self.repo.active_branch

        arguments = (
                ([self.config('git-up.rebase.arguments')] or []) +
                [target_branch.name]
        )
        try:
            self._run('rebase', *arguments)
        except GitError as e:
            raise RebaseError(current_branch.name, target_branch.name,
                              **e.__dict__)

    def fetch(self, *args, **kwargs):
        """ Fetch remote commits. """

        # Execute command
        cmd = self.git.fetch(as_process=True, *args, **kwargs)

        return self.run_cmd(cmd)

    def push(self, *args, **kwargs):
        """ Push commits to remote """
        # Execute command
        cmd = self.git.push(as_process=True, *args, **kwargs)

        return self.run_cmd(cmd)

    @staticmethod
    def stream_reader(input_stream: BufferedReader, output_stream: Optional[IO], result_list: List[str]) -> None:
        """
        Helper method to read from a stream and write to another stream.

        We use a list to store results because they are mutable and allow
        for passing data back to the caller from the thread without additional
        machinery.
        """
        captured_bytes = b""
        while True:
            read_byte = input_stream.read(1)
            captured_bytes += read_byte
            if output_stream is not None:
                output_stream.write(read_byte.decode('utf-8'))
                output_stream.flush()
            if read_byte == b"":
                break
        result_list.append(captured_bytes)

    @staticmethod
    def run_cmd(cmd: GitCmd.AutoInterrupt) -> bytes:
        """ Run a command and return stdout. """
        std_outs = []
        std_errs = []
        stdout_thread = Thread(target=GitWrapper.stream_reader,
                               args=(cmd.stdout, sys.stdout, std_outs))
        stderr_thread = Thread(target=GitWrapper.stream_reader,
                               args=(cmd.stderr, None, std_errs))

        # Wait for the process to quit
        try:
            stdout_thread.start()
            stderr_thread.start()
            cmd.wait()
            stdout_thread.join()
            stderr_thread.join()
        except GitCommandError as error:
            # Add more meta-information to errors
            message = "'{}' returned exit status {}".format(
                ' '.join(str(c) for c in error.command),
                error.status
            )

            raise GitError(message, stderr=error.stderr, stdout=std_outs[0])
        return std_outs[0].strip()

    def config(self, key):
        """ Return `git config key` output or None. """
        try:
            return self.git.config(key)
        except GitCommandError:
            return None

    @property
    def change_count(self):
        """ The number of changes in the working directory. """
        status = self.git.status(porcelain=True, untracked_files='no').strip()
        if not status:
            return 0
        else:
            return len(status.split('\n'))

    @property
    def version(self):
        """
        Return git's version as a list of numbers.

        The original repo.git.version_info has problems with tome types of
        git version strings.
        """
        return re.search(r'\d+(\.\d+)+', self.git.version()).group(0)

    def is_version_min(self, required_version):
        """ Does git's version match the requirements? """
        return self.version.split('.') >= required_version.split('.')


###############################################################################
# GitError + subclasses
###############################################################################

class GitError(Exception):
    """
    Extension of the GitCommandError class.

    New:
    - stdout
    - details: a 'nested' exception with more details
    """

    def __init__(self, message=None, stderr=None, stdout=None, details=None):
        # super(GitError, self).__init__((), None, stderr)
        self.details = details
        self.message = message

        self.stderr = stderr
        self.stdout = stdout

    def __str__(self):  # pragma: no cover
        return self.message


class StashError(GitError):
    """
    Error while stashing
    """

    def __init__(self, **kwargs):
        kwargs.pop('message', None)
        GitError.__init__(self, 'Stashing failed!', **kwargs)


class UnstashError(GitError):
    """
    Error while unstashing
    """

    def __init__(self, **kwargs):
        kwargs.pop('message', None)
        GitError.__init__(self, 'Unstashing failed!', **kwargs)


class CheckoutError(GitError):
    """
    Error during checkout
    """

    def __init__(self, branch_name, **kwargs):
        kwargs.pop('message', None)
        GitError.__init__(self, 'Failed to checkout ' + branch_name,
                          **kwargs)


class RebaseError(GitError):
    """
    Error during rebase command
    """

    def __init__(self, current_branch, target_branch, **kwargs):
        # Remove kwargs we won't pass to GitError
        kwargs.pop('message', None)
        kwargs.pop('command', None)
        kwargs.pop('status', None)

        message = "Failed to rebase {1} onto {0}".format(
            current_branch, target_branch
        )
        GitError.__init__(self, message, **kwargs)

1	"""
2	A wrapper extending GitPython's repo.git.
3
4	This wrapper class provides support for stdout messages in Git Exceptions
5	and (nearly) realtime stdout output. In addition, some methods of the
6	original repo.git are shadowed by custom methods providing functionality
7	needed for `git up`.
8	"""
9
10
11	__all__ = ['GitWrapper', 'GitError']	6✔
12
13	###############################################################################
14	# IMPORTS
15	###############################################################################
16
17	# Python libs
18	import sys	6✔
19	import re	6✔
20	import subprocess	6✔
21	import platform	6✔
22	from contextlib import contextmanager	6✔
23	from io import BufferedReader	6✔
24	from threading import Thread	6✔
25	from typing import IO, Optional, List	6✔
26
27	# 3rd party libs
28	from termcolor import colored # Assume, colorama is already initialized	6✔
29	from git import GitCommandError, CheckoutError as OrigCheckoutError, Git	6✔
30	from git.cmd import Git as GitCmd	6✔
31
32	# PyGitUp libs
33	from PyGitUp.utils import find	6✔
34
35
36	###############################################################################
37	# GitWrapper
38	###############################################################################
39
40	class GitWrapper:	6✔
41	"""
42	A wrapper for repo.git providing better stdout handling + better exceptions.
43
44	It is preferred to repo.git because it doesn't print to stdout
45	in real time. In addition, this wrapper provides better error
46	handling (it provides stdout messages inside the exception, too).
47	"""
48
49	def __init__(self, repo):	6✔
50	if repo:	6✔
51	#: :type: git.Repo
52	self.repo = repo	6✔
53	#: :type: git.Git
54	self.git = self.repo.git	6✔
55	else:
56	#: :type: git.Git
57	self.git = Git()	5✔
58
59	def __del__(self):	6✔
60	# Is the following true?
61
62	# GitPython runs persistent git processes in the working directory.
63	# Therefore, when we use 'git up' in something like a test environment,
64	# this might cause troubles because of the open file handlers (like
65	# trying to remove the directory right after the test has finished).
66	# 'clear_cache' kills the processes...
67
68	if platform.system() == 'Windows': # pragma: no cover
69	pass
70	# ... or rather "should kill", because but somehow it recently
71	# started to not kill cat_file_header out of the blue (I even
72	# tried running old code, but the once working code failed).
73	# Thus, we kill it manually here.
74	if self.git.cat_file_header is not None:
75	subprocess.call(("TASKKILL /F /T /PID {} 2>nul 1>nul".format(
76	str(self.git.cat_file_header.proc.pid)
77	)), shell=True)
78	if self.git.cat_file_all is not None:
79	subprocess.call(("TASKKILL /F /T /PID {} 2>nul 1>nul".format(
80	str(self.git.cat_file_all.proc.pid)
81	)), shell=True)
82
83	self.git.clear_cache()	6✔
84
85	def _run(self, name, args, *kwargs):	6✔
86
87	""" Run a git command specified by name and args/kwargs. """
88
89	stdout = b''	6✔
90	cmd = getattr(self.git, name)	6✔
91
92	# Ask cmd(...) to return a (status, stdout, stderr) tuple
93	kwargs['with_extended_output'] = True	6✔
94
95	# Execute command
96	try:	6✔
97	(_, stdout, _) = cmd(args, *kwargs)	6✔
98	except GitCommandError as error:	6✔
99	# Add more meta-information to errors
100	message = "'{}' returned exit status {}".format(	6✔
101	' '.join(str(c) for c in error.command),
102	error.status
103	)
104
105	raise GitError(message, stderr=error.stderr, stdout=stdout)	6✔
106
107	return stdout.strip()	6✔
108
109	def __getattr__(self, name):	6✔
110	return lambda args, kwargs: self._run(name, args, **kwargs)	6✔
111
112	###########################################################################
113	# Overwrite some methods and add new ones
114	###########################################################################
115
116	@contextmanager	6✔
117	def stasher(self):	6✔
118	"""
119	A stashing contextmanager.
120	"""
121	# nonlocal for python2
122	stashed = [False]	6✔
123	clean = [False]	6✔
124
125	def stash():	6✔
126	if clean[0] or not self.repo.is_dirty(submodules=False):	6✔
127	clean[0] = True	6✔
128	return	6✔
129	if stashed[0]:	6✔
130	return	×
131
132	if self.change_count > 1:	6✔
133	message = 'stashing {0} changes'	×
134	else:
135	message = 'stashing {0} change'	6✔
136	print(colored(	6✔
137	message.format(self.change_count),
138	'magenta'
139	))
140	try:	6✔
141	self._run('stash')	6✔
142	except GitError as git_error:	6✔
143	raise StashError(stderr=git_error.stderr, stdout=git_error.stdout)	6✔
144
145	stashed[0] = True	6✔
146
147	yield stash	6✔
148
149	if stashed[0]:	6✔
150	print(colored('unstashing', 'magenta'))	6✔
151	try:	6✔
152	self._run('stash', 'pop')	6✔
153	except GitError as e:	6✔
154	raise UnstashError(stderr=e.stderr, stdout=e.stdout)	6✔
155
156	def checkout(self, branch_name):	6✔
157	""" Checkout a branch by name. """
158	try:	6✔
159	find(	6✔
160	self.repo.branches, lambda b: b.name == branch_name
161	).checkout()
162	except OrigCheckoutError as e:	×
163	raise CheckoutError(branch_name, details=e)	×
164
165	def rebase(self, target_branch):	6✔
166	""" Rebase to target branch. """
167	current_branch = self.repo.active_branch	6✔
168
169	arguments = (	6✔
170	([self.config('git-up.rebase.arguments')] or []) +
171	[target_branch.name]
172	)
173	try:	6✔
174	self._run('rebase', *arguments)	6✔
175	except GitError as e:	6✔
176	raise RebaseError(current_branch.name, target_branch.name,	6✔
177	**e.__dict__)
178
179	def fetch(self, args, *kwargs):	6✔
180	""" Fetch remote commits. """
181
182	# Execute command
183	cmd = self.git.fetch(as_process=True, args, *kwargs)	6✔
184
185	return self.run_cmd(cmd)	6✔
186
187	def push(self, args, *kwargs):	6✔
188	""" Push commits to remote """
189	# Execute command
190	cmd = self.git.push(as_process=True, args, *kwargs)	6✔
191
192	return self.run_cmd(cmd)	6✔
193
194	@staticmethod	6✔
195	def stream_reader(input_stream: BufferedReader, output_stream: Optional[IO], result_list: List[str]) -> None:	6✔
196	"""
197	Helper method to read from a stream and write to another stream.
198
199	We use a list to store results because they are mutable and allow
200	for passing data back to the caller from the thread without additional
201	machinery.
202	"""
203	captured_bytes = b""	6✔
204	while True:	5✔
205	read_byte = input_stream.read(1)	6✔
206	captured_bytes += read_byte	6✔
207	if output_stream is not None:	6✔
208	output_stream.write(read_byte.decode('utf-8'))	6✔
209	output_stream.flush()	6✔
210	if read_byte == b"":	6✔
211	break	6✔
212	result_list.append(captured_bytes)	6✔
213
214	@staticmethod	6✔
215	def run_cmd(cmd: GitCmd.AutoInterrupt) -> bytes:	6✔
216	""" Run a command and return stdout. """
217	std_outs = []	6✔
218	std_errs = []	6✔
219	stdout_thread = Thread(target=GitWrapper.stream_reader,	6✔
220	args=(cmd.stdout, sys.stdout, std_outs))
221	stderr_thread = Thread(target=GitWrapper.stream_reader,	6✔
222	args=(cmd.stderr, None, std_errs))
223
224	# Wait for the process to quit
225	try:	6✔
226	stdout_thread.start()	6✔
227	stderr_thread.start()	6✔
228	cmd.wait()	6✔
229	stdout_thread.join()	6✔
230	stderr_thread.join()	6✔
231	except GitCommandError as error:	6✔
232	# Add more meta-information to errors
233	message = "'{}' returned exit status {}".format(	6✔
234	' '.join(str(c) for c in error.command),
235	error.status
236	)
237
238	raise GitError(message, stderr=error.stderr, stdout=std_outs[0])	6✔
239	return std_outs[0].strip()	6✔
240
241	def config(self, key):	6✔
242	""" Return `git config key` output or None. """
243	try:	6✔
244	return self.git.config(key)	6✔
245	except GitCommandError:	6✔
246	return None	6✔
247
248	@property	6✔
249	def change_count(self):	6✔
250	""" The number of changes in the working directory. """
251	status = self.git.status(porcelain=True, untracked_files='no').strip()	6✔
252	if not status:	6✔
253	return 0	×
254	else:
255	return len(status.split('\n'))	6✔
256
257	@property	6✔
258	def version(self):	6✔
259	"""
260	Return git's version as a list of numbers.
261
262	The original repo.git.version_info has problems with tome types of
263	git version strings.
264	"""
265	return re.search(r'\d+(\.\d+)+', self.git.version()).group(0)	6✔
266
267	def is_version_min(self, required_version):	6✔
268	""" Does git's version match the requirements? """
269	return self.version.split('.') >= required_version.split('.')	6✔
270
271
272	###############################################################################
273	# GitError + subclasses
274	###############################################################################
275
276	class GitError(Exception):	6✔
277	"""
278	Extension of the GitCommandError class.
279
280	New:
281	- stdout
282	- details: a 'nested' exception with more details
283	"""
284
285	def __init__(self, message=None, stderr=None, stdout=None, details=None):	6✔
286	# super(GitError, self).__init__((), None, stderr)
287	self.details = details	6✔
288	self.message = message	6✔
289
290	self.stderr = stderr	6✔
291	self.stdout = stdout	6✔
292
293	def __str__(self): # pragma: no cover
294	return self.message
295
296
297	class StashError(GitError):	6✔
298	"""
299	Error while stashing
300	"""
301
302	def __init__(self, **kwargs):	6✔
303	kwargs.pop('message', None)	6✔
304	GitError.__init__(self, 'Stashing failed!', **kwargs)	6✔
305
306
307	class UnstashError(GitError):	6✔
308	"""
309	Error while unstashing
310	"""
311
312	def __init__(self, **kwargs):	6✔
313	kwargs.pop('message', None)	6✔
314	GitError.__init__(self, 'Unstashing failed!', **kwargs)	6✔
315
316
317	class CheckoutError(GitError):	6✔
318	"""
319	Error during checkout
320	"""
321
322	def __init__(self, branch_name, **kwargs):	6✔
323	kwargs.pop('message', None)	×
324	GitError.__init__(self, 'Failed to checkout ' + branch_name,	×
325	**kwargs)
326
327
328	class RebaseError(GitError):	6✔
329	"""
330	Error during rebase command
331	"""
332
333	def __init__(self, current_branch, target_branch, **kwargs):	6✔
334	# Remove kwargs we won't pass to GitError
335	kwargs.pop('message', None)	6✔
336	kwargs.pop('command', None)	6✔
337	kwargs.pop('status', None)	6✔
338
339	message = "Failed to rebase {1} onto {0}".format(	6✔
340	current_branch, target_branch
341	)
342	GitError.__init__(self, message, **kwargs)	6✔

msiemens / PyGitUp / 11195840649

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous