gitman.git.rebuild() - Code Metrics - Inspection of "Merge pull request #201 from mttjohnson/feature/re..." - jacebrowning/gitman - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( de828f...5a60d4 )

by Jace

created 2019-05-18 21:22 UTC

gitman.git.rebuild() A

↳ Parent: gitman.git

Complexity

Conditions

Size

Total Lines	13
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	8
nop	2
dl	0
loc	13
ccs	1
cts	1
cp	1
crap	2
rs	10
c	0
b	0
f	0

"""Utilities to call Git commands."""

import logging
import os
import re
import shutil
from contextlib import suppress

from . import common, settings
from .exceptions import ShellError
from .shell import call, pwd


log = logging.getLogger(__name__)


def git(*args, **kwargs):
    return call('git', *args, **kwargs)


def gitsvn(*args, **kwargs):
    return call('git', 'svn', *args, **kwargs)


def clone(type, repo, path, *, cache=settings.CACHE, sparse_paths=None, rev=None):
    """Clone a new Git repository."""
    log.debug("Creating a new repository...")

    if type == 'git-svn':
        # just the preperation for the svn deep clone / checkout here
        # clone will be made in update function to simplify source.py).
        os.makedirs(path)
        return

    assert type == 'git'

    name = repo.split('/')[-1]
    if name.endswith(".git"):
        name = name[:-4]

    normpath = os.path.normpath(path)
    reference = os.path.join(cache, name + ".reference")
    sparse_paths_repo = repo if settings.CACHE_DISABLE else reference

    if not settings.CACHE_DISABLE and not os.path.isdir(reference):
        git('clone', '--mirror', repo, reference)

    if sparse_paths:
        os.mkdir(normpath)
        git('-C', normpath, 'init')
        git('-C', normpath, 'config', 'core.sparseCheckout', 'true')
        git('-C', normpath, 'remote', 'add', '-f', 'origin', sparse_paths_repo)

        with open(
            "%s/%s/.git/info/sparse-checkout" % (os.getcwd(), normpath), 'w'
        ) as fd:
            fd.writelines(sparse_paths)
        with open(
            "%s/%s/.git/objects/info/alternates" % (os.getcwd(), normpath), 'w'
        ) as fd:
            fd.write("%s/objects" % sparse_paths_repo)

        # We use directly the revision requested here in order to respect,
        # that not all repos have `master` as their default branch
        git('-C', normpath, 'pull', 'origin', rev)
    elif settings.CACHE_DISABLE:
        git('clone', repo, normpath)
    else:
        git('clone', '--reference', reference, repo, normpath)


def is_sha(rev):
    """Heuristically determine whether a revision corresponds to a commit SHA.

    Any sequence of 7 to 40 hexadecimal digits will be recognized as a
    commit SHA. The minimum of 7 digits is not an arbitrary choice, it
    is the default length for short SHAs in Git.
    """
    return re.match('^[0-9a-f]{7,40}$', rev) is not None


def fetch(type, repo, path, rev=None):  # pylint: disable=unused-argument
    """Fetch the latest changes from the remote repository."""

    if type == 'git-svn':
        # deep clone happens in update function
        return

    assert type == 'git'

    git('remote', 'set-url', 'origin', repo)
    args = ['fetch', '--tags', '--force', '--prune', 'origin']
    if rev:
        if is_sha(rev):
            pass  # fetch only works with a SHA if already present locally
        elif '@' in rev:
            pass  # fetch doesn't work with rev-parse
        else:
            args.append(rev)
    git(*args)


def valid():
    """Confirm the current directory is a valid working tree.

    Checking both the git working tree exists and that the top leve
    directory path matches the current directory path.
    """

    log.debug("Checking for a valid working tree...")

    try:
        git('rev-parse', '--is-inside-work-tree', _show=False)
    except ShellError:
        return False

    log.debug("Checking for a valid git top level...")
    gittoplevel = git('rev-parse', '--show-toplevel', _show=False)
    currentdir = pwd(_show=False)

    status = False
    if gittoplevel[0] == currentdir:
        status = True
    else:
        log.debug(
            "git top level: %s != current working directory: %s",
            gittoplevel[0],
            currentdir,
        )
        status = False
    return status


