File size: 3,734 Bytes
a481509 |
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 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
from __future__ import annotations
import logging
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:
flags = ["--quiet"]
elif verbosity == 1:
flags = []
else:
flags = [f"-{'v'*verbosity}"]
cmd_args = make_command(
"checkout", "--lightweight", *flags, rev_options.to_args(), url, dest
)
self.run_command(cmd_args)
def switch(
self,
dest: str,
url: HiddenText,
rev_options: RevOptions,
verbosity: int = 0,
) -> None:
self.run_command(make_command("switch", url), cwd=dest)
def update(
self,
dest: str,
url: HiddenText,
rev_options: RevOptions,
verbosity: int = 0,
) -> None:
flags = []
if verbosity <= 0:
flags.append("-q")
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", *flags, url)
self.run_command(cmd_args, cwd=dest)
cmd_args = make_command("update", *flags, rev_options.to_args())
self.run_command(cmd_args, cwd=dest)
@classmethod
def get_url_rev_and_auth(cls, url: str) -> tuple[str, str | None, 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: str | None) -> bool:
"""Always assume the versions don't match"""
return False
vcs.register(Bazaar)
|