comparison lib/python3.8/site-packages/pip/_internal/vcs/git.py @ 0:9e54283cc701 draft

"planemo upload commit d12c32a45bcd441307e632fca6d9af7d60289d44"
author guerler
date Mon, 27 Jul 2020 03:47:31 -0400 (2020-07-27)
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:9e54283cc701
1 # The following comment should be removed at some point in the future.
2 # mypy: disallow-untyped-defs=False
3
4 from __future__ import absolute_import
5
6 import logging
7 import os.path
8 import re
9
10 from pip._vendor.packaging.version import parse as parse_version
11 from pip._vendor.six.moves.urllib import parse as urllib_parse
12 from pip._vendor.six.moves.urllib import request as urllib_request
13
14 from pip._internal.exceptions import BadCommand
15 from pip._internal.utils.misc import display_path, hide_url
16 from pip._internal.utils.subprocess import make_command
17 from pip._internal.utils.temp_dir import TempDirectory
18 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
19 from pip._internal.vcs.versioncontrol import (
20 RemoteNotFoundError,
21 VersionControl,
22 find_path_to_setup_from_repo_root,
23 vcs,
24 )
25
26 if MYPY_CHECK_RUNNING:
27 from typing import Optional, Tuple
28 from pip._internal.utils.misc import HiddenText
29 from pip._internal.vcs.versioncontrol import AuthInfo, RevOptions
30
31
32 urlsplit = urllib_parse.urlsplit
33 urlunsplit = urllib_parse.urlunsplit
34
35
36 logger = logging.getLogger(__name__)
37
38
39 HASH_REGEX = re.compile('^[a-fA-F0-9]{40}$')
40
41
42 def looks_like_hash(sha):
43 return bool(HASH_REGEX.match(sha))
44
45
46 class Git(VersionControl):
47 name = 'git'
48 dirname = '.git'
49 repo_name = 'clone'
50 schemes = (
51 'git', 'git+http', 'git+https', 'git+ssh', 'git+git', 'git+file',
52 )
53 # Prevent the user's environment variables from interfering with pip:
54 # https://github.com/pypa/pip/issues/1130
55 unset_environ = ('GIT_DIR', 'GIT_WORK_TREE')
56 default_arg_rev = 'HEAD'
57
58 @staticmethod
59 def get_base_rev_args(rev):
60 return [rev]
61
62 def is_immutable_rev_checkout(self, url, dest):
63 # type: (str, str) -> bool
64 _, rev_options = self.get_url_rev_options(hide_url(url))
65 if not rev_options.rev:
66 return False
67 if not self.is_commit_id_equal(dest, rev_options.rev):
68 # the current commit is different from rev,
69 # which means rev was something else than a commit hash
70 return False
71 # return False in the rare case rev is both a commit hash
72 # and a tag or a branch; we don't want to cache in that case
73 # because that branch/tag could point to something else in the future
74 is_tag_or_branch = bool(
75 self.get_revision_sha(dest, rev_options.rev)[0]
76 )
77 return not is_tag_or_branch
78
79 def get_git_version(self):
80 VERSION_PFX = 'git version '
81 version = self.run_command(['version'], show_stdout=False)
82 if version.startswith(VERSION_PFX):
83 version = version[len(VERSION_PFX):].split()[0]
84 else:
85 version = ''
86 # get first 3 positions of the git version because
87 # on windows it is x.y.z.windows.t, and this parses as
88 # LegacyVersion which always smaller than a Version.
89 version = '.'.join(version.split('.')[:3])
90 return parse_version(version)
91
92 @classmethod
93 def get_current_branch(cls, location):
94 """
95 Return the current branch, or None if HEAD isn't at a branch
96 (e.g. detached HEAD).
97 """
98 # git-symbolic-ref exits with empty stdout if "HEAD" is a detached
99 # HEAD rather than a symbolic ref. In addition, the -q causes the
100 # command to exit with status code 1 instead of 128 in this case
101 # and to suppress the message to stderr.
102 args = ['symbolic-ref', '-q', 'HEAD']
103 output = cls.run_command(
104 args, extra_ok_returncodes=(1, ), show_stdout=False, cwd=location,
105 )
106 ref = output.strip()
107
108 if ref.startswith('refs/heads/'):
109 return ref[len('refs/heads/'):]
110
111 return None
112
113 def export(self, location, url):
114 # type: (str, HiddenText) -> None
115 """Export the Git repository at the url to the destination location"""
116 if not location.endswith('/'):
117 location = location + '/'
118
119 with TempDirectory(kind="export") as temp_dir:
120 self.unpack(temp_dir.path, url=url)
121 self.run_command(
122 ['checkout-index', '-a', '-f', '--prefix', location],
123 show_stdout=False, cwd=temp_dir.path
124 )
125
126 @classmethod
127 def get_revision_sha(cls, dest, rev):
128 """
129 Return (sha_or_none, is_branch), where sha_or_none is a commit hash
130 if the revision names a remote branch or tag, otherwise None.
131
132 Args:
133 dest: the repository directory.
134 rev: the revision name.
135 """
136 # Pass rev to pre-filter the list.
137 output = cls.run_command(['show-ref', rev], cwd=dest,
138 show_stdout=False, on_returncode='ignore')
139 refs = {}
140 for line in output.strip().splitlines():
141 try:
142 sha, ref = line.split()
143 except ValueError:
144 # Include the offending line to simplify troubleshooting if
145 # this error ever occurs.
146 raise ValueError('unexpected show-ref line: {!r}'.format(line))
147
148 refs[ref] = sha
149
150 branch_ref = 'refs/remotes/origin/{}'.format(rev)
151 tag_ref = 'refs/tags/{}'.format(rev)
152
153 sha = refs.get(branch_ref)
154 if sha is not None:
155 return (sha, True)
156
157 sha = refs.get(tag_ref)
158
159 return (sha, False)
160
161 @classmethod
162 def resolve_revision(cls, dest, url, rev_options):
163 # type: (str, HiddenText, RevOptions) -> RevOptions
164 """
165 Resolve a revision to a new RevOptions object with the SHA1 of the
166 branch, tag, or ref if found.
167
168 Args:
169 rev_options: a RevOptions object.
170 """
171 rev = rev_options.arg_rev
172 # The arg_rev property's implementation for Git ensures that the
173 # rev return value is always non-None.
174 assert rev is not None
175
176 sha, is_branch = cls.get_revision_sha(dest, rev)
177
178 if sha is not None:
179 rev_options = rev_options.make_new(sha)
180 rev_options.branch_name = rev if is_branch else None
181
182 return rev_options
183
184 # Do not show a warning for the common case of something that has
185 # the form of a Git commit hash.
186 if not looks_like_hash(rev):
187 logger.warning(
188 "Did not find branch or tag '%s', assuming revision or ref.",
189 rev,
190 )
191
192 if not rev.startswith('refs/'):
193 return rev_options
194
195 # If it looks like a ref, we have to fetch it explicitly.
196 cls.run_command(
197 make_command('fetch', '-q', url, rev_options.to_args()),
198 cwd=dest,
199 )
200 # Change the revision to the SHA of the ref we fetched
201 sha = cls.get_revision(dest, rev='FETCH_HEAD')
202 rev_options = rev_options.make_new(sha)
203
204 return rev_options
205
206 @classmethod
207 def is_commit_id_equal(cls, dest, name):
208 """
209 Return whether the current commit hash equals the given name.
210
211 Args:
212 dest: the repository directory.
213 name: a string name.
214 """
215 if not name:
216 # Then avoid an unnecessary subprocess call.
217 return False
218
219 return cls.get_revision(dest) == name
220
221 def fetch_new(self, dest, url, rev_options):
222 # type: (str, HiddenText, RevOptions) -> None
223 rev_display = rev_options.to_display()
224 logger.info('Cloning %s%s to %s', url, rev_display, display_path(dest))
225 self.run_command(make_command('clone', '-q', url, dest))
226
227 if rev_options.rev:
228 # Then a specific revision was requested.
229 rev_options = self.resolve_revision(dest, url, rev_options)
230 branch_name = getattr(rev_options, 'branch_name', None)
231 if branch_name is None:
232 # Only do a checkout if the current commit id doesn't match
233 # the requested revision.
234 if not self.is_commit_id_equal(dest, rev_options.rev):
235 cmd_args = make_command(
236 'checkout', '-q', rev_options.to_args(),
237 )
238 self.run_command(cmd_args, cwd=dest)
239 elif self.get_current_branch(dest) != branch_name:
240 # Then a specific branch was requested, and that branch
241 # is not yet checked out.
242 track_branch = 'origin/{}'.format(branch_name)
243 cmd_args = [
244 'checkout', '-b', branch_name, '--track', track_branch,
245 ]
246 self.run_command(cmd_args, cwd=dest)
247
248 #: repo may contain submodules
249 self.update_submodules(dest)
250
251 def switch(self, dest, url, rev_options):
252 # type: (str, HiddenText, RevOptions) -> None
253 self.run_command(
254 make_command('config', 'remote.origin.url', url),
255 cwd=dest,
256 )
257 cmd_args = make_command('checkout', '-q', rev_options.to_args())
258 self.run_command(cmd_args, cwd=dest)
259
260 self.update_submodules(dest)
261
262 def update(self, dest, url, rev_options):
263 # type: (str, HiddenText, RevOptions) -> None
264 # First fetch changes from the default remote
265 if self.get_git_version() >= parse_version('1.9.0'):
266 # fetch tags in addition to everything else
267 self.run_command(['fetch', '-q', '--tags'], cwd=dest)
268 else:
269 self.run_command(['fetch', '-q'], cwd=dest)
270 # Then reset to wanted revision (maybe even origin/master)
271 rev_options = self.resolve_revision(dest, url, rev_options)
272 cmd_args = make_command('reset', '--hard', '-q', rev_options.to_args())
273 self.run_command(cmd_args, cwd=dest)
274 #: update submodules
275 self.update_submodules(dest)
276
277 @classmethod
278 def get_remote_url(cls, location):
279 """
280 Return URL of the first remote encountered.
281
282 Raises RemoteNotFoundError if the repository does not have a remote
283 url configured.
284 """
285 # We need to pass 1 for extra_ok_returncodes since the command
286 # exits with return code 1 if there are no matching lines.
287 stdout = cls.run_command(
288 ['config', '--get-regexp', r'remote\..*\.url'],
289 extra_ok_returncodes=(1, ), show_stdout=False, cwd=location,
290 )
291 remotes = stdout.splitlines()
292 try:
293 found_remote = remotes[0]
294 except IndexError:
295 raise RemoteNotFoundError
296
297 for remote in remotes:
298 if remote.startswith('remote.origin.url '):
299 found_remote = remote
300 break
301 url = found_remote.split(' ')[1]
302 return url.strip()
303
304 @classmethod
305 def get_revision(cls, location, rev=None):
306 if rev is None:
307 rev = 'HEAD'
308 current_rev = cls.run_command(
309 ['rev-parse', rev], show_stdout=False, cwd=location,
310 )
311 return current_rev.strip()
312
313 @classmethod
314 def get_subdirectory(cls, location):
315 """
316 Return the path to setup.py, relative to the repo root.
317 Return None if setup.py is in the repo root.
318 """
319 # find the repo root
320 git_dir = cls.run_command(
321 ['rev-parse', '--git-dir'],
322 show_stdout=False, cwd=location).strip()
323 if not os.path.isabs(git_dir):
324 git_dir = os.path.join(location, git_dir)
325 repo_root = os.path.abspath(os.path.join(git_dir, '..'))
326 return find_path_to_setup_from_repo_root(location, repo_root)
327
328 @classmethod
329 def get_url_rev_and_auth(cls, url):
330 # type: (str) -> Tuple[str, Optional[str], AuthInfo]
331 """
332 Prefixes stub URLs like 'user@hostname:user/repo.git' with 'ssh://'.
333 That's required because although they use SSH they sometimes don't
334 work with a ssh:// scheme (e.g. GitHub). But we need a scheme for
335 parsing. Hence we remove it again afterwards and return it as a stub.
336 """
337 # Works around an apparent Git bug
338 # (see https://article.gmane.org/gmane.comp.version-control.git/146500)
339 scheme, netloc, path, query, fragment = urlsplit(url)
340 if scheme.endswith('file'):
341 initial_slashes = path[:-len(path.lstrip('/'))]
342 newpath = (
343 initial_slashes +
344 urllib_request.url2pathname(path)
345 .replace('\\', '/').lstrip('/')
346 )
347 url = urlunsplit((scheme, netloc, newpath, query, fragment))
348 after_plus = scheme.find('+') + 1
349 url = scheme[:after_plus] + urlunsplit(
350 (scheme[after_plus:], netloc, newpath, query, fragment),
351 )
352
353 if '://' not in url:
354 assert 'file:' not in url
355 url = url.replace('git+', 'git+ssh://')
356 url, rev, user_pass = super(Git, cls).get_url_rev_and_auth(url)
357 url = url.replace('ssh://', '')
358 else:
359 url, rev, user_pass = super(Git, cls).get_url_rev_and_auth(url)
360
361 return url, rev, user_pass
362
363 @classmethod
364 def update_submodules(cls, location):
365 if not os.path.exists(os.path.join(location, '.gitmodules')):
366 return
367 cls.run_command(
368 ['submodule', 'update', '--init', '--recursive', '-q'],
369 cwd=location,
370 )
371
372 @classmethod
373 def controls_location(cls, location):
374 if super(Git, cls).controls_location(location):
375 return True
376 try:
377 r = cls.run_command(['rev-parse'],
378 cwd=location,
379 show_stdout=False,
380 on_returncode='ignore',
381 log_failed_cmd=False)
382 return not r
383 except BadCommand:
384 logger.debug("could not determine if %s is under git control "
385 "because git is not available", location)
386 return False
387
388
389 vcs.register(Git)