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