20541811130

Committed 27 Dec 2025 04:58PM UTC coverage: 88.366% (-4.9%) from 93.284%

Build # 20541811130

Build Type

push

github

Committed by

msiemens

Commit Message

chore: update dependencies, switch to uv package manager

Run Details

357 of 404 relevant lines covered (88.37%)

8.84 hits per line

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

84.32

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

msiemens / PyGitUp / 20541811130

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