11756390499

Committed 07 Oct 2024 05:25PM UTC coverage: 93.284%. Remained the same

Build # 11756390499

Build Type

push

github

Committed by

web-flow

Commit Message

infra: update pytest to v8 (#135)

* Tested with Python 3.12

* Upgrade pytest to the latest version 8.3.3

* Rename setup/teardown functions

Old names were there for compatibility with nose and they're no longer supported in pytest 8.

---------

Co-authored-by: Markus Siemens <markus@m-siemens.de>

Run Details

375 of 402 relevant lines covered (93.28%)

16.45 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] if std_outs else None)

        return std_outs[0].strip() if std_outs else bytes()

    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']	18✔
12
13	###############################################################################
14	# IMPORTS
15	###############################################################################
16
17	# Python libs
18	import sys	18✔
19	import re	18✔
20	import subprocess	18✔
21	import platform	18✔
22	from contextlib import contextmanager	18✔
23	from io import BufferedReader	18✔
24	from threading import Thread	18✔
25	from typing import IO, Optional, List	18✔
26
27	# 3rd party libs
28	from termcolor import colored # Assume, colorama is already initialized	18✔
29	from git import GitCommandError, CheckoutError as OrigCheckoutError, Git	18✔
30	from git.cmd import Git as GitCmd	18✔
31
32	# PyGitUp libs
33	from PyGitUp.utils import find	18✔
34
35
36	###############################################################################
37	# GitWrapper
38	###############################################################################
39
40	class GitWrapper:	18✔
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):	18✔
50	if repo:	18✔
51	#: :type: git.Repo
52	self.repo = repo	18✔
53	#: :type: git.Git
54	self.git = self.repo.git	18✔
55	else:
56	#: :type: git.Git
57	self.git = Git()	12✔
58
59	def __del__(self):	18✔
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()	18✔
84
85	def _run(self, name, args, *kwargs):	18✔
86
87	""" Run a git command specified by name and args/kwargs. """
88
89	stdout = b''	18✔
90	cmd = getattr(self.git, name)	18✔
91
92	# Ask cmd(...) to return a (status, stdout, stderr) tuple
93	kwargs['with_extended_output'] = True	18✔
94
95	# Execute command
96	try:	18✔
97	(_, stdout, _) = cmd(args, *kwargs)	18✔
98	except GitCommandError as error:	18✔
99	# Add more meta-information to errors
100	message = "'{}' returned exit status {}".format(	18✔
101	' '.join(str(c) for c in error.command),
102	error.status
103	)
104
105	raise GitError(message, stderr=error.stderr, stdout=stdout)	18✔
106
107	return stdout.strip()	18✔
108
109	def __getattr__(self, name):	18✔
110	return lambda args, kwargs: self._run(name, args, **kwargs)	18✔
111
112	###########################################################################
113	# Overwrite some methods and add new ones
114	###########################################################################
115
116	@contextmanager	18✔
117	def stasher(self):	18✔
118	"""
119	A stashing contextmanager.
120	"""
121	# nonlocal for python2
122	stashed = [False]	18✔
123	clean = [False]	18✔
124
125	def stash():	18✔
126	if clean[0] or not self.repo.is_dirty(submodules=False):	18✔
127	clean[0] = True	18✔
128	return	18✔
129	if stashed[0]:	18✔
130	return	×
131
132	if self.change_count > 1:	18✔
133	message = 'stashing {0} changes'	×
134	else:
135	message = 'stashing {0} change'	18✔
136	print(colored(	18✔
137	message.format(self.change_count),
138	'magenta'
139	))
140	try:	18✔
141	self._run('stash')	18✔
142	except GitError as git_error:	18✔
143	raise StashError(stderr=git_error.stderr, stdout=git_error.stdout)	18✔
144
145	stashed[0] = True	18✔
146
147	yield stash	18✔
148
149	if stashed[0]:	18✔
150	print(colored('unstashing', 'magenta'))	18✔
151	try:	18✔
152	self._run('stash', 'pop')	18✔
153	except GitError as e:	18✔
154	raise UnstashError(stderr=e.stderr, stdout=e.stdout)	18✔
155
156	def checkout(self, branch_name):	18✔
157	""" Checkout a branch by name. """
158	try:	18✔
159	find(	18✔
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):	18✔
166	""" Rebase to target branch. """
167	current_branch = self.repo.active_branch	18✔
168
169	arguments = (	18✔
170	([self.config('git-up.rebase.arguments')] or []) +
171	[target_branch.name]
172	)
173	try:	18✔
174	self._run('rebase', *arguments)	18✔
175	except GitError as e:	18✔
176	raise RebaseError(current_branch.name, target_branch.name,	18✔
177	**e.__dict__)
178
179	def fetch(self, args, *kwargs):	18✔
180	""" Fetch remote commits. """
181
182	# Execute command
183	cmd = self.git.fetch(as_process=True, args, *kwargs)	18✔
184
185	return self.run_cmd(cmd)	18✔
186
187	def push(self, args, *kwargs):	18✔
188	""" Push commits to remote """
189	# Execute command
190	cmd = self.git.push(as_process=True, args, *kwargs)	18✔
191
192	return self.run_cmd(cmd)	18✔
193
194	@staticmethod	18✔
195	def stream_reader(input_stream: BufferedReader, output_stream: Optional[IO], result_list: List[str]) -> None:	18✔
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""	18✔
204	while True:	12✔
205	read_byte = input_stream.read(1)	18✔
206	captured_bytes += read_byte	18✔
207	if output_stream is not None:	18✔
208	output_stream.write(read_byte.decode('utf-8'))	18✔
209	output_stream.flush()	18✔
210	if read_byte == b"":	18✔
211	break	18✔
212	result_list.append(captured_bytes)	18✔
213
214	@staticmethod	18✔
215	def run_cmd(cmd: GitCmd.AutoInterrupt) -> bytes:	18✔
216	""" Run a command and return stdout. """
217	std_outs = []	18✔
218	std_errs = []	18✔
219	stdout_thread = Thread(target=GitWrapper.stream_reader,	18✔
220	args=(cmd.stdout, sys.stdout, std_outs))
221	stderr_thread = Thread(target=GitWrapper.stream_reader,	18✔
222	args=(cmd.stderr, None, std_errs))
223
224	# Wait for the process to quit
225	try:	18✔
226	stdout_thread.start()	18✔
227	stderr_thread.start()	18✔
228	cmd.wait()	18✔
229	stdout_thread.join()	18✔
230	stderr_thread.join()	18✔
231	except GitCommandError as error:	18✔
232	# Add more meta-information to errors
233	message = "'{}' returned exit status {}".format(	18✔
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] if std_outs else None)	18✔
239
240	return std_outs[0].strip() if std_outs else bytes()	18✔
241
242	def config(self, key):	18✔
243	""" Return `git config key` output or None. """
244	try:	18✔
245	return self.git.config(key)	18✔
246	except GitCommandError:	18✔
247	return None	18✔
248
249	@property	18✔
250	def change_count(self):	18✔
251	""" The number of changes in the working directory. """
252	status = self.git.status(porcelain=True, untracked_files='no').strip()	18✔
253	if not status:	18✔
254	return 0	×
255	else:
256	return len(status.split('\n'))	18✔
257
258	@property	18✔
259	def version(self):	18✔
260	"""
261	Return git's version as a list of numbers.
262
263	The original repo.git.version_info has problems with tome types of
264	git version strings.
265	"""
266	return re.search(r'\d+(\.\d+)+', self.git.version()).group(0)	18✔
267
268	def is_version_min(self, required_version):	18✔
269	""" Does git's version match the requirements? """
270	return self.version.split('.') >= required_version.split('.')	18✔
271
272
273	###############################################################################
274	# GitError + subclasses
275	###############################################################################
276
277	class GitError(Exception):	18✔
278	"""
279	Extension of the GitCommandError class.
280
281	New:
282	- stdout
283	- details: a 'nested' exception with more details
284	"""
285
286	def __init__(self, message=None, stderr=None, stdout=None, details=None):	18✔
287	# super(GitError, self).__init__((), None, stderr)
288	self.details = details	18✔
289	self.message = message	18✔
290
291	self.stderr = stderr	18✔
292	self.stdout = stdout	18✔
293
294	def __str__(self): # pragma: no cover
295	return self.message
296
297
298	class StashError(GitError):	18✔
299	"""
300	Error while stashing
301	"""
302
303	def __init__(self, **kwargs):	18✔
304	kwargs.pop('message', None)	18✔
305	GitError.__init__(self, 'Stashing failed!', **kwargs)	18✔
306
307
308	class UnstashError(GitError):	18✔
309	"""
310	Error while unstashing
311	"""
312
313	def __init__(self, **kwargs):	18✔
314	kwargs.pop('message', None)	18✔
315	GitError.__init__(self, 'Unstashing failed!', **kwargs)	18✔
316
317
318	class CheckoutError(GitError):	18✔
319	"""
320	Error during checkout
321	"""
322
323	def __init__(self, branch_name, **kwargs):	18✔
324	kwargs.pop('message', None)	×
325	GitError.__init__(self, 'Failed to checkout ' + branch_name,	×
326	**kwargs)
327
328
329	class RebaseError(GitError):	18✔
330	"""
331	Error during rebase command
332	"""
333
334	def __init__(self, current_branch, target_branch, **kwargs):	18✔
335	# Remove kwargs we won't pass to GitError
336	kwargs.pop('message', None)	18✔
337	kwargs.pop('command', None)	18✔
338	kwargs.pop('status', None)	18✔
339
340	message = "Failed to rebase {1} onto {0}".format(	18✔
341	current_branch, target_branch
342	)
343	GitError.__init__(self, message, **kwargs)	18✔

msiemens / PyGitUp / 11756390499

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