clone() - Code Metrics - Inspection of "Feature git svn support #151" - jacebrowning/gitman - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#164)

unknown

created 2018-03-19 09:19 UTC

clone() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	15
CRAP Score	8

Importance

Changes	4
Bugs	0	Features	0

Metric	Value
cc	8
c	4
b	0
f	0
dl	0
loc	33
ccs	15
cts	15
cp	1
crap	8
rs	4

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

import os
import logging
from contextlib import suppress
from enum import Enum
from . import common, settings
from .shell import call
from .exceptions import ShellError


log = logging.getLogger(__name__)


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

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

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


"""Represnents a repository type"""

class RepoType(Enum):
    UNKNOWN = -1
    GIT = 1
    SVN = 2
    GIT_SVN = 3
    

def _get_local_repotype():
    """Resolves the type of the local repository"""

    repotype = RepoType.UNKNOWN

    # In case of a git repo
    # git show-ref
    # 309189a3dfd8a0132f300c4bf0373efb1a654add refs/heads/master
    # 309189a3dfd8a0132f300c4bf0373efb1a654add refs/remotes/origin/HEAD
    # 309189a3dfd8a0132f300c4bf0373efb1a654add refs/remotes/origin/master
    # f5c86bb39ff8f32feee7d3189546bf29af777fc7 refs/remotes/origin/release/1.0.x
    # fddfe0e02c497ed2847748767a2237c61d847723 refs/tags/release/1.0.0
    #
    # In case of GIT-SVN-REPO
    #git show-ref
    #cd351f6007d1e4d446fe5956dd3e576464e7617c refs/heads/master
    #cd351f6007d1e4d446fe5956dd3e576464e7617c refs/remotes/git-svn

    output = git('show-ref', _show=False, _ignore=True)
    if any('refs/remotes/git-svn' in line for line in output): # check if out contains refs/remotes/git-svn -> GIT_SVN

        repotype = RepoType.GIT_SVN
    elif any('refs/remotes/origin' in line for line in output): # check if output contains refs/remotes/origin -> GIT

        repotype = RepoType.GIT
        

    return repotype

def _get_remote_repotype(repo): 

    """Resolves the type of the remote repository"""

    repotype = RepoType.UNKNOWN

    # git ls-remote [email protected]:Brosche/waf-prototyping-lib.git
    # 519503d686dcf690df3eea051d82a3f99af80805        HEAD
    # 519503d686dcf690df3eea051d82a3f99af80805        refs/heads/master
    # 94ac960c84ceb2542df2b6b7adfe2271ec990d01        refs/tags/release/1.0.0

    #git ls-remote --heads http://svn-ldc/svn/Controls/trunk/XP/TAs/Implementation/i900
    #fatal: unable to access 'http://svn-ldc/svn/Controls/trunk/XP/TAs/Implementation/i900/': The requested URL returned error: 502
    #git ls-remote http://google.de
    #fatal: repository 'http://google.de/' not found

    output = git('ls-remote', repo, _show=False, _ignore=True)
    if any('refs/heads' in line for line in output): # check if output contains refs/heads -> GIT
        repotype = RepoType.GIT    

    elif any('fatal' in line for line in output): # check if output contains fatal -> NOT GIT
        # svn info http://svn-ldc/svn/Controls/trunk/XP/TAs/Implementation/i900
        # Path: i900
        # URL: http://svn-ldc/svn/Controls/trunk/XP/TAs/Implementation/i900
        # Relative URL: ^/trunk/XP/TAs/Implementation/i900
        # Repository Root: http://svn-ldc/svn/Controls
        # Repository UUID: 6cb63ced-a357-0410-a762-939f09bf00c8
        # Revision: 72500
        # Node Kind: directory
        # Last Changed Author: [email protected]
        # Last Changed Rev: 72435
        # Last Changed Date: 2018-03-06 14:36:40 +0100 (Tue, 06 Mar 2018)
        output = svn('info', repo, _show=False, _ignore=True)
        if any('UUID' in line for line in output): # check if output contains UUID -> SVN
            repotype = RepoType.SVN

    return repotype





