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

92.31

/PyGitUp/gitup.py

from git import Git
from git import GitCommandNotFound

__all__ = ['GitUp']

###############################################################################
# IMPORTS and LIBRARIES SETUP
###############################################################################

# Python libs
import argparse
import codecs
import errno
import sys
import os
import re
import json
import subprocess
from io import StringIO
from tempfile import NamedTemporaryFile
from urllib.error import HTTPError, URLError
from urllib.request import urlopen

# 3rd party libs
try:
    # noinspection PyUnresolvedReferences
    import pkg_resources as pkg
except ImportError:  # pragma: no cover
    NO_DISTRIBUTE = True
else:  # pragma: no cover
    NO_DISTRIBUTE = False

import colorama
from git import Repo, GitCmdObjectDB
from termcolor import colored

# PyGitUp libs
from PyGitUp.utils import execute, uniq, find
from PyGitUp.git_wrapper import GitWrapper, GitError

ON_WINDOWS = sys.platform == 'win32'

###############################################################################
# Setup of 3rd party libs
###############################################################################

colorama.init(autoreset=True, convert=ON_WINDOWS)

###############################################################################
# Setup constants
###############################################################################

PYPI_URL = 'https://pypi.python.org/pypi/git-up/json'


###############################################################################
# GitUp
###############################################################################

def get_git_dir():
    toplevel_dir = execute(['git', 'rev-parse', '--show-toplevel'])

    if toplevel_dir is not None \
            and os.path.isfile(os.path.join(toplevel_dir, '.git')):
        # Not a normal git repo. Check if it's a submodule, then use
        # toplevel_dir. Otherwise it's a worktree, thus use  common_dir.
        # NOTE: git worktree support only comes with git v2.5.0 or
        # later, on earlier versions toplevel_dir is the best we can do.

        cmd = ['git', 'rev-parse', '--is-inside-work-tree']
        inside_worktree = execute(cmd, cwd=os.path.join(toplevel_dir, '..'))

        if inside_worktree == 'true' or Git().version_info[:3] < (2, 5, 0):
            return toplevel_dir
        else:
            return execute(['git', 'rev-parse', '--git-common-dir'])

    return toplevel_dir


