Spaces:
Sleeping
Sleeping
File size: 3,519 Bytes
6219ec9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
import logging
from typing import List, Optional, Tuple
from pip._internal.utils.misc import HiddenText, display_path
from pip._internal.utils.subprocess import make_command
from pip._internal.utils.urls import path_to_url
from pip._internal.vcs.versioncontrol import (
AuthInfo,
RemoteNotFoundError,
RevOptions,
VersionControl,
vcs,
)
logger = logging.getLogger(__name__)
class Bazaar(VersionControl):
name = "bzr"
dirname = ".bzr"
repo_name = "branch"
schemes = (
"bzr+http",
"bzr+https",
"bzr+ssh",
"bzr+sftp",
"bzr+ftp",
"bzr+lp",
"bzr+file",
)
@staticmethod
def get_base_rev_args(rev: str) -> List[str]:
return ["-r", rev]
def fetch_new(
self, dest: str, url: HiddenText, rev_options: RevOptions, verbosity: int
) -> None:
rev_display = rev_options.to_display()
logger.info(
"Checking out %s%s to %s",
url,
rev_display,
display_path(dest),
)
if verbosity <= 0:
flag = "--quiet"
elif verbosity == 1:
flag = ""
else:
flag = f"-{'v'*verbosity}"
cmd_args = make_command(
"checkout", "--lightweight", flag, rev_options.to_args(), url, dest
)
self.run_command(cmd_args)
def switch(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None:
self.run_command(make_command("switch", url), cwd=dest)
def update(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None:
output = self.run_command(
make_command("info"), show_stdout=False, stdout_only=True, cwd=dest
)
if output.startswith("Standalone "):
# Older versions of pip used to create standalone branches.
# Convert the standalone branch to a checkout by calling "bzr bind".
cmd_args = make_command("bind", "-q", url)
self.run_command(cmd_args, cwd=dest)
cmd_args = make_command("update", "-q", rev_options.to_args())
self.run_command(cmd_args, cwd=dest)
@classmethod
def get_url_rev_and_auth(cls, url: str) -> Tuple[str, Optional[str], AuthInfo]:
# hotfix the URL scheme after removing bzr+ from bzr+ssh:// re-add it
url, rev, user_pass = super().get_url_rev_and_auth(url)
if url.startswith("ssh://"):
url = "bzr+" + url
return url, rev, user_pass
@classmethod
def get_remote_url(cls, location: str) -> str:
urls = cls.run_command(
["info"], show_stdout=False, stdout_only=True, cwd=location
)
for line in urls.splitlines():
line = line.strip()
for x in ("checkout of branch: ", "parent branch: "):
if line.startswith(x):
repo = line.split(x)[1]
if cls._is_local_repository(repo):
return path_to_url(repo)
return repo
raise RemoteNotFoundError
@classmethod
def get_revision(cls, location: str) -> str:
revision = cls.run_command(
["revno"],
show_stdout=False,
stdout_only=True,
cwd=location,
)
return revision.splitlines()[-1]
@classmethod
def is_commit_id_equal(cls, dest: str, name: Optional[str]) -> bool:
"""Always assume the versions don't match"""
return False
vcs.register(Bazaar)
|