def clone(repo, path, *, cache=settings.CACHE, sparse_paths=None, rev=None):
    """Clone a new Git repository."""
    repotype = _get_remote_repotype(repo)
    log.debug("Clone a new {} repository...".format(str(repotype)))

    

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

        reference = os.path.join(cache, name + ".reference")
        if not os.path.isdir(reference):
            git('clone', '--mirror', repo, reference)

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

            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" % reference)

            # 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)
        else:
            git('clone', '--reference', reference, repo, os.path.normpath(path))
    elif repotype == RepoType.SVN:
        gitsvn('clone', '-r', 'HEAD', repo, path)

def fetch(repo, rev=None):
    """Fetch the latest changes from the remote repository."""

    repotype = _get_local_repotype()
    log.debug("Fetch the latest change from the remote {} repository...".format(str(repotype)))


    if repotype == RepoType.GIT:
        git('remote', 'set-url', 'origin', repo)
        args = ['fetch', '--tags', '--force', '--prune', 'origin']
        if rev:
            if len(rev) == 40:
                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)
    elif repotype == RepoType.GIT_SVN and rev:
        gitsvn('rebase', rev)

def valid():
    """Confirm the current directory is a valid working tree."""
    log.debug("Checking for a valid working tree...")

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


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

    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(rev, *, clean=True, fetch=False):  # pylint: disable=redefined-outer-name
    """Update the working tree to the specified revision."""
    hide = {'_show': False, '_ignore': True}

    repotype = _get_local_repotype()
    log.debug("Update the {} repository...".format(str(repotype)))


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

    if repotype == RepoType.GIT:
        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)
    elif repotype == RepoType.GIT_SVN:
        gitsvn('rebase', rev)

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


def get_hash(_show=False):
    """Get the current working tree's hash."""
    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 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),
                  branch, _show=False)[0]
    return rev


1		"""Utilities to call Git commands."""
2
3	1	import os
4	1	import logging
5	1	from contextlib import suppress
6		from enum import Enum
7	1	from . import common, settings
8	1	from .shell import call
9	1	from .exceptions import ShellError
10
11
12	1	log = logging.getLogger(__name__)
13
14
15	1	def git(args, *kwargs):
16	1	return call('git', args, *kwargs)
17
18		def gitsvn(args, *kwargs):
19	1	return call('git', 'svn', args, *kwargs)
20
21	1	def svn(args, *kwargs):
22		return call('svn', args, *kwargs)
23	1
24	1
25	1	"""Represnents a repository type"""
		0 ignored issues – show Unused Code introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report This string statement has no effect and could be removed. Loading history...
26		class RepoType(Enum):
27	1	UNKNOWN = -1
28	1	GIT = 1
29	1	SVN = 2
30		GIT_SVN = 3
31	1
		0 ignored issues – show Coding Style introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
