Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/pip/_internal/vcs/versioncontrol.py @ 1:56ad4e20f292 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:32:28 -0400
parents
children
comparison
equal deleted inserted replaced
0:d30785e31577 1:56ad4e20f292
1 """Handles all VCS (version control) support"""
2 from __future__ import absolute_import
3
4 import errno
5 import logging
6 import os
7 import shutil
8 import sys
9
10 from pip._vendor import pkg_resources
11 from pip._vendor.six.moves.urllib import parse as urllib_parse
12
13 from pip._internal.exceptions import BadCommand
14 from pip._internal.utils.misc import (
15 ask_path_exists, backup_dir, call_subprocess, display_path, rmtree,
16 )
17 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
18
19 if MYPY_CHECK_RUNNING:
20 from typing import (
21 Any, Dict, Iterable, List, Mapping, Optional, Text, Tuple, Type
22 )
23 from pip._internal.utils.ui import SpinnerInterface
24
25 AuthInfo = Tuple[Optional[str], Optional[str]]
26
27 __all__ = ['vcs']
28
29
30 logger = logging.getLogger(__name__)
31
32
33 def make_vcs_requirement_url(repo_url, rev, project_name, subdir=None):
34 """
35 Return the URL for a VCS requirement.
36
37 Args:
38 repo_url: the remote VCS url, with any needed VCS prefix (e.g. "git+").
39 project_name: the (unescaped) project name.
40 """
41 egg_project_name = pkg_resources.to_filename(project_name)
42 req = '{}@{}#egg={}'.format(repo_url, rev, egg_project_name)
43 if subdir:
44 req += '&subdirectory={}'.format(subdir)
45
46 return req
47
48
49 class RemoteNotFoundError(Exception):
50 pass
51
52
53 class RevOptions(object):
54
55 """
56 Encapsulates a VCS-specific revision to install, along with any VCS
57 install options.
58
59 Instances of this class should be treated as if immutable.
60 """
61
62 def __init__(
63 self,
64 vc_class, # type: Type[VersionControl]
65 rev=None, # type: Optional[str]
66 extra_args=None, # type: Optional[List[str]]
67 ):
68 # type: (...) -> None
69 """
70 Args:
71 vc_class: a VersionControl subclass.
72 rev: the name of the revision to install.
73 extra_args: a list of extra options.
74 """
75 if extra_args is None:
76 extra_args = []
77
78 self.extra_args = extra_args
79 self.rev = rev
80 self.vc_class = vc_class
81
82 def __repr__(self):
83 return '<RevOptions {}: rev={!r}>'.format(self.vc_class.name, self.rev)
84
85 @property
86 def arg_rev(self):
87 # type: () -> Optional[str]
88 if self.rev is None:
89 return self.vc_class.default_arg_rev
90
91 return self.rev
92
93 def to_args(self):
94 # type: () -> List[str]
95 """
96 Return the VCS-specific command arguments.
97 """
98 args = [] # type: List[str]
99 rev = self.arg_rev
100 if rev is not None:
101 args += self.vc_class.get_base_rev_args(rev)
102 args += self.extra_args
103
104 return args
105
106 def to_display(self):
107 # type: () -> str
108 if not self.rev:
109 return ''
110
111 return ' (to revision {})'.format(self.rev)
112
113 def make_new(self, rev):
114 # type: (str) -> RevOptions
115 """
116 Make a copy of the current instance, but with a new rev.
117
118 Args:
119 rev: the name of the revision for the new object.
120 """
121 return self.vc_class.make_rev_options(rev, extra_args=self.extra_args)
122
123
124 class VcsSupport(object):
125 _registry = {} # type: Dict[str, VersionControl]
126 schemes = ['ssh', 'git', 'hg', 'bzr', 'sftp', 'svn']
127
128 def __init__(self):
129 # type: () -> None
130 # Register more schemes with urlparse for various version control
131 # systems
132 urllib_parse.uses_netloc.extend(self.schemes)
133 # Python >= 2.7.4, 3.3 doesn't have uses_fragment
134 if getattr(urllib_parse, 'uses_fragment', None):
135 urllib_parse.uses_fragment.extend(self.schemes)
136 super(VcsSupport, self).__init__()
137
138 def __iter__(self):
139 return self._registry.__iter__()
140
141 @property
142 def backends(self):
143 # type: () -> List[VersionControl]
144 return list(self._registry.values())
145
146 @property
147 def dirnames(self):
148 # type: () -> List[str]
149 return [backend.dirname for backend in self.backends]
150
151 @property
152 def all_schemes(self):
153 # type: () -> List[str]
154 schemes = [] # type: List[str]
155 for backend in self.backends:
156 schemes.extend(backend.schemes)
157 return schemes
158
159 def register(self, cls):
160 # type: (Type[VersionControl]) -> None
161 if not hasattr(cls, 'name'):
162 logger.warning('Cannot register VCS %s', cls.__name__)
163 return
164 if cls.name not in self._registry:
165 self._registry[cls.name] = cls()
166 logger.debug('Registered VCS backend: %s', cls.name)
167
168 def unregister(self, name):
169 # type: (str) -> None
170 if name in self._registry:
171 del self._registry[name]
172
173 def get_backend_for_dir(self, location):
174 # type: (str) -> Optional[VersionControl]
175 """
176 Return a VersionControl object if a repository of that type is found
177 at the given directory.
178 """
179 for vcs_backend in self._registry.values():
180 if vcs_backend.controls_location(location):
181 logger.debug('Determine that %s uses VCS: %s',
182 location, vcs_backend.name)
183 return vcs_backend
184 return None
185
186 def get_backend(self, name):
187 # type: (str) -> Optional[VersionControl]
188 """
189 Return a VersionControl object or None.
190 """
191 name = name.lower()
192 return self._registry.get(name)
193
194
195 vcs = VcsSupport()
196
197
198 class VersionControl(object):
199 name = ''
200 dirname = ''
201 repo_name = ''
202 # List of supported schemes for this Version Control
203 schemes = () # type: Tuple[str, ...]
204 # Iterable of environment variable names to pass to call_subprocess().
205 unset_environ = () # type: Tuple[str, ...]
206 default_arg_rev = None # type: Optional[str]
207
208 @classmethod
209 def should_add_vcs_url_prefix(cls, remote_url):
210 """
211 Return whether the vcs prefix (e.g. "git+") should be added to a
212 repository's remote url when used in a requirement.
213 """
214 return not remote_url.lower().startswith('{}:'.format(cls.name))
215
216 @classmethod
217 def get_subdirectory(cls, repo_dir):
218 """
219 Return the path to setup.py, relative to the repo root.
220 """
221 return None
222
223 @classmethod
224 def get_requirement_revision(cls, repo_dir):
225 """
226 Return the revision string that should be used in a requirement.
227 """
228 return cls.get_revision(repo_dir)
229
230 @classmethod
231 def get_src_requirement(cls, repo_dir, project_name):
232 """
233 Return the requirement string to use to redownload the files
234 currently at the given repository directory.
235
236 Args:
237 project_name: the (unescaped) project name.
238
239 The return value has a form similar to the following:
240
241 {repository_url}@{revision}#egg={project_name}
242 """
243 repo_url = cls.get_remote_url(repo_dir)
244 if repo_url is None:
245 return None
246
247 if cls.should_add_vcs_url_prefix(repo_url):
248 repo_url = '{}+{}'.format(cls.name, repo_url)
249
250 revision = cls.get_requirement_revision(repo_dir)
251 subdir = cls.get_subdirectory(repo_dir)
252 req = make_vcs_requirement_url(repo_url, revision, project_name,
253 subdir=subdir)
254
255 return req
256
257 @staticmethod
258 def get_base_rev_args(rev):
259 """
260 Return the base revision arguments for a vcs command.
261
262 Args:
263 rev: the name of a revision to install. Cannot be None.
264 """
265 raise NotImplementedError
266
267 @classmethod
268 def make_rev_options(cls, rev=None, extra_args=None):
269 # type: (Optional[str], Optional[List[str]]) -> RevOptions
270 """
271 Return a RevOptions object.
272
273 Args:
274 rev: the name of a revision to install.
275 extra_args: a list of extra options.
276 """
277 return RevOptions(cls, rev, extra_args=extra_args)
278
279 @classmethod
280 def _is_local_repository(cls, repo):
281 # type: (str) -> bool
282 """
283 posix absolute paths start with os.path.sep,
284 win32 ones start with drive (like c:\\folder)
285 """
286 drive, tail = os.path.splitdrive(repo)
287 return repo.startswith(os.path.sep) or bool(drive)
288
289 def export(self, location, url):
290 """
291 Export the repository at the url to the destination location
292 i.e. only download the files, without vcs informations
293
294 :param url: the repository URL starting with a vcs prefix.
295 """
296 raise NotImplementedError
297
298 @classmethod
299 def get_netloc_and_auth(cls, netloc, scheme):
300 """
301 Parse the repository URL's netloc, and return the new netloc to use
302 along with auth information.
303
304 Args:
305 netloc: the original repository URL netloc.
306 scheme: the repository URL's scheme without the vcs prefix.
307
308 This is mainly for the Subversion class to override, so that auth
309 information can be provided via the --username and --password options
310 instead of through the URL. For other subclasses like Git without
311 such an option, auth information must stay in the URL.
312
313 Returns: (netloc, (username, password)).
314 """
315 return netloc, (None, None)
316
317 @classmethod
318 def get_url_rev_and_auth(cls, url):
319 # type: (str) -> Tuple[str, Optional[str], AuthInfo]
320 """
321 Parse the repository URL to use, and return the URL, revision,
322 and auth info to use.
323
324 Returns: (url, rev, (username, password)).
325 """
326 scheme, netloc, path, query, frag = urllib_parse.urlsplit(url)
327 if '+' not in scheme:
328 raise ValueError(
329 "Sorry, {!r} is a malformed VCS url. "
330 "The format is <vcs>+<protocol>://<url>, "
331 "e.g. svn+http://myrepo/svn/MyApp#egg=MyApp".format(url)
332 )
333 # Remove the vcs prefix.
334 scheme = scheme.split('+', 1)[1]
335 netloc, user_pass = cls.get_netloc_and_auth(netloc, scheme)
336 rev = None
337 if '@' in path:
338 path, rev = path.rsplit('@', 1)
339 url = urllib_parse.urlunsplit((scheme, netloc, path, query, ''))
340 return url, rev, user_pass
341
342 @staticmethod
343 def make_rev_args(username, password):
344 """
345 Return the RevOptions "extra arguments" to use in obtain().
346 """
347 return []
348
349 def get_url_rev_options(self, url):
350 # type: (str) -> Tuple[str, RevOptions]
351 """
352 Return the URL and RevOptions object to use in obtain() and in
353 some cases export(), as a tuple (url, rev_options).
354 """
355 url, rev, user_pass = self.get_url_rev_and_auth(url)
356 username, password = user_pass
357 extra_args = self.make_rev_args(username, password)
358 rev_options = self.make_rev_options(rev, extra_args=extra_args)
359
360 return url, rev_options
361
362 @staticmethod
363 def normalize_url(url):
364 # type: (str) -> str
365 """
366 Normalize a URL for comparison by unquoting it and removing any
367 trailing slash.
368 """
369 return urllib_parse.unquote(url).rstrip('/')
370
371 @classmethod
372 def compare_urls(cls, url1, url2):
373 # type: (str, str) -> bool
374 """
375 Compare two repo URLs for identity, ignoring incidental differences.
376 """
377 return (cls.normalize_url(url1) == cls.normalize_url(url2))
378
379 def fetch_new(self, dest, url, rev_options):
380 """
381 Fetch a revision from a repository, in the case that this is the
382 first fetch from the repository.
383
384 Args:
385 dest: the directory to fetch the repository to.
386 rev_options: a RevOptions object.
387 """
388 raise NotImplementedError
389
390 def switch(self, dest, url, rev_options):
391 """
392 Switch the repo at ``dest`` to point to ``URL``.
393
394 Args:
395 rev_options: a RevOptions object.
396 """
397 raise NotImplementedError
398
399 def update(self, dest, url, rev_options):
400 """
401 Update an already-existing repo to the given ``rev_options``.
402
403 Args:
404 rev_options: a RevOptions object.
405 """
406 raise NotImplementedError
407
408 @classmethod
409 def is_commit_id_equal(cls, dest, name):
410 """
411 Return whether the id of the current commit equals the given name.
412
413 Args:
414 dest: the repository directory.
415 name: a string name.
416 """
417 raise NotImplementedError
418
419 def obtain(self, dest, url):
420 # type: (str, str) -> None
421 """
422 Install or update in editable mode the package represented by this
423 VersionControl object.
424
425 :param dest: the repository directory in which to install or update.
426 :param url: the repository URL starting with a vcs prefix.
427 """
428 url, rev_options = self.get_url_rev_options(url)
429
430 if not os.path.exists(dest):
431 self.fetch_new(dest, url, rev_options)
432 return
433
434 rev_display = rev_options.to_display()
435 if self.is_repository_directory(dest):
436 existing_url = self.get_remote_url(dest)
437 if self.compare_urls(existing_url, url):
438 logger.debug(
439 '%s in %s exists, and has correct URL (%s)',
440 self.repo_name.title(),
441 display_path(dest),
442 url,
443 )
444 if not self.is_commit_id_equal(dest, rev_options.rev):
445 logger.info(
446 'Updating %s %s%s',
447 display_path(dest),
448 self.repo_name,
449 rev_display,
450 )
451 self.update(dest, url, rev_options)
452 else:
453 logger.info('Skipping because already up-to-date.')
454 return
455
456 logger.warning(
457 '%s %s in %s exists with URL %s',
458 self.name,
459 self.repo_name,
460 display_path(dest),
461 existing_url,
462 )
463 prompt = ('(s)witch, (i)gnore, (w)ipe, (b)ackup ',
464 ('s', 'i', 'w', 'b'))
465 else:
466 logger.warning(
467 'Directory %s already exists, and is not a %s %s.',
468 dest,
469 self.name,
470 self.repo_name,
471 )
472 # https://github.com/python/mypy/issues/1174
473 prompt = ('(i)gnore, (w)ipe, (b)ackup ', # type: ignore
474 ('i', 'w', 'b'))
475
476 logger.warning(
477 'The plan is to install the %s repository %s',
478 self.name,
479 url,
480 )
481 response = ask_path_exists('What to do? %s' % prompt[0], prompt[1])
482
483 if response == 'a':
484 sys.exit(-1)
485
486 if response == 'w':
487 logger.warning('Deleting %s', display_path(dest))
488 rmtree(dest)
489 self.fetch_new(dest, url, rev_options)
490 return
491
492 if response == 'b':
493 dest_dir = backup_dir(dest)
494 logger.warning(
495 'Backing up %s to %s', display_path(dest), dest_dir,
496 )
497 shutil.move(dest, dest_dir)
498 self.fetch_new(dest, url, rev_options)
499 return
500
501 # Do nothing if the response is "i".
502 if response == 's':
503 logger.info(
504 'Switching %s %s to %s%s',
505 self.repo_name,
506 display_path(dest),
507 url,
508 rev_display,
509 )
510 self.switch(dest, url, rev_options)
511
512 def unpack(self, location, url):
513 # type: (str, str) -> None
514 """
515 Clean up current location and download the url repository
516 (and vcs infos) into location
517
518 :param url: the repository URL starting with a vcs prefix.
519 """
520 if os.path.exists(location):
521 rmtree(location)
522 self.obtain(location, url=url)
523
524 @classmethod
525 def get_remote_url(cls, location):
526 """
527 Return the url used at location
528
529 Raises RemoteNotFoundError if the repository does not have a remote
530 url configured.
531 """
532 raise NotImplementedError
533
534 @classmethod
535 def get_revision(cls, location):
536 """
537 Return the current commit id of the files at the given location.
538 """
539 raise NotImplementedError
540
541 @classmethod
542 def run_command(
543 cls,
544 cmd, # type: List[str]
545 show_stdout=True, # type: bool
546 cwd=None, # type: Optional[str]
547 on_returncode='raise', # type: str
548 extra_ok_returncodes=None, # type: Optional[Iterable[int]]
549 command_desc=None, # type: Optional[str]
550 extra_environ=None, # type: Optional[Mapping[str, Any]]
551 spinner=None # type: Optional[SpinnerInterface]
552 ):
553 # type: (...) -> Text
554 """
555 Run a VCS subcommand
556 This is simply a wrapper around call_subprocess that adds the VCS
557 command name, and checks that the VCS is available
558 """
559 cmd = [cls.name] + cmd
560 try:
561 return call_subprocess(cmd, show_stdout, cwd,
562 on_returncode=on_returncode,
563 extra_ok_returncodes=extra_ok_returncodes,
564 command_desc=command_desc,
565 extra_environ=extra_environ,
566 unset_environ=cls.unset_environ,
567 spinner=spinner)
568 except OSError as e:
569 # errno.ENOENT = no such file or directory
570 # In other words, the VCS executable isn't available
571 if e.errno == errno.ENOENT:
572 raise BadCommand(
573 'Cannot find command %r - do you have '
574 '%r installed and in your '
575 'PATH?' % (cls.name, cls.name))
576 else:
577 raise # re-raise exception if a different error occurred
578
579 @classmethod
580 def is_repository_directory(cls, path):
581 # type: (str) -> bool
582 """
583 Return whether a directory path is a repository directory.
584 """
585 logger.debug('Checking in %s for %s (%s)...',
586 path, cls.dirname, cls.name)
587 return os.path.exists(os.path.join(path, cls.dirname))
588
589 @classmethod
590 def controls_location(cls, location):
591 # type: (str) -> bool
592 """
593 Check if a location is controlled by the vcs.
594 It is meant to be overridden to implement smarter detection
595 mechanisms for specific vcs.
596
597 This can do more than is_repository_directory() alone. For example,
598 the Git override checks that Git is actually available.
599 """
600 return cls.is_repository_directory(location)