Coverage for app/venv/lib/python3.14/site-packages/weblate/vcs/mercurial.py: 23%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5"""Mercurial version control system abstraction for Weblate needs.""" 

6 

7from __future__ import annotations 

8 

9import os.path 

10import re 

11from configparser import RawConfigParser 

12from pathlib import Path 

13from shutil import which 

14from typing import TYPE_CHECKING, ClassVar 

15 

16from django.utils.translation import gettext_lazy 

17 

18from weblate.auth.utils import format_address 

19from weblate.vcs.base import Repository, RepositoryError 

20from weblate.vcs.ssh import SSH_WRAPPER 

21 

22if TYPE_CHECKING: 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 from collections.abc import Iterator 

24 from datetime import datetime 

25 

26 from django_stubs_ext import StrOrPromise 

27 

28 

29VERSION_RE = re.compile(r".*\(version ([^)]*)\).*") 

30 

31 

32class HgRepository(Repository): 

33 """Repository implementation for Mercurial.""" 

34 

35 _cmd: ClassVar[str] = "rhg" if which("rhg") is not None else "hg" 

36 _cmd_last_revision: ClassVar[list[str]] = [ 

37 "log", 

38 "--limit", 

39 "1", 

40 "--template", 

41 "{node}", 

42 ] 

43 _cmd_last_remote_revision: ClassVar[list[str]] = [ 

44 "log", 

45 "--limit", 

46 "1", 

47 "--template", 

48 "{node}", 

49 "--branch", 

50 ".", 

51 ] 

52 _cmd_list_changed_files: ClassVar[list[str]] = ["status", "--rev"] 

53 _version: ClassVar[str | None] = None 

54 

55 name: ClassVar[StrOrPromise] = "Mercurial" 

56 push_label: ClassVar[StrOrPromise] = gettext_lazy( 

57 "This will push changes to the upstream Mercurial repository." 

58 ) 

59 req_version: ClassVar[str] = "6.8" 

60 default_branch: ClassVar[str] = "default" 

61 ref_to_remote: ClassVar[str] = "head() and branch(.) and not closed() - ." 

62 ref_from_remote: ClassVar[str] = "outgoing()" 

63 

64 def is_valid(self): 

65 """Check whether this is a valid repository.""" 

66 return os.path.exists(os.path.join(self.path, ".hg", "requires")) 

67 

68 @classmethod 

69 def create_blank_repository(cls, path: str) -> None: 

70 """Initialize the repository.""" 

71 cls._popen(["init", path]) 

72 

73 def check_config(self) -> None: 

74 """Check VCS configuration.""" 

75 self.set_config_values(("ui", "ssh", SSH_WRAPPER.filename.as_posix())) 

76 

77 @classmethod 

78 def _clone(cls, source: str, target: str, branch: str) -> None: 

79 """Clone repository.""" 

80 cls._popen(["clone", f"--branch={branch}", "--", source, target]) 

81 

82 def get_config(self, section: str, option: str) -> str | None: 

83 """Read entry from configuration.""" 

84 result = None 

85 filename = os.path.join(self.path, ".hg", "hgrc") 

86 config = RawConfigParser() 

87 config.read(filename) 

88 if config.has_option(section, option): 

89 result = config.get(section, option) 

90 return result 

91 

92 def set_config_values(self, *values: tuple[str, str, str]) -> None: 

93 """Set entry in local configuration.""" 

94 if not self.lock.is_locked: 

95 msg = "Repository operation without lock held!" 

96 raise RuntimeError(msg) 

97 filename = os.path.join(self.path, ".hg", "hgrc") 

98 self.set_config_file(filename, *values) 

99 

100 @staticmethod 

101 def set_config_file(filename: str | Path, *values: tuple[str, str, str]) -> None: 

102 config = RawConfigParser() 

103 config.read(filename) 

104 changed = False 

105 for section, option, value in values: 

106 if not config.has_section(section): 

107 config.add_section(section) 

108 if ( 

109 config.has_option(section, option) 

110 and config.get(section, option) == value 

111 ): 

112 continue 

113 config.set(section, option, value) 

114 changed = True 

115 if changed: 

116 with open(filename, "w") as handle: 

117 config.write(handle) 

118 

119 def set_committer(self, name, mail) -> None: 

120 """Configure committer name.""" 

121 self.set_config_values(("ui", "username", format_address(name, mail))) 

122 

123 def reset(self) -> None: 

124 """Reset working copy to match remote branch.""" 

125 self.set_config_values(("extensions", "strip", "")) 

126 self.execute(["update", "--clean", "remote(.)"]) 

127 if self.needs_push(): 

128 self.execute(["strip", "roots(outgoing())"]) 

129 self.clean_revision_cache() 

130 

131 def configure_merge(self) -> None: 

132 """Select the correct merge tool.""" 

133 updates: list[tuple[str, str, str]] = [ 

134 ("ui", "merge", "internal:merge"), 

135 ] 