def rebuild(type, repo):  # pylint: disable=unused-argument
    """Rebuild a missing repo .git directory."""

    if type == 'git-svn':
        # ignore rebuild in case of git-svn
        return

    assert type == 'git'

    common.show("Rebuilding mising git repo...", color='message')
    git('init', _show=True)
    git('remote', 'add', 'origin', repo, _show=True)
    common.show("Rebuilt git repo...", color='message')


def changes(type, include_untracked=False, display_status=True, _show=False):
    """Determine if there are changes in the working tree."""
    status = False

    if type == 'git-svn':
        # ignore changes in case of git-svn
        return status

    assert type == 'git'

    try:
        # Refresh changes
        git('update-index', '-q', '--refresh', _show=False)

        # Check for uncommitted changes
        git('diff-index', '--quiet', 'HEAD', _show=_show)

        # Check for untracked files
        lines = git('ls-files', '--others', '--exclude-standard', _show=_show)

    except ShellError:
        status = True

    else:
        status = bool(lines) and include_untracked

    if status and display_status:
        with suppress(ShellError):
            lines = git('status', _show=True)
            common.show(*lines, color='git_changes')

    return status


def update(
    type, repo, path, *, clean=True, fetch=False, rev=None
):  # pylint: disable=redefined-outer-name,unused-argument

    if type == 'git-svn':
        # make deep clone here for simplification of sources.py
        # and to realize consistent readonly clone (always forced)

        # completly empty current directory (remove also hidden content)
        for root, dirs, files in os.walk('.'):
            for f in files:
                os.unlink(os.path.join(root, f))
            for d in dirs:
                shutil.rmtree(os.path.join(root, d))

        # clone specified svn revision
        gitsvn('clone', '-r', rev, repo, '.')
        return

    assert type == 'git'

    # Update the working tree to the specified revision.
    hide = {'_show': False, '_ignore': True}

    git('stash', **hide)
    if clean:
        git('clean', '--force', '-d', '-x', _show=False)

    rev = _get_sha_from_rev(rev)
    git('checkout', '--force', rev)
    git('branch', '--set-upstream-to', 'origin/' + rev, **hide)

    if fetch:
        # if `rev` was a branch it might be tracking something older
        git('pull', '--ff-only', '--no-rebase', **hide)


def get_url(type):
    """Get the current repository's URL."""
    if type == 'git-svn':
        return git('config', '--get', 'svn-remote.svn.url', _show=False)[0]

    assert type == 'git'

    return git('config', '--get', 'remote.origin.url', _show=False)[0]


def get_hash(type, _show=False):
    """Get the current working tree's hash."""
    if type == 'git-svn':
        return ''.join(filter(str.isdigit, gitsvn('info', _show=_show)[4]))

    assert type == 'git'

    return git('rev-parse', 'HEAD', _show=_show)[0]


def get_tag():
    """Get the current working tree's tag (if on a tag)."""
    return git('describe', '--tags', '--exact-match', _show=False, _ignore=True)[0]


def is_fetch_required(type, rev):
    if type == 'git-svn':
        return False

    assert type == 'git'

    return rev not in (get_branch(), get_hash(type), get_tag())


def get_branch():
    """Get the current working tree's branch."""
    return git('rev-parse', '--abbrev-ref', 'HEAD', _show=False)[0]


def _get_sha_from_rev(rev):
    """Get a rev-parse string's hash."""
    if '@{' in rev:  # TODO: use regex for this
        parts = rev.split('@')
        branch = parts[0]
        date = parts[1].strip("{}")
        git('checkout', '--force', branch, _show=False)
        rev = git(
            'rev-list',
            '-n',
            '1',
            '--before={!r}'.format(date),
            '--first-parent',
            branch,
            _show=False,
        )[0]
    return rev