class GitUp:
    """ Conainter class for GitUp methods """

    default_settings = {
        'fetch.prune': True,
        'fetch.all': False,
        'rebase.show-hashes': False,
        'rebase.arguments': None,
        'rebase.auto': True,
        'rebase.log-hook': None,
        'updates.check': True,
        'push.auto': False,
        'push.tags': False,
        'push.all': False,
    }

    def __init__(self, testing=False, sparse=False):
        # Sparse init: config only
        if sparse:
            self.git = GitWrapper(None)

            # Load configuration
            self.settings = self.default_settings.copy()
            self.load_config()
            return

        # Testing: redirect stderr to stdout
        self.testing = testing
        if self.testing:
            self.stderr = sys.stdout  # Quiet testing
        else:  # pragma: no cover
            self.stderr = sys.stderr

        self.states = []
        self.should_fetch = True
        self.pushed = False

        # Check, if we're in a git repo
        try:
            repo_dir = get_git_dir()
        except (OSError, GitCommandNotFound) as e:
            if isinstance(e, GitCommandNotFound) or e.errno == errno.ENOENT:
                exc = GitError("The git executable could not be found")
                raise exc
            else:
                raise
        else:
            if repo_dir is None:
                exc = GitError("We don't seem to be in a git repository.")
                raise exc

            self.repo = Repo(repo_dir, odbt=GitCmdObjectDB)

        # Check for branch tracking information
        if not any(b.tracking_branch() for b in self.repo.branches):
            exc = GitError("Can\'t update your repo because it doesn\'t has "
                           "any branches with tracking information.")
            self.print_error(exc)

            raise exc

        self.git = GitWrapper(self.repo)

        # target_map: map local branch names to remote tracking branches
        #: :type: dict[str, git.refs.remote.RemoteReference]
        self.target_map = dict()

        for branch in self.repo.branches:
            target = branch.tracking_branch()

            if target:
                if target.name.startswith('./'):
                    # Tracking branch is in local repo
                    target.is_local = True
                else:
                    target.is_local = False

                self.target_map[branch.name] = target

        # branches: all local branches with tracking information
        #: :type: list[git.refs.head.Head]
        self.branches = [b for b in self.repo.branches if b.tracking_branch()]
        self.branches.sort(key=lambda br: br.name)

        # remotes: all remotes that are associated with local branches
        #: :type: list[git.refs.remote.RemoteReference]
        self.remotes = uniq(
            # name = '<remote>/<branch>' -> '<remote>'
            [r.name.split('/', 2)[0]
             for r in list(self.target_map.values())]
        )

        # change_count: Number of unstaged changes
        self.change_count = len(
            self.git.status(porcelain=True, untracked_files='no').split('\n')
        )

        # Load configuration
        self.settings = self.default_settings.copy()
        self.load_config()

    def run(self):
        """ Run all the git-up stuff. """
        try:
            if self.should_fetch:
                self.fetch()

            self.rebase_all_branches()

            if self.settings['push.auto']:
                self.push()

        except GitError as error:
            self.print_error(error)

            # Used for test cases
            if self.testing:
                raise
            else:  # pragma: no cover
                sys.exit(1)
        except KeyboardInterrupt:
            sys.exit(130)

    def rebase_all_branches(self):
        """ Rebase all branches, if possible. """
        col_width = max(len(b.name) for b in self.branches) + 1
        if self.repo.head.is_detached:
            raise GitError("You're not currently on a branch. I'm exiting"
                           " in case you're in the middle of something.")
        original_branch = self.repo.active_branch

        with self.git.stasher() as stasher:
            for branch in self.branches:
                target = self.target_map[branch.name]

                # Print branch name
                if branch.name == original_branch.name:
                    attrs = ['bold']
                else:
                    attrs = []
                print(colored(branch.name.ljust(col_width), attrs=attrs),
                        end=' ')

                # Check, if target branch exists
                try:
                    if target.name.startswith('./'):
                        # Check, if local branch exists
                        self.git.rev_parse(target.name[2:])
                    else:
                        # Check, if remote branch exists
                        _ = target.commit

                except (ValueError, GitError):
                    # Remote branch doesn't exist!
                    print(colored('error: remote branch doesn\'t exist', 'red'))
                    self.states.append('remote branch doesn\'t exist')

                    continue

                # Get tracking branch
                if target.is_local:
                    target = find(self.repo.branches,
                                  lambda b: b.name == target.name[2:])

                # Check status and act appropriately
                if target.commit.hexsha == branch.commit.hexsha:
                    print(colored('up to date', 'green'))
                    self.states.append('up to date')

                    continue  # Do not do anything

                base = self.git.merge_base(branch.name, target.name)

                if base == target.commit.hexsha:
                    print(colored('ahead of upstream', 'cyan'))
                    self.states.append('ahead')

                    continue  # Do not do anything

                fast_fastforward = False
                if base == branch.commit.hexsha:
                    print(colored('fast-forwarding...', 'yellow'), end='')
                    self.states.append('fast-forwarding')
                    # Don't fast fast-forward the currently checked-out branch
                    fast_fastforward = (branch.name !=
                                        self.repo.active_branch.name)

                elif not self.settings['rebase.auto']:
                    print(colored('diverged', 'red'))
                    self.states.append('diverged')

                    continue  # Do not do anything
                else:
                    print(colored('rebasing', 'yellow'), end='')
                    self.states.append('rebasing')

                if self.settings['rebase.show-hashes']:
                    print(' {}..{}'.format(base[0:7],
                                           target.commit.hexsha[0:7]))
                else:
                    print()

                self.log(branch, target)
                if fast_fastforward:
                    branch.commit = target.commit
                else:
                    stasher()
                    self.git.checkout(branch.name)
                    self.git.rebase(target)

            if (self.repo.head.is_detached  # Only on Travis CI,
                    # we get a detached head after doing our rebase *confused*.
                    # Running self.repo.active_branch would fail.
                    or not self.repo.active_branch.name == original_branch.name):
                print(colored(f'returning to {original_branch.name}',
                              'magenta'))
                original_branch.checkout()

    def fetch(self):
        """
        Fetch the recent refs from the remotes.

        Unless git-up.fetch.all is set to true, all remotes with
        locally existent branches will be fetched.
        """
        fetch_kwargs = {'multiple': True}
        fetch_args = []

        if self.is_prune():
            fetch_kwargs['prune'] = True

        if self.settings['fetch.all']:
            fetch_kwargs['all'] = True
        else:
            if '.' in self.remotes:
                self.remotes.remove('.')

                if not self.remotes:
                    # Only local target branches,
                    # `git fetch --multiple` will fail
                    return

            fetch_args.append(self.remotes)

        try:
            self.git.fetch(*fetch_args, **fetch_kwargs)
        except GitError as error:
            error.message = "`git fetch` failed"
            raise error

    def push(self):
        """
        Push the changes back to the remote(s) after fetching
        """
        print('pushing...')
        push_kwargs = {}
        push_args = []

        if self.settings['push.tags']:
            push_kwargs['push'] = True

        if self.settings['push.all']:
            push_kwargs['all'] = True
        else:
            if '.' in self.remotes:
                self.remotes.remove('.')

                if not self.remotes:
                    # Only local target branches,
                    # `git push` will fail
                    return

            push_args.append(self.remotes)

        try:
            self.git.push(*push_args, **push_kwargs)
            self.pushed = True
        except GitError as error:
            error.message = "`git push` failed"
            raise error

    def log(self, branch, remote):
        """ Call a log-command, if set by git-up.fetch.all. """
        log_hook = self.settings['rebase.log-hook']

        if log_hook:
            if ON_WINDOWS:  # pragma: no cover
                # Running a string in CMD from Python is not that easy on
                # Windows. Running 'cmd /C log_hook' produces problems when
                # using multiple statements or things like 'echo'. Therefore,
                # we write the string to a bat file and execute it.

                # In addition, we replace occurrences of $1 with %1 and so forth
                # in case the user is used to Bash or sh.
                # If there are occurrences of %something, we'll replace it with
                # %%something. This is the case when running something like
                # 'git log --pretty=format:"%Cred%h..."'.
                # Also, we replace a semicolon with a newline, because if you
                # start with 'echo' on Windows, it will simply echo the
                # semicolon and the commands behind instead of echoing and then
                # running other commands

                # Prepare log_hook
                log_hook = re.sub(r'\$(\d+)', r'%\1', log_hook)
                log_hook = re.sub(r'%(?!\d)', '%%', log_hook)
                log_hook = re.sub(r'; ?', r'\n', log_hook)

                # Write log_hook to an temporary file and get it's path
                with NamedTemporaryFile(
                        prefix='PyGitUp.', suffix='.bat', delete=False
                ) as bat_file:
                    # Don't echo all commands
                    bat_file.file.write(b'@echo off\n')
                    # Run log_hook
                    bat_file.file.write(log_hook.encode('utf-8'))

                # Run bat_file
                state = subprocess.call(
                    [bat_file.name, branch.name, remote.name]
                )

                # Clean up file
                os.remove(bat_file.name)
            else:  # pragma: no cover
                # Run log_hook via 'shell -c'
                state = subprocess.call(
                    [log_hook, 'git-up', branch.name, remote.name],
                    shell=True
                )

            if self.testing:
                assert state == 0, 'log_hook returned != 0'

    def version_info(self):
        """ Tell, what version we're running at and if it's up to date. """

        # Retrive and show local version info
        package = pkg.get_distribution('git-up')
        local_version_str = package.version
        local_version = package.parsed_version

        print('GitUp version is: ' + colored('v' + local_version_str, 'green'))

        if not self.settings['updates.check']:
            return

        # Check for updates
        print('Checking for updates...', end='')

        try:
            # Get version information from the PyPI JSON API
            reader = codecs.getreader('utf-8')
            details = json.load(reader(urlopen(PYPI_URL)))
            online_version = details['info']['version']
        except (HTTPError, URLError, ValueError):
            recent = True  # To not disturb the user with HTTP/parsing errors
        else:
            recent = local_version >= pkg.parse_version(online_version)

        if not recent:
            # noinspection PyUnboundLocalVariable
            print(
                '\rRecent version is: '
                + colored('v' + online_version, color='yellow', attrs=['bold'])
            )
            print('Run \'pip install -U git-up\' to get the update.')
        else:
            # Clear the update line
            sys.stdout.write('\r' + ' ' * 80 + '\n')

    ###########################################################################
    # Helpers
    ###########################################################################

    def load_config(self):
        """
        Load the configuration from git config.
        """
        for key in self.settings:
            value = self.config(key)
            # Parse true/false
            if value == '' or value is None:
                continue  # Not set by user, go on
            if value.lower() == 'true':
                value = True
            elif value.lower() == 'false':
                value = False
            elif value:
                pass  # A user-defined string, store the value later

            self.settings[key] = value

    def config(self, key):
        """ Get a git-up-specific config value. """
        return self.git.config(f'git-up.{key}')

    def is_prune(self):
        """
        Return True, if `git fetch --prune` is allowed.

        Because of possible incompatibilities, this requires special
        treatment.
        """
        required_version = "1.6.6"
        config_value = self.settings['fetch.prune']

        if self.git.is_version_min(required_version):
            return config_value is not False
        else:  # pragma: no cover
            if config_value == 'true':
                print(colored(
                    "Warning: fetch.prune is set to 'true' but your git"
                    "version doesn't seem to support it ({} < {})."
                    "Defaulting to 'false'.".format(self.git.version,
                                                    required_version),
                    'yellow'
                ))

    def print_error(self, error):
        """
        Print more information about an error.

        :type error: GitError
        """
        print(colored(error.message, 'red'), file=self.stderr)

        if error.stdout or error.stderr:
            print(file=self.stderr)
            print("Here's what git said:", file=self.stderr)
            print(file=self.stderr)

            if error.stdout:
                print(error.stdout, file=self.stderr)
            if error.stderr:
                print(error.stderr, file=self.stderr)

        if error.details:
            print(file=self.stderr)
            print("Here's what we know:", file=self.stderr)
            print(str(error.details), file=self.stderr)
            print(file=self.stderr)


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