32		def _get_local_repotype():
33		"""Resolves the type of the local repository"""
34	1
35		repotype = RepoType.UNKNOWN
36	1
37	1	# In case of a git repo
38	1	# git show-ref
39	1	# 309189a3dfd8a0132f300c4bf0373efb1a654add refs/heads/master
40	1	# 309189a3dfd8a0132f300c4bf0373efb1a654add refs/remotes/origin/HEAD
41	1	# 309189a3dfd8a0132f300c4bf0373efb1a654add refs/remotes/origin/master
42	1	# f5c86bb39ff8f32feee7d3189546bf29af777fc7 refs/remotes/origin/release/1.0.x
43		# fddfe0e02c497ed2847748767a2237c61d847723 refs/tags/release/1.0.0
44	1	#
45	1	# In case of GIT-SVN-REPO
46		#git show-ref
47		#cd351f6007d1e4d446fe5956dd3e576464e7617c refs/heads/master
48	1	#cd351f6007d1e4d446fe5956dd3e576464e7617c refs/remotes/git-svn
49
50	1	output = git('show-ref', _show=False, _ignore=True)
51		if any('refs/remotes/git-svn' in line for line in output): # check if out contains refs/remotes/git-svn -> GIT_SVN
		0 ignored issues – show Coding Style introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (118/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
52	1	repotype = RepoType.GIT_SVN
53	1	elif any('refs/remotes/origin' in line for line in output): # check if output contains refs/remotes/origin -> GIT
		0 ignored issues – show Coding Style introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (117/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
54	1	repotype = RepoType.GIT
55	1
		0 ignored issues – show Coding Style introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
56		return repotype
57	1
58		def _get_remote_repotype(repo):
		0 ignored issues – show Coding Style introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
59		"""Resolves the type of the remote repository"""
60	1
61		repotype = RepoType.UNKNOWN
62	1
63		# git ls-remote [email protected]:Brosche/waf-prototyping-lib.git
64	1	# 519503d686dcf690df3eea051d82a3f99af80805 HEAD
65		# 519503d686dcf690df3eea051d82a3f99af80805 refs/heads/master
66	1	# 94ac960c84ceb2542df2b6b7adfe2271ec990d01 refs/tags/release/1.0.0
67
68		#git ls-remote --heads http://svn-ldc/svn/Controls/trunk/XP/TAs/Implementation/i900
69	1	#fatal: unable to access 'http://svn-ldc/svn/Controls/trunk/XP/TAs/Implementation/i900/': The requested URL returned error: 502
70		#git ls-remote http://google.de
71		#fatal: repository 'http://google.de/' not found
72	1
73		output = git('ls-remote', repo, _show=False, _ignore=True)
74	1	if any('refs/heads' in line for line in output): # check if output contains refs/heads -> GIT
75	1	repotype = RepoType.GIT
		0 ignored issues – show Coding Style introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
76		elif any('fatal' in line for line in output): # check if output contains fatal -> NOT GIT
77		# svn info http://svn-ldc/svn/Controls/trunk/XP/TAs/Implementation/i900
78	1	# Path: i900
79		# URL: http://svn-ldc/svn/Controls/trunk/XP/TAs/Implementation/i900
80	1	# Relative URL: ^/trunk/XP/TAs/Implementation/i900
81	1	# Repository Root: http://svn-ldc/svn/Controls
82	1	# Repository UUID: 6cb63ced-a357-0410-a762-939f09bf00c8
83	1	# Revision: 72500
84		# Node Kind: directory
85	1	# Last Changed Author: [email protected]
86		# Last Changed Rev: 72435
87		# Last Changed Date: 2018-03-06 14:36:40 +0100 (Tue, 06 Mar 2018)
88	1	output = svn('info', repo, _show=False, _ignore=True)
89		if any('UUID' in line for line in output): # check if output contains UUID -> SVN
90	1	repotype = RepoType.SVN
91
92	1	return repotype
93	1
94	1
95
96	1
97	1
98	1	def clone(repo, path, *, cache=settings.CACHE, sparse_paths=None, rev=None):
99		"""Clone a new Git repository."""
100	1	repotype = _get_remote_repotype(repo)
101		log.debug("Clone a new {} repository...".format(str(repotype)))
		0 ignored issues – show introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report Use formatting in logging functions and pass the parameters as arguments Loading history...
102	1
		0 ignored issues – show Coding Style introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
103		if repotype == RepoType.GIT:
104		name = repo.split('/')[-1]
105	1	if name.endswith(".git"):
106		name = name[:-4]
107	1
108		reference = os.path.join(cache, name + ".reference")
109		if not os.path.isdir(reference):
110	1	git('clone', '--mirror', repo, reference)
111
112	1	normpath = os.path.normpath(path)
113		if sparse_paths:
114		os.mkdir(normpath)
115	1	git('-C', normpath, 'init')
116		git('-C', normpath, 'config', 'core.sparseCheckout', 'true')
117	1	git('-C', normpath, 'remote', 'add', '-f', 'origin', reference)
118
119		with open("%s/%s/.git/info/sparse-checkout" % (os.getcwd(), normpath), 'w') as fd:
120		fd.writelines(sparse_paths)
121	1	with open("%s/%s/.git/objects/info/alternates" % (os.getcwd(), normpath), 'w') as fd:
122		fd.write("%s/objects" % reference)
123	1
124		# We use directly the revision requested here in order to respect,
125		# that not all repos have `master` as their default branch
126	1	git('-C', normpath, 'pull', 'origin', rev)
127		else:
128	1	git('clone', '--reference', reference, repo, os.path.normpath(path))
129	1	elif repotype == RepoType.SVN:
130	1	gitsvn('clone', '-r', 'HEAD', repo, path)
131	1
132	1	def fetch(repo, rev=None):
133	1	"""Fetch the latest changes from the remote repository."""
134
135	1	repotype = _get_local_repotype()
136		log.debug("Fetch the latest change from the remote {} repository...".format(str(repotype)))
		0 ignored issues – show introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report Use formatting in logging functions and pass the parameters as arguments Loading history...
137
138		if repotype == RepoType.GIT:
139		git('remote', 'set-url', 'origin', repo)
140		args = ['fetch', '--tags', '--force', '--prune', 'origin']
141		if rev:
142		if len(rev) == 40:
143		pass # fetch only works with a SHA if already present locally
144		elif '@' in rev:
145		pass # fetch doesn't work with rev-parse
146		else:
147		args.append(rev)
148		git(*args)
149		elif repotype == RepoType.GIT_SVN and rev:
150		gitsvn('rebase', rev)
151
152		def valid():
153		"""Confirm the current directory is a valid working tree."""
154		log.debug("Checking for a valid working tree...")
155
156		try:
157		git('rev-parse', '--is-inside-work-tree', _show=False)
158		except ShellError:
159		return False
160		else:
161		return True
162
163
164		def changes(include_untracked=False, display_status=True, _show=False):
165		"""Determine if there are changes in the working tree."""
166		status = False
167
168		try:
169		# Refresh changes
170		git('update-index', '-q', '--refresh', _show=False)
171
172		# Check for uncommitted changes
173		git('diff-index', '--quiet', 'HEAD', _show=_show)
174
175		# Check for untracked files
176		lines = git('ls-files', '--others', '--exclude-standard', _show=_show)
177
178		except ShellError:
179		status = True
180
181		else:
182		status = bool(lines) and include_untracked
183
184		if status and display_status:
185		with suppress(ShellError):
186		lines = git('status', _show=True)
187		common.show(*lines, color='git_changes')
188
189		return status
190
191
192		def update(rev, *, clean=True, fetch=False): # pylint: disable=redefined-outer-name
193		"""Update the working tree to the specified revision."""
194		hide = {'_show': False, '_ignore': True}
195
196		repotype = _get_local_repotype()
197		log.debug("Update the {} repository...".format(str(repotype)))
		0 ignored issues – show introduced 2018-03-19 09:21 UTC by Report Bug Copy Issue Report Use formatting in logging functions and pass the parameters as arguments Loading history...
198
199		git('stash', **hide)
200		if clean:
201		git('clean', '--force', '-d', '-x', _show=False)
202
203		if repotype == RepoType.GIT:
204		rev = _get_sha_from_rev(rev)
205		git('checkout', '--force', rev)
206		git('branch', '--set-upstream-to', 'origin/' + rev, **hide)
207
208		if fetch:
209		# if `rev` was a branch it might be tracking something older
210		git('pull', '--ff-only', '--no-rebase', **hide)
211		elif repotype == RepoType.GIT_SVN:
212		gitsvn('rebase', rev)
213
214		def get_url():
215		"""Get the current repository's URL."""
216		return git('config', '--get', 'remote.origin.url', _show=False)[0]
217
218
219		def get_hash(_show=False):
220		"""Get the current working tree's hash."""
221		return git('rev-parse', 'HEAD', _show=_show)[0]
222
223
224		def get_tag():
225		"""Get the current working tree's tag (if on a tag)."""
226		return git('describe', '--tags', '--exact-match',
227		_show=False, _ignore=True)[0]
228
229
230		def get_branch():
231		"""Get the current working tree's branch."""
232		return git('rev-parse', '--abbrev-ref', 'HEAD', _show=False)[0]
233
234
235		def _get_sha_from_rev(rev):
236		"""Get a rev-parse string's hash."""
237		if '@{' in rev: # TODO: use regex for this
238		parts = rev.split('@')
239		branch = parts[0]
240		date = parts[1].strip("{}")
241		git('checkout', '--force', branch, _show=False)
242		rev = git('rev-list', '-n', '1', '--before={!r}'.format(date),
243		branch, _show=False)[0]
244		return rev
245

jacebrowning / gitman

Pull Request — develop (#164)

clone() D

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like