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