136 merge_driver = self.get_merge_driver("po") 

137 if merge_driver is not None: 

138 updates.extend( 

139 ( 

140 ( 

141 "merge-tools", 

142 "weblate-merge-gettext-po.executable", 

143 merge_driver, 

144 ), 

145 ( 

146 "merge-tools", 

147 "weblate-merge-gettext-po.args", 

148 "$base $local $other $output", 

149 ), 

150 ( 

151 "merge-patterns", 

152 "**.po", 

153 "weblate-merge-gettext-po", 

154 ), 

155 ) 

156 ) 

157 self.set_config_values(*updates) 

158 

159 def rebase(self, abort=False) -> None: 

160 """Rebase working copy on top of remote branch.""" 

161 self.set_config_values(("extensions", "rebase", "")) 

162 if abort: 

163 self.execute(["rebase", "--abort"]) 

164 elif self.needs_merge(): 

165 if self.needs_ff(): 

166 self.execute(["update", "--clean", "remote(.)"]) 

167 else: 

168 self.configure_merge() 

169 try: 

170 self.execute(["rebase", "-d", "remote(.)"]) 

171 except RepositoryError as error: 

172 # Mercurial 3.8 changed error code and output 

173 if ( 

174 error.retcode in {1, 255} 

175 and "nothing to rebase" in error.args[0] 

176 ): 

177 self.execute(["update", "--clean", "remote(.)"]) 

178 self.clean_revision_cache() 

179 return 

180 raise 

181 self.clean_revision_cache() 

182 

183 def merge( 

184 self, abort: bool = False, message: str | None = None, no_ff: bool = False 

185 ) -> None: 

186 """Merge remote branch or reverts the merge.""" 

187 if abort: 

188 self.execute(["update", "--clean", "."]) 

189 elif self.needs_merge(): 

190 if self.needs_ff() and not no_ff: 

191 self.execute(["update", "--clean", "remote(.)"]) 

192 else: 

193 self.configure_merge() 

194 # Fallback to merge 

195 try: 

196 self.execute(["merge", "-r", "remote(.)"]) 

197 except RepositoryError as error: 

198 if error.retcode == 255: 

199 # Nothing to merge 

200 self.clean_revision_cache() 

201 return 

202 raise 

203 self.execute(["commit", "--message", "Merge"]) 

204 self.clean_revision_cache() 

205 

206 def needs_commit(self, filenames: list[str] | None = None): 

207 """Check whether repository needs commit.""" 

208 cmd = ["status", "--"] 

209 if filenames: 

210 cmd.extend(filenames) 

211 status = self.execute(cmd, needs_lock=False) 

212 return bool(status) 

213 

214 def _get_revision_info(self, revision): 

215 """Return dictionary with detailed revision information.""" 

216 template = """ 

217 author_name: {person(author)} 

218 author_email: {email(author)} 

219 author: {author} 

220 authordate: {rfc822date(date)} 

221 commit_name: {person(author)} 

222 commit_email: {email(author)} 

223 commit: {author} 

224 commitdate: {rfc822date(date)} 

225 shortrevision: {short(node)} 

226 message: 

227 {desc} 

228 """ 

229 text = self.execute( 

230 ["log", "--limit", "1", "--template", template, "--rev", revision], 

231 needs_lock=False, 

232 merge_err=False, 

233 ) 

234 

235 result = {"revision": revision} 

236 

237 message = [] 

238 header = True 

239 

240 for line in text.splitlines(): 

241 line = line.strip() 

242 if not line: 

243 continue 

244 if not header: 

245 message.append(line) 

246 continue 

247 if line == "message:": 

248 header = False 

249 continue 

250 name, value = line.strip().split(":", 1) 

251 value = value.strip() 

252 name = name.lower() 

253 result[name] = value 

254 

255 result["message"] = "\n".join(message) 

256 result["summary"] = message[0] 

257 

258 return result 

259 

260 def log_revisions(self, refspec): 

261 """Return revisin log for given refspec.""" 

262 return self.execute( 

263 ["log", "--template", "{node}\n", "--rev", refspec], 

264 needs_lock=False, 

265 merge_err=False, 

266 ).splitlines() 

267 

268 def needs_ff(self): 

269 """ 

270 Check whether repository needs a fast-forward to upstream. 

271 

272 Checks whether the path to the upstream is linear. 

273 """ 

274 return bool(self.log_revisions(".::remote(.) - .")) 

275 

276 @classmethod 

277 def _get_version(cls): 

278 """Return VCS program version.""" 

279 output = cls._popen(["version", "-q"], merge_err=False) 

280 matches = VERSION_RE.match(output) 

281 if matches is None: 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true

282 msg = f"Could not parse version string: {output}" 

283 raise OSError(msg) 

284 return matches.group(1) 

285 

