Mercurial > repos > guerler > hhblits

comparison lib/python3.8/site-packages/pip/_internal/vcs/git.py @ 1:64071f2a4cf0 draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Deleted selected files
author guerler
date Mon, 27 Jul 2020 03:55:49 -0400
parents 9e54283cc701
children
comparison
equal deleted inserted replaced
0:9e54283cc701 1:64071f2a4cf0
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)