26187412830

Committed 20 May 2026 08:15PM UTC coverage: 91.253% (+0.08%) from 91.169%

Build # 26187412830

Build Type

push

github

Committed by

msiemens

Commit Message

chore: update dependencies

Coverage Stats

386 of 423 relevant lines covered (91.25%)

9.13 hits per line

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

89.02

/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:
    from importlib import metadata
except ImportError:  # pragma: no cover
    metadata = None
    NO_DISTRIBUTE = True
else:  # pragma: no cover
    NO_DISTRIBUTE = False

from packaging.version import InvalidVersion, Version

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'

def normalize_path(path):
    if ON_WINDOWS and path and path[0] == '/':
        return execute(['cygpath', '-m', path])

    return path

###############################################################################
# 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'])
    toplevel_dir = normalize_path(toplevel_dir)

    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:
            common_dir = execute(['git', 'rev-parse', '--git-common-dir'])
            return normalize_path(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:
            def _escape_positional(value):
                # Neutralize command substitution/backticks in branch names
                return value.replace('$', r'\$').replace('`', r'\`')

            branch_safe = _escape_positional(branch.name)
            remote_safe = _escape_positional(remote.name)

            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'
                # Disable globbing and word-splitting to keep $1/$2 safe
                state = subprocess.call(
                    ['sh', '-c', 'set -f; IFS=; ' + log_hook,
                     'git-up', branch_safe, remote_safe]
                )

            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
        try:
            local_version_str = metadata.version('git-up')
        except (AttributeError, metadata.PackageNotFoundError):
            print(
                colored(
                    "Please install 'git-up' via pip in order to get version information.",
                    'yellow',
                )
            )
            return

        try:
            local_version = Version(local_version_str)
        except InvalidVersion:
            print('GitUp version is: ' + colored('v' + local_version_str, 'green'))
            return

        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:
            try:
                recent = local_version >= Version(online_version)
            except InvalidVersion:
                recent = True

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

msiemens / PyGitUp / 26187412830

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