EPILOG = '''
For configuration options, please see
https://github.com/msiemens/PyGitUp#readme.

\b
Python port of https://github.com/aanand/git-up/
Project Author: Markus Siemens <markus@m-siemens.de>
Project URL: https://github.com/msiemens/PyGitUp
\b
'''


def run():  # pragma: no cover
    """
    A nicer `git pull`.
    """

    parser = argparse.ArgumentParser(description="A nicer `git pull`.", epilog=EPILOG)
    parser.add_argument('-V', '--version', action='store_true',
                        help='Show version (and if there is a newer version).')
    parser.add_argument('-q', '--quiet', action='store_true',
                        help='Be quiet, only print error messages.')
    parser.add_argument('--no-fetch', '--no-f', dest='fetch', action='store_false',
                        help='Don\'t try to fetch from origin.')
    parser.add_argument('-p', '--push', action='store_true',
                        help='Push the changes after pulling successfully.')

    args = parser.parse_args()

    if args.version:
        if NO_DISTRIBUTE:
            print(colored('Please install \'git-up\' via pip in order to '
                          'get version information.', 'yellow'))
        else:
            GitUp(sparse=True).version_info()
        return

    if args.quiet:
        sys.stdout = StringIO()

    try:
        gitup = GitUp()
        gitup.settings['push.auto'] = args.push
        gitup.should_fetch = args.fetch
    except GitError:
        sys.exit(1)  # Error in constructor
    else:
        gitup.run()


