Completed
Push — develop ( 08af06...aaf505 )
by Jace
10s
created

is_sha()   A

Complexity

Conditions 1

Size

Total Lines 8

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 5
CRAP Score 1

Importance

Changes 1
Bugs 0 Features 0
Metric Value
cc 1
dl 0
loc 8
ccs 5
cts 5
cp 1
crap 1
rs 9.4285
c 1
b 0
f 0
1
"""Utilities to call Git commands."""
2
3 1
import os
4 1
import logging
5 1
from contextlib import suppress
6
import re
7 1
8 1
from . import common, settings
9 1
from .shell import call
10
from .exceptions import ShellError
11
12 1
13
log = logging.getLogger(__name__)
14
15 1
16 1
def git(*args, **kwargs):
17
    return call('git', *args, **kwargs)
18
19 1
20
def clone(repo, path, *, cache=settings.CACHE, sparse_paths=None, rev=None):
21 1
    """Clone a new Git repository."""
22
    log.debug("Creating a new repository...")
23 1
24 1
    name = repo.split('/')[-1]
25 1
    if name.endswith(".git"):
26
        name = name[:-4]
27 1
28 1
    reference = os.path.join(cache, name + ".reference")
29 1
    if not os.path.isdir(reference):
30
        git('clone', '--mirror', repo, reference)
31 1
32
    normpath = os.path.normpath(path)
33
    if sparse_paths:
34 1
        os.mkdir(normpath)
35
        git('-C', normpath, 'init')
36 1
        git('-C', normpath, 'config', 'core.sparseCheckout', 'true')
37 1
        git('-C', normpath, 'remote', 'add', '-f', 'origin', reference)
38 1
39 1
        with open("%s/%s/.git/info/sparse-checkout" % (os.getcwd(), normpath), 'w') as fd:
40 1
            fd.writelines(sparse_paths)
41 1
        with open("%s/%s/.git/objects/info/alternates" % (os.getcwd(), normpath), 'w') as fd:
42 1
            fd.write("%s/objects" % reference)
43
44 1
        # We use directly the revision requested here in order to respect,
45 1
        # that not all repos have `master` as their default branch
46
        git('-C', normpath, 'pull', 'origin', rev)
47
    else:
48 1
        git('clone', '--reference', reference, repo, os.path.normpath(path))
49
50 1
51
def is_sha(rev):
52 1
    """Heuristically determine whether a revision corresponds to a commit SHA.
53 1
54 1
    Any sequence of 7 to 40 hexadecimal digits will be recognized as a
55 1
    commit SHA. The minimum of 7 digits is not an arbitrary choice, it
56
    is the default length for short SHAs in Git.
57 1
    """
58
    return re.match('^[0-9a-f]{7,40}$', rev) is not None
59
60 1
61
def fetch(repo, rev=None):
62 1
    """Fetch the latest changes from the remote repository."""
63
    git('remote', 'set-url', 'origin', repo)
64 1
    args = ['fetch', '--tags', '--force', '--prune', 'origin']
65
    if rev:
66 1
        if is_sha(rev):
67
            pass  # fetch only works with a SHA if already present locally
68
        elif '@' in rev:
69 1
            pass  # fetch doesn't work with rev-parse
70
        else:
71
            args.append(rev)
72 1
    git(*args)
73
74 1
75 1
def valid():
76
    """Confirm the current directory is a valid working tree."""
77
    log.debug("Checking for a valid working tree...")
78 1
79
    try:
80 1
        git('rev-parse', '--is-inside-work-tree', _show=False)
81 1
    except ShellError:
82 1
        return False
83 1
    else:
84
        return True
85 1
86
87
def changes(include_untracked=False, display_status=True, _show=False):
88 1
    """Determine if there are changes in the working tree."""
89
    status = False
90 1
91
    try:
92 1
        # Refresh changes
93 1
        git('update-index', '-q', '--refresh', _show=False)
94 1
95
        # Check for uncommitted changes
96 1
        git('diff-index', '--quiet', 'HEAD', _show=_show)
97 1
98 1
        # Check for untracked files
99
        lines = git('ls-files', '--others', '--exclude-standard', _show=_show)
100 1
101
    except ShellError:
102 1
        status = True
103
104
    else:
105 1
        status = bool(lines) and include_untracked
106
107 1
    if status and display_status:
108
        with suppress(ShellError):
109
            lines = git('status', _show=True)
110 1
            common.show(*lines, color='git_changes')
111
112 1
    return status
113
114
115 1
def update(rev, *, clean=True, fetch=False):  # pylint: disable=redefined-outer-name
116
    """Update the working tree to the specified revision."""
117 1
    hide = {'_show': False, '_ignore': True}
118
119
    git('stash', **hide)
120
    if clean:
121 1
        git('clean', '--force', '-d', '-x', _show=False)
122
123 1
    rev = _get_sha_from_rev(rev)
124
    git('checkout', '--force', rev)
125
    git('branch', '--set-upstream-to', 'origin/' + rev, **hide)
126 1
127
    if fetch:
128 1
        # if `rev` was a branch it might be tracking something older
129 1
        git('pull', '--ff-only', '--no-rebase', **hide)
130 1
131 1
132 1
def get_url():
133 1
    """Get the current repository's URL."""
134
    return git('config', '--get', 'remote.origin.url', _show=False)[0]
135 1
136
137
def get_hash(_show=False):
138
    """Get the current working tree's hash."""
139
    return git('rev-parse', 'HEAD', _show=_show)[0]
140
141
142
def get_tag():
143
    """Get the current working tree's tag (if on a tag)."""
144
    return git('describe', '--tags', '--exact-match',
145
               _show=False, _ignore=True)[0]
146
147
148
def get_branch():
149
    """Get the current working tree's branch."""
150
    return git('rev-parse', '--abbrev-ref', 'HEAD', _show=False)[0]
151
152
153
def _get_sha_from_rev(rev):
154
    """Get a rev-parse string's hash."""
155
    if '@{' in rev:  # TODO: use regex for this
156
        parts = rev.split('@')
157
        branch = parts[0]
158
        date = parts[1].strip("{}")
159
        git('checkout', '--force', branch, _show=False)
160
        rev = git('rev-list', '-n', '1', '--before={!r}'.format(date),
161
                  branch, _show=False)[0]
162
    return rev
163