286 def commit( 

287 self, 

288 message: str, 

289 author: str | None = None, 

290 timestamp: datetime | None = None, 

291 files: list[str] | None = None, 

292 ) -> bool: 

293 """Create new revision.""" 

294 # Build the commit command 

295 cmd = ["commit", "--message", message] 

296 if author is not None: 

297 cmd.extend(["--user", author]) 

298 if timestamp is not None: 

299 cmd.extend(["--date", timestamp.ctime()]) 

300 

301 # Add files one by one, this has to deal with 

302 # removed, untracked and non existing files 

303 if files is not None: 

304 for name in files: 

305 try: 

306 self.execute(["add", "--", name]) 

307 except RepositoryError: 

308 try: 

309 self.execute(["remove", "--", name]) 

310 except RepositoryError: 

311 continue 

312 cmd.append(name) 

313 

314 # Bail out if there is nothing to commit. 

315 # This can easily happen with squashing and reverting changes. 

316 if not self.needs_commit(files): 

317 return False 

318 

319 # Execute it 

320 self.execute(cmd) 

321 # Clean cache 

322 self.clean_revision_cache() 

323 

324 return True 

325 

326 def remove(self, files: list[str], message: str, author: str | None = None) -> None: 

327 """Remove files and creates new revision.""" 

328 self.execute(["remove", "--force", "--", *files]) 

329 self.commit(message, author) 

330 

331 def configure_remote( 

332 self, pull_url: str, push_url: str, branch: str, fast: bool = True 

333 ) -> None: 

334 """Configure remote repository.""" 

335 old_pull = self.get_config("paths", "default") 

336 old_push = self.get_config("paths", "default-push") 

337 

338 updates: list[tuple[str, str, str]] = [ 

339 # We enable some necessary extensions here 

340 ("extensions", "strip", ""), 

341 ("extensions", "rebase", ""), 

342 ("experimental", "evolution", "all"), 

343 ("phases", "publish", "False"), 

344 ] 

345 

346 if old_pull != pull_url: 

347 # No origin existing or URL changed? 

348 updates.append(("paths", "default", pull_url)) 

349 

350 if old_push != push_url: 

351 updates.append(("paths", "default-push", push_url)) 

352 

353 self.set_config_values(*updates) 

354 

355 self.branch = branch 

356 

357 def on_branch(self, branch): 

358 return branch == self.execute(["branch"], merge_err=False).strip() 

359 

360 def configure_branch(self, branch) -> None: 

361 """Configure repository branch.""" 

362 if not self.on_branch(branch): 

363 self.execute(["update", "--", branch]) 

364 self.branch = branch 

365 

366 def describe(self): 

367 """Verbosely describes current revision.""" 

368 return self.execute( 

369 [ 

370 "log", 

371 "-r", 

372 ".", 

373 "--template", 

374 "{latesttag}-{latesttagdistance}-{node|short}", 

375 ], 

376 needs_lock=False, 

377 merge_err=False, 

378 ).strip() 

379 

380 def push(self, branch) -> None: 

381 """Push given branch to remote repository.""" 

382 try: 

383 self.execute(["push", f"--branch={self.branch}"]) 

384 except RepositoryError as error: 

385 if error.retcode == 1: 

386 # No changes found 

387 return 

388 raise 

389 

390 def get_file(self, path, revision): 

391 """Return content of file at given revision.""" 

392 return self.execute( 

393 ["cat", "--rev", revision, path], needs_lock=False, merge_err=False 

394 ) 

395 

396 def cleanup(self) -> None: 

397 """Remove not tracked files from the repository.""" 

398 self.set_config_values(("extensions", "purge", "")) 

399 self.execute(["purge"]) 

400 

401 def update_remote(self) -> None: 

402 """Update remote repository.""" 

403 self.execute(["pull", f"--branch={self.branch}"]) 

404 self.clean_revision_cache() 

405 

406 def parse_changed_files(self, lines: list[str]) -> Iterator[str]: 

407 """Parse output with changed files.""" 

408 # Strip action prefix we do not use 

409 yield from (line[2:] for line in lines) 

410 

411 def list_changed_files(self, refspec: str) -> list: 

412 try: 

413 return super().list_changed_files(refspec) 

414 except RepositoryError as error: 

415 if error.retcode == 255: 

416 # Empty revision set 

417 return [] 

418 raise 

419 

420 @classmethod 

421 def global_setup(cls) -> None: 

422 """Perform global settings.""" 

423 if cls._cmd == "rhg": 423 ↛ 424line 423 didn't jump to line 424 because the condition on line 423 was never true

424 cls.set_config_file( 

425 Path.home() / ".hgrc", 

426 ("rhg", "fallback-executable", "hg"), 

427 ("rhg", "on-unsupported", "fallback"), 

428 ) 

429 

430 def show(self, revision: str) -> str: 

431 return self.execute( 

432 ["diff", "--change", revision], needs_lock=False, merge_err=False 

433 )