python (3.12.0)
1 import configparser
2 import logging
3 import os
4 from typing import List, Optional, Tuple
5
6 from pip._internal.exceptions import BadCommand, InstallationError
7 from pip._internal.utils.misc import HiddenText, display_path
8 from pip._internal.utils.subprocess import make_command
9 from pip._internal.utils.urls import path_to_url
10 from pip._internal.vcs.versioncontrol import (
11 RevOptions,
12 VersionControl,
13 find_path_to_project_root_from_repo_root,
14 vcs,
15 )
16
17 logger = logging.getLogger(__name__)
18
19
20 class ESC[4;38;5;81mMercurial(ESC[4;38;5;149mVersionControl):
21 name = "hg"
22 dirname = ".hg"
23 repo_name = "clone"
24 schemes = (
25 "hg+file",
26 "hg+http",
27 "hg+https",
28 "hg+ssh",
29 "hg+static-http",
30 )
31
32 @staticmethod
33 def get_base_rev_args(rev: str) -> List[str]:
34 return ["-r", rev]
35
36 def fetch_new(
37 self, dest: str, url: HiddenText, rev_options: RevOptions, verbosity: int
38 ) -> None:
39 rev_display = rev_options.to_display()
40 logger.info(
41 "Cloning hg %s%s to %s",
42 url,
43 rev_display,
44 display_path(dest),
45 )
46 if verbosity <= 0:
47 flags: Tuple[str, ...] = ("--quiet",)
48 elif verbosity == 1:
49 flags = ()
50 elif verbosity == 2:
51 flags = ("--verbose",)
52 else:
53 flags = ("--verbose", "--debug")
54 self.run_command(make_command("clone", "--noupdate", *flags, url, dest))
55 self.run_command(
56 make_command("update", *flags, rev_options.to_args()),
57 cwd=dest,
58 )
59
60 def switch(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None:
61 repo_config = os.path.join(dest, self.dirname, "hgrc")
62 config = configparser.RawConfigParser()
63 try:
64 config.read(repo_config)
65 config.set("paths", "default", url.secret)
66 with open(repo_config, "w") as config_file:
67 config.write(config_file)
68 except (OSError, configparser.NoSectionError) as exc:
69 logger.warning("Could not switch Mercurial repository to %s: %s", url, exc)
70 else:
71 cmd_args = make_command("update", "-q", rev_options.to_args())
72 self.run_command(cmd_args, cwd=dest)
73
74 def update(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None:
75 self.run_command(["pull", "-q"], cwd=dest)
76 cmd_args = make_command("update", "-q", rev_options.to_args())
77 self.run_command(cmd_args, cwd=dest)
78
79 @classmethod
80 def get_remote_url(cls, location: str) -> str:
81 url = cls.run_command(
82 ["showconfig", "paths.default"],
83 show_stdout=False,
84 stdout_only=True,
85 cwd=location,
86 ).strip()
87 if cls._is_local_repository(url):
88 url = path_to_url(url)
89 return url.strip()
90
91 @classmethod
92 def get_revision(cls, location: str) -> str:
93 """
94 Return the repository-local changeset revision number, as an integer.
95 """
96 current_revision = cls.run_command(
97 ["parents", "--template={rev}"],
98 show_stdout=False,
99 stdout_only=True,
100 cwd=location,
101 ).strip()
102 return current_revision
103
104 @classmethod
105 def get_requirement_revision(cls, location: str) -> str:
106 """
107 Return the changeset identification hash, as a 40-character
108 hexadecimal string
109 """
110 current_rev_hash = cls.run_command(
111 ["parents", "--template={node}"],
112 show_stdout=False,
113 stdout_only=True,
114 cwd=location,
115 ).strip()
116 return current_rev_hash
117
118 @classmethod
119 def is_commit_id_equal(cls, dest: str, name: Optional[str]) -> bool:
120 """Always assume the versions don't match"""
121 return False
122
123 @classmethod
124 def get_subdirectory(cls, location: str) -> Optional[str]:
125 """
126 Return the path to Python project root, relative to the repo root.
127 Return None if the project root is in the repo root.
128 """
129 # find the repo root
130 repo_root = cls.run_command(
131 ["root"], show_stdout=False, stdout_only=True, cwd=location
132 ).strip()
133 if not os.path.isabs(repo_root):
134 repo_root = os.path.abspath(os.path.join(location, repo_root))
135 return find_path_to_project_root_from_repo_root(location, repo_root)
136
137 @classmethod
138 def get_repository_root(cls, location: str) -> Optional[str]:
139 loc = super().get_repository_root(location)
140 if loc:
141 return loc
142 try:
143 r = cls.run_command(
144 ["root"],
145 cwd=location,
146 show_stdout=False,
147 stdout_only=True,
148 on_returncode="raise",
149 log_failed_cmd=False,
150 )
151 except BadCommand:
152 logger.debug(
153 "could not determine if %s is under hg control "
154 "because hg is not available",
155 location,
156 )
157 return None
158 except InstallationError:
159 return None
160 return os.path.normpath(r.rstrip("\r\n"))
161
162
163 vcs.register(Mercurial)
Browse
Usage
Versions