if __name__ == '__main__':  # pragma: no cover
    run()

1	from git import Git	18✔
2	from git import GitCommandNotFound	18✔
3
4	__all__ = ['GitUp']	18✔
5
6	###############################################################################
7	# IMPORTS and LIBRARIES SETUP
8	###############################################################################
9
10	# Python libs
11	import argparse	18✔
12	import codecs	18✔
13	import errno	18✔
14	import sys	18✔
15	import os	18✔
16	import re	18✔
17	import json	18✔
18	import subprocess	18✔
19	from io import StringIO	18✔
20	from tempfile import NamedTemporaryFile	18✔
21	from urllib.error import HTTPError, URLError	18✔
22	from urllib.request import urlopen	18✔
23
24	# 3rd party libs
25	try:	18✔
26	# noinspection PyUnresolvedReferences
27	import pkg_resources as pkg	18✔
28	except ImportError: # pragma: no cover
29	NO_DISTRIBUTE = True
30	else: # pragma: no cover
31	NO_DISTRIBUTE = False
32
33	import colorama	18✔
34	from git import Repo, GitCmdObjectDB	18✔
35	from termcolor import colored	18✔
36
37	# PyGitUp libs
38	from PyGitUp.utils import execute, uniq, find	18✔
39	from PyGitUp.git_wrapper import GitWrapper, GitError	18✔
40
41	ON_WINDOWS = sys.platform == 'win32'	18✔
42
43	###############################################################################
44	# Setup of 3rd party libs
45	###############################################################################
46
47	colorama.init(autoreset=True, convert=ON_WINDOWS)	18✔
48
49	###############################################################################
50	# Setup constants
51	###############################################################################
52
53	PYPI_URL = 'https://pypi.python.org/pypi/git-up/json'	18✔
54
55
56	###############################################################################
57	# GitUp
58	###############################################################################
59
60	def get_git_dir():	18✔
61	toplevel_dir = execute(['git', 'rev-parse', '--show-toplevel'])	18✔
62
63	if toplevel_dir is not None \	18✔
64	and os.path.isfile(os.path.join(toplevel_dir, '.git')):
65	# Not a normal git repo. Check if it's a submodule, then use
66	# toplevel_dir. Otherwise it's a worktree, thus use common_dir.
67	# NOTE: git worktree support only comes with git v2.5.0 or
68	# later, on earlier versions toplevel_dir is the best we can do.
69
70	cmd = ['git', 'rev-parse', '--is-inside-work-tree']	18✔
71	inside_worktree = execute(cmd, cwd=os.path.join(toplevel_dir, '..'))	18✔
72
73	if inside_worktree == 'true' or Git().version_info[:3] < (2, 5, 0):	18✔
74	return toplevel_dir	18✔
75	else:
76	return execute(['git', 'rev-parse', '--git-common-dir'])	18✔
77
78	return toplevel_dir	18✔
79
80
81	class GitUp:	18✔
82	""" Conainter class for GitUp methods """
83
84	default_settings = {	18✔
85	'fetch.prune': True,
86	'fetch.all': False,
87	'rebase.show-hashes': False,
88	'rebase.arguments': None,
89	'rebase.auto': True,
90	'rebase.log-hook': None,
91	'updates.check': True,
92	'push.auto': False,
93	'push.tags': False,
94	'push.all': False,
95	}
96
97	def __init__(self, testing=False, sparse=False):	18✔
98	# Sparse init: config only
99	if sparse:	18✔
100	self.git = GitWrapper(None)	12✔
101
102	# Load configuration
103	self.settings = self.default_settings.copy()	12✔
104	self.load_config()	12✔
105	return	12✔
106
107	# Testing: redirect stderr to stdout
108	self.testing = testing	18✔
109	if self.testing:
110	self.stderr = sys.stdout # Quiet testing
111	else: # pragma: no cover
112	self.stderr = sys.stderr
113
114	self.states = []	18✔
115	self.should_fetch = True	18✔
116	self.pushed = False	18✔
117
118	# Check, if we're in a git repo
119	try:	18✔
120	repo_dir = get_git_dir()	18✔
121	except (OSError, GitCommandNotFound) as e:	18✔
122	if isinstance(e, GitCommandNotFound) or e.errno == errno.ENOENT:	18✔
123	exc = GitError("The git executable could not be found")	18✔
124	raise exc	18✔
125	else:
126	raise	3✔
127	else:
128	if repo_dir is None:	18✔
129	exc = GitError("We don't seem to be in a git repository.")	18✔
130	raise exc	18✔
131
132	self.repo = Repo(repo_dir, odbt=GitCmdObjectDB)	18✔
133
134	# Check for branch tracking information
135	if not any(b.tracking_branch() for b in self.repo.branches):	18✔
136	exc = GitError("Can\'t update your repo because it doesn\'t has "	18✔
137	"any branches with tracking information.")
138	self.print_error(exc)	18✔
139
140	raise exc	18✔
141
142	self.git = GitWrapper(self.repo)	18✔
143
144	# target_map: map local branch names to remote tracking branches
145	#: :type: dict[str, git.refs.remote.RemoteReference]
146	self.target_map = dict()	18✔
147
148	for branch in self.repo.branches:	18✔
149	target = branch.tracking_branch()	18✔
150
151	if target:	18✔
152	if target.name.startswith('./'):	18✔
153	# Tracking branch is in local repo
154	target.is_local = True	18✔
155	else:
156	target.is_local = False	18✔
157
158	self.target_map[branch.name] = target	18✔
159
160	# branches: all local branches with tracking information
161	#: :type: list[git.refs.head.Head]
162	self.branches = [b for b in self.repo.branches if b.tracking_branch()]	18✔
163	self.branches.sort(key=lambda br: br.name)	18✔
164
165	# remotes: all remotes that are associated with local branches
166	#: :type: list[git.refs.remote.RemoteReference]
167	self.remotes = uniq(	18✔
168	# name = '<remote>/<branch>' -> '<remote>'
169	[r.name.split('/', 2)[0]
170	for r in list(self.target_map.values())]
171	)
172
173	# change_count: Number of unstaged changes
174	self.change_count = len(	18✔
175	self.git.status(porcelain=True, untracked_files='no').split('\n')
176	)
177
178	# Load configuration
179	self.settings = self.default_settings.copy()	18✔
180	self.load_config()	18✔
181
182	def run(self):	18✔
183	""" Run all the git-up stuff. """
184	try:	18✔
185	if self.should_fetch:	18✔
186	self.fetch()	18✔
187
188	self.rebase_all_branches()	18✔
189
190	if self.settings['push.auto']:	18✔
191	self.push()	18✔
192
193	except GitError as error:	18✔
194	self.print_error(error)	18✔
195
196	# Used for test cases
197	if self.testing:
198	raise
199	else: # pragma: no cover
200	sys.exit(1)
201	except KeyboardInterrupt:	18✔
202	sys.exit(130)	18✔
203
204	def rebase_all_branches(self):	18✔
205	""" Rebase all branches, if possible. """
206	col_width = max(len(b.name) for b in self.branches) + 1	18✔
207	if self.repo.head.is_detached:	18✔
208	raise GitError("You're not currently on a branch. I'm exiting"	18✔
209	" in case you're in the middle of something.")
210	original_branch = self.repo.active_branch	18✔
211
212	with self.git.stasher() as stasher:	18✔
213	for branch in self.branches:	18✔
214	target = self.target_map[branch.name]	18✔
215
216	# Print branch name
217	if branch.name == original_branch.name:	18✔
218	attrs = ['bold']	18✔
219	else:
220	attrs = []	18✔
221	print(colored(branch.name.ljust(col_width), attrs=attrs),	18✔
222	end=' ')
223
224	# Check, if target branch exists
225	try:	18✔
226	if target.name.startswith('./'):	18✔
227	# Check, if local branch exists
228	self.git.rev_parse(target.name[2:])	18✔
229	else:
230	# Check, if remote branch exists
231	_ = target.commit	18✔
232
233	except (ValueError, GitError):	18✔
234	# Remote branch doesn't exist!
235	print(colored('error: remote branch doesn\'t exist', 'red'))	18✔
236	self.states.append('remote branch doesn\'t exist')	18✔
237
238	continue	18✔
239
240	# Get tracking branch
241	if target.is_local:	18✔
242	target = find(self.repo.branches,	18✔
243	lambda b: b.name == target.name[2:])
244
245	# Check status and act appropriately
246	if target.commit.hexsha == branch.commit.hexsha:	18✔
247	print(colored('up to date', 'green'))	18✔
248	self.states.append('up to date')	18✔
249
250	continue # Do not do anything	18✔
251
252	base = self.git.merge_base(branch.name, target.name)	18✔
253
254	if base == target.commit.hexsha:	18✔
255	print(colored('ahead of upstream', 'cyan'))	18✔
256	self.states.append('ahead')	18✔
257
258	continue # Do not do anything	18✔
259
260	fast_fastforward = False	18✔
261	if base == branch.commit.hexsha:	18✔
262	print(colored('fast-forwarding...', 'yellow'), end='')	18✔
263	self.states.append('fast-forwarding')	18✔
264	# Don't fast fast-forward the currently checked-out branch
265	fast_fastforward = (branch.name !=	18✔
266	self.repo.active_branch.name)
267
268	elif not self.settings['rebase.auto']:	18✔
269	print(colored('diverged', 'red'))	18✔
270	self.states.append('diverged')	18✔
271
272	continue # Do not do anything	18✔
273	else:
274	print(colored('rebasing', 'yellow'), end='')	18✔
275	self.states.append('rebasing')	18✔
276
277	if self.settings['rebase.show-hashes']:	18✔
278	print(' {}..{}'.format(base[0:7],	×
279	target.commit.hexsha[0:7]))
280	else:
281	print()	18✔
282
283	self.log(branch, target)	18✔
284	if fast_fastforward:	18✔
285	branch.commit = target.commit	18✔
286	else:
287	stasher()	18✔
288	self.git.checkout(branch.name)	18✔
289	self.git.rebase(target)	18✔
290
291	if (self.repo.head.is_detached # Only on Travis CI,	18✔
292	# we get a detached head after doing our rebase confused.
293	# Running self.repo.active_branch would fail.
294	or not self.repo.active_branch.name == original_branch.name):
295	print(colored(f'returning to {original_branch.name}',	18✔
296	'magenta'))
297	original_branch.checkout()	18✔
298
299	def fetch(self):	18✔
300	"""
301	Fetch the recent refs from the remotes.
302
303	Unless git-up.fetch.all is set to true, all remotes with
304	locally existent branches will be fetched.
305	"""
306	fetch_kwargs = {'multiple': True}	18✔
307	fetch_args = []	18✔
308
309	if self.is_prune():	18✔
310	fetch_kwargs['prune'] = True	18✔
311
312	if self.settings['fetch.all']:	18✔
313	fetch_kwargs['all'] = True	18✔
314	else:
315	if '.' in self.remotes:	18✔
316	self.remotes.remove('.')	18✔
317
318	if not self.remotes:	18✔
319	# Only local target branches,
320	# `git fetch --multiple` will fail
321	return	18✔
322
323	fetch_args.append(self.remotes)	18✔
324
325	try:	18✔
326	self.git.fetch(fetch_args, *fetch_kwargs)	18✔
327	except GitError as error:	18✔
328	error.message = "`git fetch` failed"	18✔
329	raise error	18✔
330
331	def push(self):	18✔
332	"""
333	Push the changes back to the remote(s) after fetching
334	"""
335	print('pushing...')	18✔
336	push_kwargs = {}	18✔
337	push_args = []	18✔
338
339	if self.settings['push.tags']:	18✔
340	push_kwargs['push'] = True	×
341
342	if self.settings['push.all']:	18✔
343	push_kwargs['all'] = True	×
344	else:
345	if '.' in self.remotes:	18✔
346	self.remotes.remove('.')	×
347
348	if not self.remotes:	×
349	# Only local target branches,
350	# `git push` will fail
351	return	×
352
353	push_args.append(self.remotes)	18✔
354
355	try:	18✔
356	self.git.push(push_args, *push_kwargs)	18✔
357	self.pushed = True	18✔
358	except GitError as error:	×
359	error.message = "`git push` failed"	×
360	raise error	×
361
362	def log(self, branch, remote):	18✔
363	""" Call a log-command, if set by git-up.fetch.all. """
364	log_hook = self.settings['rebase.log-hook']	18✔
365
366	if log_hook:	18✔
367	if ON_WINDOWS: # pragma: no cover
368	# Running a string in CMD from Python is not that easy on
369	# Windows. Running 'cmd /C log_hook' produces problems when
370	# using multiple statements or things like 'echo'. Therefore,
371	# we write the string to a bat file and execute it.
372
373	# In addition, we replace occurrences of $1 with %1 and so forth
374	# in case the user is used to Bash or sh.
375	# If there are occurrences of %something, we'll replace it with
376	# %%something. This is the case when running something like
377	# 'git log --pretty=format:"%Cred%h..."'.
378	# Also, we replace a semicolon with a newline, because if you
379	# start with 'echo' on Windows, it will simply echo the
380	# semicolon and the commands behind instead of echoing and then
381	# running other commands
382
383	# Prepare log_hook
384	log_hook = re.sub(r'\$(\d+)', r'%\1', log_hook)
385	log_hook = re.sub(r'%(?!\d)', '%%', log_hook)
386	log_hook = re.sub(r'; ?', r'\n', log_hook)
387
388	# Write log_hook to an temporary file and get it's path
389	with NamedTemporaryFile(
390	prefix='PyGitUp.', suffix='.bat', delete=False
391	) as bat_file:
392	# Don't echo all commands
393	bat_file.file.write(b'@echo off\n')
394	# Run log_hook
395	bat_file.file.write(log_hook.encode('utf-8'))
396
397	# Run bat_file
398	state = subprocess.call(
399	[bat_file.name, branch.name, remote.name]
400	)
401
402	# Clean up file
403	os.remove(bat_file.name)
404	else: # pragma: no cover
405	# Run log_hook via 'shell -c'
406	state = subprocess.call(
407	[log_hook, 'git-up', branch.name, remote.name],
408	shell=True
409	)
410
411	if self.testing:
412	assert state == 0, 'log_hook returned != 0'
413
414	def version_info(self):	18✔
415	""" Tell, what version we're running at and if it's up to date. """
416
417	# Retrive and show local version info
418	package = pkg.get_distribution('git-up')	12✔
419	local_version_str = package.version	12✔
420	local_version = package.parsed_version	12✔
421
422	print('GitUp version is: ' + colored('v' + local_version_str, 'green'))	12✔
423
424	if not self.settings['updates.check']:	12✔
425	return	×
426
427	# Check for updates
428	print('Checking for updates...', end='')	12✔
429
430	try:	12✔
431	# Get version information from the PyPI JSON API
432	reader = codecs.getreader('utf-8')	12✔
433	details = json.load(reader(urlopen(PYPI_URL)))	12✔
434	online_version = details['info']['version']	12✔
435	except (HTTPError, URLError, ValueError):	×
436	recent = True # To not disturb the user with HTTP/parsing errors	×
437	else:
438	recent = local_version >= pkg.parse_version(online_version)	12✔
439
440	if not recent:	12✔
441	# noinspection PyUnboundLocalVariable
442	print(	×
443	'\rRecent version is: '
444	+ colored('v' + online_version, color='yellow', attrs=['bold'])
445	)
446	print('Run \'pip install -U git-up\' to get the update.')	×
447	else:
448	# Clear the update line
449	sys.stdout.write('\r' + ' ' * 80 + '\n')	12✔
450
451	###########################################################################
452	# Helpers
453	###########################################################################
454
455	def load_config(self):	18✔
456	"""
457	Load the configuration from git config.
458	"""
459	for key in self.settings:	18✔
460	value = self.config(key)	18✔
461	# Parse true/false
462	if value == '' or value is None:	18✔
463	continue # Not set by user, go on	18✔
464	if value.lower() == 'true':	18✔
465	value = True	18✔
466	elif value.lower() == 'false':	18✔
467	value = False	18✔
468	elif value:	18✔
469	pass # A user-defined string, store the value later	12✔
470
471	self.settings[key] = value	18✔
472
473	def config(self, key):	18✔
474	""" Get a git-up-specific config value. """
475	return self.git.config(f'git-up.{key}')	18✔
476
477	def is_prune(self):	18✔
478	"""
479	Return True, if `git fetch --prune` is allowed.
480
481	Because of possible incompatibilities, this requires special
482	treatment.
483	"""
484	required_version = "1.6.6"	18✔
485	config_value = self.settings['fetch.prune']	18✔
486
487	if self.git.is_version_min(required_version):	18✔
488	return config_value is not False	18✔
489	else: # pragma: no cover
490	if config_value == 'true':
491	print(colored(
492	"Warning: fetch.prune is set to 'true' but your git"
493	"version doesn't seem to support it ({} < {})."
494	"Defaulting to 'false'.".format(self.git.version,
495	required_version),
496	'yellow'
497	))
498
499	def print_error(self, error):	18✔
500	"""
501	Print more information about an error.
502
503	:type error: GitError
504	"""
505	print(colored(error.message, 'red'), file=self.stderr)	18✔
506
507	if error.stdout or error.stderr:	18✔
508	print(file=self.stderr)	18✔
509	print("Here's what git said:", file=self.stderr)	18✔
510	print(file=self.stderr)	18✔
511
512	if error.stdout:	18✔
513	print(error.stdout, file=self.stderr)	17✔
514	if error.stderr:	18✔
515	print(error.stderr, file=self.stderr)	18✔
516
517	if error.details:	18✔
518	print(file=self.stderr)	×
519	print("Here's what we know:", file=self.stderr)	×
520	print(str(error.details), file=self.stderr)	×
521	print(file=self.stderr)	×
522
523
524	###############################################################################
525
526
527	EPILOG = '''	18✔
528	For configuration options, please see
529	https://github.com/msiemens/PyGitUp#readme.
530
531	\b
532	Python port of https://github.com/aanand/git-up/
533	Project Author: Markus Siemens <markus@m-siemens.de>
534	Project URL: https://github.com/msiemens/PyGitUp
535	\b
536	'''
537
538
539	def run(): # pragma: no cover
540	"""
541	A nicer `git pull`.
542	"""
543
544	parser = argparse.ArgumentParser(description="A nicer `git pull`.", epilog=EPILOG)
545	parser.add_argument('-V', '--version', action='store_true',
546	help='Show version (and if there is a newer version).')
547	parser.add_argument('-q', '--quiet', action='store_true',
548	help='Be quiet, only print error messages.')
549	parser.add_argument('--no-fetch', '--no-f', dest='fetch', action='store_false',
550	help='Don\'t try to fetch from origin.')
551	parser.add_argument('-p', '--push', action='store_true',
552	help='Push the changes after pulling successfully.')
553
554	args = parser.parse_args()
555
556	if args.version:
557	if NO_DISTRIBUTE:
558	print(colored('Please install \'git-up\' via pip in order to '
559	'get version information.', 'yellow'))
560	else:
561	GitUp(sparse=True).version_info()
562	return
563
564	if args.quiet:
565	sys.stdout = StringIO()
566
567	try:
568	gitup = GitUp()
569	gitup.settings['push.auto'] = args.push
570	gitup.should_fetch = args.fetch
571	except GitError:
572	sys.exit(1) # Error in constructor
573	else:
574	gitup.run()
575
576
577	if __name__ == '__main__': # pragma: no cover
578	run()

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