1		"""Utilities to call Git commands."""
2
3	1	import logging
4	1	import os
5	1	import re
6		import shutil
7	1	from contextlib import suppress
8	1
9	1	from . import common, settings
10		from .exceptions import ShellError
11		from .shell import call, pwd
12	1
13
14		log = logging.getLogger(__name__)
15	1
16	1
17		def git(args, *kwargs):
18		return call('git', args, *kwargs)
19	1
20
21	1	def gitsvn(args, *kwargs):
22		return call('git', 'svn', args, *kwargs)
23	1
24	1
25	1	def clone(type, repo, path, *, cache=settings.CACHE, sparse_paths=None, rev=None):
26		"""Clone a new Git repository."""
27	1	log.debug("Creating a new repository...")
28	1
29	1	if type == 'git-svn':
30		# just the preperation for the svn deep clone / checkout here
31	1	# clone will be made in update function to simplify source.py).
32		os.makedirs(path)
33		return
34	1
35		assert type == 'git'
36	1
37	1	name = repo.split('/')[-1]
38	1	if name.endswith(".git"):
39	1	name = name[:-4]
40	1
41	1	normpath = os.path.normpath(path)
42	1	reference = os.path.join(cache, name + ".reference")
43		sparse_paths_repo = repo if settings.CACHE_DISABLE else reference
44	1
45	1	if not settings.CACHE_DISABLE and not os.path.isdir(reference):
46		git('clone', '--mirror', repo, reference)
47
48	1	if sparse_paths:
49		os.mkdir(normpath)
50	1	git('-C', normpath, 'init')
51		git('-C', normpath, 'config', 'core.sparseCheckout', 'true')
52	1	git('-C', normpath, 'remote', 'add', '-f', 'origin', sparse_paths_repo)
53	1
54	1	with open(
55	1	"%s/%s/.git/info/sparse-checkout" % (os.getcwd(), normpath), 'w'
56		) as fd:
57	1	fd.writelines(sparse_paths)
58		with open(
59		"%s/%s/.git/objects/info/alternates" % (os.getcwd(), normpath), 'w'
60	1	) as fd:
61		fd.write("%s/objects" % sparse_paths_repo)
62	1
63		# We use directly the revision requested here in order to respect,
64	1	# that not all repos have `master` as their default branch
65		git('-C', normpath, 'pull', 'origin', rev)
66	1	elif settings.CACHE_DISABLE:
67		git('clone', repo, normpath)
68		else:
69	1	git('clone', '--reference', reference, repo, normpath)
70
71
72	1	def is_sha(rev):
73		"""Heuristically determine whether a revision corresponds to a commit SHA.
74	1
75	1	Any sequence of 7 to 40 hexadecimal digits will be recognized as a
76		commit SHA. The minimum of 7 digits is not an arbitrary choice, it
77		is the default length for short SHAs in Git.
78	1	"""
79		return re.match('^[0-9a-f]{7,40}$', rev) is not None
80	1
81	1
82	1	def fetch(type, repo, path, rev=None): # pylint: disable=unused-argument
83	1	"""Fetch the latest changes from the remote repository."""
84
85	1	if type == 'git-svn':
86		# deep clone happens in update function
87		return
88	1
89		assert type == 'git'
90	1
91		git('remote', 'set-url', 'origin', repo)
92	1	args = ['fetch', '--tags', '--force', '--prune', 'origin']
93	1	if rev:
94	1	if is_sha(rev):
95		pass # fetch only works with a SHA if already present locally
96	1	elif '@' in rev:
97	1	pass # fetch doesn't work with rev-parse
98	1	else:
99		args.append(rev)
100	1	git(*args)
101
102	1
103		def valid():
104		"""Confirm the current directory is a valid working tree.
105	1
106		Checking both the git working tree exists and that the top leve
107	1	directory path matches the current directory path.
108		"""
109
110	1	log.debug("Checking for a valid working tree...")
111
112	1	try:
113		git('rev-parse', '--is-inside-work-tree', _show=False)
114		except ShellError:
115	1	return False
116
117	1	log.debug("Checking for a valid git top level...")
118		gittoplevel = git('rev-parse', '--show-toplevel', _show=False)
119		currentdir = pwd(_show=False)
120
121	1	status = False
122		if gittoplevel[0] == currentdir:
123	1	status = True
124		else:
125		log.debug(
126	1	"git top level: %s != current working directory: %s",
127		gittoplevel[0],
128	1	currentdir,
129	1	)
130	1	status = False
131	1	return status
132	1
133	1
134		def rebuild(type, repo): # pylint: disable=unused-argument
135	1	"""Rebuild a missing repo .git directory."""
136
137		if type == 'git-svn':
138		# ignore rebuild in case of git-svn
139		return
140
141		assert type == 'git'
142
143		common.show("Rebuilding mising git repo...", color='message')
144		git('init', _show=True)
145		git('remote', 'add', 'origin', repo, _show=True)
146		common.show("Rebuilt git repo...", color='message')
147
148
149		def changes(type, include_untracked=False, display_status=True, _show=False):
150		"""Determine if there are changes in the working tree."""
151		status = False
152
153		if type == 'git-svn':
154		# ignore changes in case of git-svn
155		return status
156
157		assert type == 'git'
158
159		try:
160		# Refresh changes
161		git('update-index', '-q', '--refresh', _show=False)
162
163		# Check for uncommitted changes
164		git('diff-index', '--quiet', 'HEAD', _show=_show)
165
166		# Check for untracked files
167		lines = git('ls-files', '--others', '--exclude-standard', _show=_show)
168
169		except ShellError:
170		status = True
171
172		else:
173		status = bool(lines) and include_untracked
174
175		if status and display_status:
176		with suppress(ShellError):
177		lines = git('status', _show=True)
178		common.show(*lines, color='git_changes')
179
180		return status
181
182
183		def update(
184		type, repo, path, *, clean=True, fetch=False, rev=None
185		): # pylint: disable=redefined-outer-name,unused-argument
186
187		if type == 'git-svn':
188		# make deep clone here for simplification of sources.py
189		# and to realize consistent readonly clone (always forced)
190
191		# completly empty current directory (remove also hidden content)
192		for root, dirs, files in os.walk('.'):
193		for f in files:
194		os.unlink(os.path.join(root, f))
195		for d in dirs:
196		shutil.rmtree(os.path.join(root, d))
197
198		# clone specified svn revision
199		gitsvn('clone', '-r', rev, repo, '.')
200		return
201
202		assert type == 'git'
203
204		# Update the working tree to the specified revision.
205		hide = {'_show': False, '_ignore': True}
206
207		git('stash', **hide)
208		if clean:
209		git('clean', '--force', '-d', '-x', _show=False)
210
211		rev = _get_sha_from_rev(rev)
212		git('checkout', '--force', rev)
213		git('branch', '--set-upstream-to', 'origin/' + rev, **hide)
214
215		if fetch:
216		# if `rev` was a branch it might be tracking something older
217		git('pull', '--ff-only', '--no-rebase', **hide)
218
219
220		def get_url(type):
221		"""Get the current repository's URL."""
222		if type == 'git-svn':
223		return git('config', '--get', 'svn-remote.svn.url', _show=False)[0]
224
225		assert type == 'git'
226
227		return git('config', '--get', 'remote.origin.url', _show=False)[0]
228
229
230		def get_hash(type, _show=False):
231		"""Get the current working tree's hash."""
232		if type == 'git-svn':
233		return ''.join(filter(str.isdigit, gitsvn('info', _show=_show)[4]))
234
235		assert type == 'git'
236
237		return git('rev-parse', 'HEAD', _show=_show)[0]
238
239
240		def get_tag():
241		"""Get the current working tree's tag (if on a tag)."""
242		return git('describe', '--tags', '--exact-match', _show=False, _ignore=True)[0]
243
244
245		def is_fetch_required(type, rev):
246		if type == 'git-svn':
247		return False
248
249		assert type == 'git'
250
251		return rev not in (get_branch(), get_hash(type), get_tag())
252
253
254		def get_branch():
255		"""Get the current working tree's branch."""
256		return git('rev-parse', '--abbrev-ref', 'HEAD', _show=False)[0]
257
258
259		def _get_sha_from_rev(rev):
260		"""Get a rev-parse string's hash."""
261		if '@{' in rev: # TODO: use regex for this
262		parts = rev.split('@')
263		branch = parts[0]
264		date = parts[1].strip("{}")
265		git('checkout', '--force', branch, _show=False)
266		rev = git(
267		'rev-list',
268		'-n',
269		'1',
270		'--before={!r}'.format(date),
271		'--first-parent',
272		branch,
273		_show=False,
274		)[0]
275		return rev
276

jacebrowning / gitman

Push — develop ( de828f...5a60d4 )

gitman.git.rebuild() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like