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
|
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)
|
