Completed
Push — develop ( f5b93c...24eb3d )
by Jace
11s
created

gitman.git.is_fetch_required()   A

Complexity

Conditions 2

Size

Total Lines 9
Code Lines 7

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 6

Importance

Changes 0
Metric Value
cc 2
eloc 7
nop 2
dl 0
loc 9
ccs 0
cts 0
cp 0
crap 6
rs 10
c 0
b 0
f 0
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
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
    reference = os.path.join(cache, name + ".reference")
42 1
    if not os.path.isdir(reference):
43
        git('clone', '--mirror', repo, reference)
44 1
45 1
    normpath = os.path.normpath(path)
46
    if sparse_paths:
47
        os.mkdir(normpath)
48 1
        git('-C', normpath, 'init')
49
        git('-C', normpath, 'config', 'core.sparseCheckout', 'true')
50 1
        git('-C', normpath, 'remote', 'add', '-f', 'origin', reference)
51
52 1
        with open("%s/%s/.git/info/sparse-checkout" %
53 1
                  (os.getcwd(), normpath), 'w') as fd:
54 1
            fd.writelines(sparse_paths)
55 1
        with open("%s/%s/.git/objects/info/alternates" %
56
                  (os.getcwd(), normpath), 'w') as fd:
57 1
            fd.write("%s/objects" % reference)
58
59
        # We use directly the revision requested here in order to respect,
60 1
        # that not all repos have `master` as their default branch
61
        git('-C', normpath, 'pull', 'origin', rev)
62 1
    else:
63
        git('clone', '--reference', reference, repo, os.path.normpath(path))
64 1
65
66 1
def is_sha(rev):
67
    """Heuristically determine whether a revision corresponds to a commit SHA.
68
69 1
    Any sequence of 7 to 40 hexadecimal digits will be recognized as a
70
    commit SHA. The minimum of 7 digits is not an arbitrary choice, it
71
    is the default length for short SHAs in Git.
72 1
    """
73
    return re.match('^[0-9a-f]{7,40}$', rev) is not None
74 1
75 1
76
def fetch(type, repo, path, rev=None):  # pylint: disable=unused-argument
77
    """Fetch the latest changes from the remote repository."""
78 1
79
    if type == 'git-svn':
80 1
        # deep clone happens in update function
81 1
        return
82 1
83 1
    assert type == 'git'
84
85 1
    git('remote', 'set-url', 'origin', repo)
86
    args = ['fetch', '--tags', '--force', '--prune', 'origin']
87
    if rev:
88 1
        if is_sha(rev):
89
            pass  # fetch only works with a SHA if already present locally
90 1
        elif '@' in rev:
91
            pass  # fetch doesn't work with rev-parse
92 1
        else:
93 1
            args.append(rev)
94 1
    git(*args)
95
96 1
97 1
def valid():
98 1
    """Confirm the current directory is a valid working tree."""
99
    log.debug("Checking for a valid working tree...")
100 1
101
    try:
102 1
        git('rev-parse', '--is-inside-work-tree', _show=False)
103
    except ShellError:
104
        return False
105 1
    else:
106
        return True
107 1
108
109
def changes(type, include_untracked=False, display_status=True, _show=False):
110 1
    """Determine if there are changes in the working tree."""
111
    status = False
112 1
113
    if type == 'git-svn':
114
        # ignore changes in case of git-svn
115 1
        return status
116
117 1
    assert type == 'git'
118
119
    try:
120
        # Refresh changes
121 1
        git('update-index', '-q', '--refresh', _show=False)
122
123 1
        # Check for uncommitted changes
124
        git('diff-index', '--quiet', 'HEAD', _show=_show)
125
126 1
        # Check for untracked files
127
        lines = git('ls-files', '--others', '--exclude-standard', _show=_show)
128 1
129 1
    except ShellError:
130 1
        status = True
131 1
132 1
    else:
133 1
        status = bool(lines) and include_untracked
134
135 1
    if status and display_status:
136
        with suppress(ShellError):
137
            lines = git('status', _show=True)
138
            common.show(*lines, color='git_changes')
139
140
    return status
141
142
143
def update(type, repo, path, *, clean=True, fetch=False, rev=None):  # pylint: disable=redefined-outer-name,unused-argument
144
145
    if type == 'git-svn':
146
        # make deep clone here for simplification of sources.py
147
        # and to realize consistent readonly clone (always forced)
148
149
        # completly empty current directory (remove also hidden content)
150
        for root, dirs, files in os.walk('.'):
151
            for f in files:
152
                os.unlink(os.path.join(root, f))
153
            for d in dirs:
154
                shutil.rmtree(os.path.join(root, d))
155
156
        # clone specified svn revision
157
        gitsvn('clone', '-r', rev, repo, '.')
158
        return
159
160
    assert type == 'git'
161
162
    # Update the working tree to the specified revision.
163
    hide = {'_show': False, '_ignore': True}
164
165
    git('stash', **hide)
166
    if clean:
167
        git('clean', '--force', '-d', '-x', _show=False)
168
169
    rev = _get_sha_from_rev(rev)
170
    git('checkout', '--force', rev)
171
    git('branch', '--set-upstream-to', 'origin/' + rev, **hide)
172
173
    if fetch:
174
        # if `rev` was a branch it might be tracking something older
175
        git('pull', '--ff-only', '--no-rebase', **hide)
176
177
178
def get_url(type):
179
    """Get the current repository's URL."""
180
    if type == 'git-svn':
181
        return git('config', '--get', 'svn-remote.svn.url', _show=False)[0]
182
183
    assert type == 'git'
184
185
    return git('config', '--get', 'remote.origin.url', _show=False)[0]
186
187
188
def get_hash(type, _show=False):
189
    """Get the current working tree's hash."""
190
    if type == 'git-svn':
191
        return ''.join(filter(str.isdigit, gitsvn('info', _show=_show)[4]))
192
193
    assert type == 'git'
194
195
    return git('rev-parse', 'HEAD', _show=_show)[0]
196
197
198
def get_tag():
199
    """Get the current working tree's tag (if on a tag)."""
200
    return git('describe', '--tags', '--exact-match',
201
               _show=False, _ignore=True)[0]
202
203
204
def is_fetch_required(type, rev):
205
    if type == 'git-svn':
206
        return False
207
208
    assert type == 'git'
209
210
    return rev not in (get_branch(),
211
                       get_hash(type),
212
                       get_tag())
213
214
215
def get_branch():
216
    """Get the current working tree's branch."""
217
    return git('rev-parse', '--abbrev-ref', 'HEAD', _show=False)[0]
218
219
220
def _get_sha_from_rev(rev):
221
    """Get a rev-parse string's hash."""
222
    if '@{' in rev:  # TODO: use regex for this
223
        parts = rev.split('@')
224
        branch = parts[0]
225
        date = parts[1].strip("{}")
226
        git('checkout', '--force', branch, _show=False)
227
        rev = git('rev-list', '-n', '1', '--before={!r}'.format(date),
228
                  branch, _show=False)[0]
229
    return rev
230