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

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

5"""Git-based version-control-system abstraction for Weblate needs."""

7from __future__ import annotations

9import logging

10import os

11import os.path

12import random

13import re

14import urllib.parse

15from configparser import NoOptionError, NoSectionError, RawConfigParser

16from contextlib import contextmanager

17from json import JSONDecodeError, dumps

18from pathlib import Path

19from time import sleep, time

20from typing import (

21 TYPE_CHECKING,

22 Any,

23 BinaryIO,

24 ClassVar,

25 NoReturn,

26 NotRequired,

27 Self,

28 TypedDict,

29 cast,

30)

31from urllib.parse import urlparse, urlunparse

32from zipfile import ZipFile

34import requests

35import sentry_sdk

36from django.conf import settings

37from django.core.cache import cache

38from django.utils.functional import cached_property

39from django.utils.translation import gettext, gettext_lazy

40from git.config import GitConfigParser

41from requests.exceptions import HTTPError

43from weblate.utils.data import data_dir, data_path

44from weblate.utils.errors import report_error

45from weblate.utils.files import is_excluded, remove_tree

46from weblate.utils.lock import WeblateLock, WeblateLockTimeoutError

47from weblate.utils.render import render_template

48from weblate.utils.xml import parse_xml

49from weblate.vcs.base import Repository, RepositoryError

50from weblate.vcs.gpg import get_gpg_sign_key

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

53 from collections.abc import Iterator

54 from datetime import datetime

56 from django_stubs_ext import StrOrPromise

57 from requests.auth import AuthBase

59 from weblate.trans.models import Component

62class GitCredentials(TypedDict):

63 url: str

64 owner: str

65 slug: str

66 hostname: str

67 username: str

68 token: str

69 scheme: str

70 push_scheme: str

71 workspace: NotRequired[str]

72 organization: NotRequired[str]

73 workItemIds: NotRequired[list[str]]

76class GitRepository(Repository):

77 """Repository implementation for Git."""

79 _cmd: ClassVar[str] = "git"

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

81 "log",

82 "-n",

83 "1",

84 "--format=format:%H",

85 "HEAD",

86 "--",

87 ]

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

89 "log",

90 "-n",

91 "1",

92 "--format=format:%H",

93 "@{upstream}",

94 "--",

95 ]

96 _cmd_list_changed_files: ClassVar[list[str]] = ["diff", "--name-status"]

97 _cmd_push: ClassVar[list[str]] = ["push"]

98 _cmd_status: ClassVar[list[str]] = ["--no-optional-locks", "status"]

100 name: ClassVar[StrOrPromise] = "Git"

101 push_label: ClassVar[StrOrPromise] = gettext_lazy(

102 "This will push changes to the upstream Git repository."

103 )

104 req_version: ClassVar[str | None] = "2.28"

105 default_branch: ClassVar[str] = "master"

106 ref_to_remote: ClassVar[str] = "..{0}"

107 ref_from_remote: ClassVar[str] = "{0}.."

108

109 def is_valid(self):

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

111 return os.path.exists(

112 os.path.join(self.path, ".git", "config")

113 ) or os.path.exists(os.path.join(self.path, "config"))

114

115 @classmethod

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

117 """Initialize the repository."""

118 cls._popen(

119 ["init", "--template=", "--initial-branch", cls.default_branch, path]

120 )

121

122 @classmethod

123 def get_remote_branch(cls, repo: str):

124 if not repo: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 return super().get_remote_branch(repo)

126 try:

127 result = cls._popen(["ls-remote", "--symref", "--", repo, "HEAD"])

128 except RepositoryError:

129 report_error("Listing remote branch")

130 return super().get_remote_branch(repo)

131 for line in result.splitlines():

132 if not line.startswith("ref: "):

133 continue

134 # Parses 'ref: refs/heads/main\tHEAD'

135 return line.split("\t")[0].split("refs/heads/")[1]

136

137 report_error("Could not figure out remote branch", message=True)

138 raise RepositoryError(0, "Could not figure out remote branch")

139

140 @staticmethod

141 def git_config_update(

142 filename: Path, *updates: tuple[str, str, str | None]

143 ) -> None:

144 # First, open file read-only to check current settings

145 modify = False

146 with GitConfigParser(file_or_files=filename, read_only=True) as config:

147 for section, key, value in updates:

148 try:

149 old = config.get(section, key)

150 if value is None: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 modify = True

152 break

153 if old == value:

154 continue

155 except (NoSectionError, NoOptionError):

156 pass

157 if value is not None:

158 modify = True

159 if not modify:

160 return

161 # In case changes are needed, open it for writing as that creates a lock

162 # file

163 with GitConfigParser(file_or_files=filename, read_only=False) as config:

164 for section, key, value in updates:

165 try:

166 old = config.get(section, key)

167 if value is None: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true

168 config.remove_option(section, key)

169 continue

170 if old == value:

171 continue

172 except NoSectionError:

173 if value is not None: 173 ↛ 177line 173 didn't jump to line 177 because the condition on line 173 was always true

174 config.add_section(section)

175 except NoOptionError:

176 pass

177 if value is not None:

178 config.set(section, key, value)

179

180 def config_update(self, *updates: tuple[str, str, str | None]) -> None:

181 filename = Path(self.path) / ".git" / "config"

182 self.git_config_update(filename, *updates)

183

184 def check_config(self) -> None:

185 """Check VCS configuration."""

186 self.config_update(("push", "default", "current"))

187

188 @staticmethod

189 def get_depth() -> Iterator[str]:

190 if settings.VCS_CLONE_DEPTH:

191 yield "--depth"

192 yield str(settings.VCS_CLONE_DEPTH)

193

194 @staticmethod

195 def _get_auth_args(repo: str) -> Iterator[str]:

196 if repo.startswith("https://"): 196 ↛ 197line 196 didn't jump to line 197 because the condition on line 196 was never true

197 parsed = urlparse(repo)

198 if parsed.password:

199 # proactive HTTP authentication, needs Git 2.46

200 yield "-c"

201 yield "http.proactiveAuth=auto"

202

203 def get_auth_args(self) -> list[str]:

204 if self.component: 204 ↛ 206line 204 didn't jump to line 206 because the condition on line 204 was always true

205 return list(self._get_auth_args(self.component.repo))

206 return []

207

208 @classmethod

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

210 """Clone repository."""

211 cls._popen(

212 [

213 *cls._get_auth_args(source),

214 "clone",

215 *cls.get_depth(),

216 "--branch",

217 branch,

218 "--",

219 source,

220 target,

221 ]

222 )

223

224 def get_config(self, path):

225 """Read entry from configuration."""

226 return self.execute(["config", path], needs_lock=False, merge_err=False).strip()

227

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

229 """Configure committer name."""

230 self.config_update(("user", "name", name), ("user", "email", mail))

231

232 def reset(self) -> None:

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

234 self.execute(["reset", "--hard", self.get_remote_branch_name()])

235 self.clean_revision_cache()

236

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

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

239 if abort:

240 if self.has_git_file("rebase-apply") or self.has_git_file("rebase-merge"):

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

242 if self.needs_commit():

243 self.execute(["reset", "--hard"])

244 else:

245 cmd = ["rebase"]

246 cmd.extend(self.get_gpg_sign_args())

247 cmd.append(self.get_remote_branch_name())

248 self.execute(cmd, environment={"WEBLATE_MERGE_SKIP": "1"})

249 self.clean_revision_cache()

250

251 def has_git_file(self, name):

252 return os.path.exists(os.path.join(self.path, ".git", name))

253

254 def has_rev(self, rev) -> bool:

255 try:

256 self.execute(["rev-parse", "--verify", rev], needs_lock=False)

257 except RepositoryError:

258 return False

259 return True

260

261 def merge(

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

263 ) -> None:

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

265 tmp = "weblate-merge-tmp"

266 if abort:

267 # Abort merge if there is one to abort

268 if self.has_rev("MERGE_HEAD"):

269 self.execute(["merge", "--abort"])

270 if self.needs_commit():

271 self.execute(["reset", "--hard"])

272 # Checkout original branch (we might be on tmp)

273 self.execute(["checkout", self.branch])

274 else:

275 self.delete_branch(tmp)

276 # We don't do simple git merge origin/branch as that leads

277 # to different merge order than expected and most GUI tools

278 # then show confusing diff (not changes done by Weblate, but

279 # changes merged into Weblate)

280 remote = self.get_remote_branch_name()

281 # Create local branch for upstream

282 self.execute(["branch", tmp, remote])

283 # Checkout upstream branch

284 self.execute(["checkout", tmp])

285 # Merge current Weblate changes, this can lead to conflict

286 cmd = [

287 "merge",

288 "--message",

289 message or f"Merge branch {remote!r} into Weblate",

290 ]

291 if no_ff:

292 cmd.append("--no-ff")

293 cmd.extend(self.get_gpg_sign_args())

294 cmd.append(self.branch)

295 self.execute(cmd)

296 # Checkout branch with Weblate changes

297 self.execute(["checkout", self.branch])

298 # Merge temporary branch (this is fast forward so does not create

299 # merge commit)

300 self.execute(["merge", tmp])

301

302 # Delete temporary branch

303 self.delete_branch(tmp)

304 self.clean_revision_cache()

305

306 def delete_branch(self, name) -> None:

307 if self.has_branch(name):

308 self.execute(["branch", "-D", name])

309

310 def needs_commit(self, filenames: list[str] | None = None) -> bool:

311 """Check whether repository needs commit."""

312 cmd = ["--no-optional-locks", "status", "--porcelain"]

313 if filenames:

314 cmd.extend(["--untracked-files=all", "--ignored=traditional", "--"])

315 cmd.extend(filenames)

316 with self.lock:

317 status = self.execute(cmd, merge_err=False)

318 return bool(status)

319

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

321 """

322 Get the content of the revision.

323

324 Used in tests.

325 """

326 return self.execute(["show", revision], needs_lock=False, merge_err=False)

327

328 @staticmethod

329 def get_gpg_sign_args():

330 sign_key = get_gpg_sign_key()

331 if sign_key: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true

332 return [f"--gpg-sign={sign_key}"]

333 return []

334

335 def _get_revision_info(self, revision):

336 """Return dictionary with detailed revision info."""

337 text = self.execute(

338 ["log", "-1", "--format=fuller", "--date=rfc", "--abbrev-commit", revision],

339 needs_lock=False,

340 merge_err=False,

341 )

342

343 result = {"revision": revision}

344

345 message = []

346

347 header = True

348

349 for line in text.splitlines():

350 if header:

351 if not line:

352 header = False

353 elif line.startswith("commit"):

354 result["shortrevision"] = line.split()[1]

355 else:

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

357 value = value.strip()

358 name = name.lower()

359 result[name] = value

360 if "<" in value:

361 parsed = value.split("<", 1)

362 result[f"{name}_name"] = parsed[0].strip()

363 result[f"{name}_email"] = parsed[1].rstrip(">")

364 else:

365 message.append(line.strip())

366

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

368 result["summary"] = message[0] if message else ""

369

370 return result

371

372 def log_revisions(self, refspec):

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

374 return self.execute(

375 ["log", "--format=format:%H", refspec, "--"],

376 needs_lock=False,

377 merge_err=False,

378 ).splitlines()

379

380 @classmethod

381 def _get_version(cls):

382 """Return VCS program version."""

383 return cls._popen(["--version"], merge_err=False).split()[2]

384

385 def commit(

386 self,

387 message: str,

388 author: str | None = None,

389 timestamp: datetime | None = None,

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

391 ) -> bool:

392 """Create new revision."""

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

394 # removed, untracked and non existing files

395 if files:

396 for name in files:

397 try:

398 self.execute(["add", "--force", "--", self.resolve_symlinks(name)])

399 except RepositoryError:

400 continue

401 else:

402 self.execute(["add", self.path])

403

404 # Bail out if there is nothing to commit.

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

406 if not self.needs_commit(files):

407 return False

408

409 # Build the commit command

410 cmd = ["commit", "--file", "-"]

411 if author:

412 cmd.extend(["--author", author])

413 if timestamp is not None:

414 cmd.extend(["--date", timestamp.isoformat()])

415 cmd.extend(self.get_gpg_sign_args())

416

417 # Execute it

418 self.execute(cmd, stdin=message)

419 # Clean cache

420 self.clean_revision_cache()

421

422 return True

423

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

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

426 self.execute(["rm", "--force", "--", *files])

427 self.commit(message, author)

428

429 def get_remote_configure(

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

431 ) -> tuple[tuple[str, str, str | None], ...]:

432 return ()

433

434 def configure_remote(

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

436 ) -> None:

437 """Configure remote repository."""

438 escaped_branch = dumps(branch, ensure_ascii=False)

439 self.config_update(

440 # Pull url

441 ('remote "origin"', "url", pull_url),

442 # Push URL, None remove it

443 ('remote "origin"', "pushurl", push_url or None),

444 # Fetch only current branch, others are fetched later in post_configure

445 (

446 'remote "origin"',

447 "fetch",

448 (

449 dumps(

450 f"+refs/heads/{branch}:refs/remotes/origin/{branch}",

451 ensure_ascii=False,

452 )

453 if fast

454 else "+refs/heads/*:refs/remotes/origin/*"

455 ),

456 ),

457 # Disable fetching tags

458 ('remote "origin"', "tagOpt", "--no-tags"),

459 # Set branch to track

460 (f"branch {escaped_branch}", "remote", "origin"),

461 (

462 f"branch {escaped_branch}",

463 "merge",

464 dumps(f"refs/heads/{branch}", ensure_ascii=False),

465 ),

466 *self.get_remote_configure(pull_url, push_url, branch, fast),

467 )

468 self.branch = branch

469

470 def list_branches(self, *args):

471 cmd = ["branch", "--list"]

472 cmd.extend(args)

473 # (we get additional * there indicating current branch)

474 return [

475 x.lstrip("*").strip()

476 for x in self.execute(cmd, needs_lock=False, merge_err=False).splitlines()

477 ]

478

479 def has_branch(self, branch):

480 branches = self.list_branches()

481 return branch in branches

482

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

484 """Configure repository branch."""

485 # Add branch

486 if not self.has_branch(branch): 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true

487 self.execute(["checkout", "-b", branch, f"origin/{branch}"])

488 else:

489 # Ensure it tracks correct upstream

490 self.config_update((f'branch "{branch}"', "remote", "origin"))

491

492 # Checkout

493 self.execute(["checkout", branch])

494 self.branch = branch

495

496 def describe(self) -> str:

497 """Verbosely describes current revision."""

498 return self.execute(

499 ["describe", "--always"], needs_lock=False, merge_err=False

500 ).strip()

501

502 @classmethod

503 def global_setup(cls) -> None:

504 """Perform global settings."""

505 home = data_path("home")

506 merge_driver = cls.get_merge_driver("po")

507 updates = [

508 ("user", "email", settings.DEFAULT_COMMITER_EMAIL),

509 ("user", "name", settings.DEFAULT_COMMITER_NAME),

510 (

511 'protocol "file"',

512 "allow",

513 "always" if settings.VCS_FILE_PROTOCOL else "never",

514 ),

515 ]

516 if merge_driver is not None: 516 ↛ 532line 516 didn't jump to line 532 because the condition on line 516 was always true

517 updates.extend(

518 (

519 (

520 'merge "weblate-merge-gettext-po"',

521 "name",

522 "Weblate merge driver for gettext PO files",

523 ),

524 (

525 'merge "weblate-merge-gettext-po"',

526 "driver",

527 f"{merge_driver} %O %A %B %P",

528 ),

529 )

530 )

531

532 filename = home / ".gitconfig"

533 attempts = 0

534 while attempts < 5: 534 ↛ 543line 534 didn't jump to line 543 because the condition on line 534 was always true

535 try:

536 cls.git_config_update(filename, *updates)

537 break

538 except OSError:

539 attempts += 1

540 sleep(attempts * 0.1)

541

542 # Use it for *.po by default

543 configfile = home / ".config" / "git" / "attributes"

544 if configfile.parent.is_dir(): 544 ↛ 545line 544 didn't jump to line 545 because the condition on line 544 was never true

545 configfile.parent.mkdir(parents=True, exist_ok=True)

546 configfile.write_text("*.po merge=weblate-merge-gettext-po\n")

547

548 def get_file(self, path, revision) -> str:

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

550 return self.execute(

551 ["show", f"{revision}:{path}"],

552 needs_lock=False,

553 merge_err=False,

554 )

555

556 def cleanup(self) -> None:

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

558 self.execute(["clean", "-f", "-d"])

559 # Remove possible stale branches

560 for branch in self.list_branches():

561 if branch != self.branch: 561 ↛ 562line 561 didn't jump to line 562 because the condition on line 561 was never true

562 self.execute(["branch", "--delete", "--force", branch])

563 # Remove any tags

564 for tag in self.execute(["tag", "--list"], merge_err=False).splitlines(): 564 ↛ 565line 564 didn't jump to line 565 because the loop on line 564 never started

565 self.execute(["tag", "--delete", tag])

566

567 def list_remote_branches(self):

568 return [

569 branch[7:]

570 for branch in self.list_branches("--remote", "origin/*")

571 if not branch.startswith("origin/HEAD")

572 ]

573

574 def update_remote(self) -> None:

575 """Update remote repository."""

576 auth_args = self.get_auth_args()

577 self.execute([*auth_args, "remote", "prune", "origin"])

578 if self.list_remote_branches():

579 # Updating existing fork

580 self.execute([*auth_args, "fetch", "origin"])

581 else:

582 # Doing initial fetch

583 try:

584 self.execute([*auth_args, "fetch", "origin", *self.get_depth()])

585 except RepositoryError as error:

586 if error.retcode == 1 and not error.args[0]:

587 # Fetch with --depth fails on blank repo

588 self.execute([*auth_args, "fetch", "origin"])

589 else:

590 raise

591

592 self.clean_revision_cache()

593

594 def push(self, branch: str) -> None:

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

596 refspec = f"{self.branch}:{branch}" if branch else self.branch

597 self.execute([*self._cmd_push, "origin", refspec])

598

599 def unshallow(self) -> None:

600 self.execute([*self.get_auth_args(), "fetch", "--unshallow"])

601

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

603 """Parse output with changed files."""

604 # Strip action prefix we do not use

605 for line in lines:

606 yield from line.split("\t")[1:]

607

608 def status(self) -> str:

609 result = [super().status()]

610 cleanups = self.execute(["clean", "-f", "-d", "-n"], needs_lock=False)

611 if cleanups: 611 ↛ 612line 611 didn't jump to line 612 because the condition on line 611 was never true

612 result.extend(("", gettext("Possible cleanups:"), "", cleanups))

613

614 return "\n".join(result)

615

616 def compact(self) -> None:

617 self.execute(["gc"])

618

619

620class GitWithGerritRepository(GitRepository):

621 name: ClassVar[StrOrPromise] = "Gerrit"

622 req_version: ClassVar[str] = "1.27.0"

623 push_label: ClassVar[StrOrPromise] = gettext_lazy(

624 "This will push changes to Gerrit for a review."

625 )

626 pushes_to_different_location: ClassVar[bool] = True

627

628 _version: ClassVar[str | None] = None

629

630 @classmethod

631 def _get_version(cls):

632 """Return VCS program version."""

633 return cls._popen(["review", "--version"], merge_err=True).split()[-1]

634

635 def get_username_from_url(self, url) -> str:

636 if url is not None:

637 if url.startswith("git@"):

638 return url.split(":")[-1].split("/")[0]

639 if url.startswith(("ssh://", "https://")) and "@" in url:

640 return url.split("//")[-1].split("@")[0]

641 if url.startswith(("ssh://", "https://")):

642 return url.split("//")[-1].split("/")[1]

643 return ""

644 return ""

645

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

647 if self.needs_push():

648 try:

649 self.execute(["review", "--yes", self.branch])

650 except RepositoryError as error:

651 if "(no new changes)" in str(error):

652 return

653 raise

654

655 def get_remote_configure(

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

657 ) -> tuple[tuple[str, str, str | None], ...]:

658 if push_url:

659 gerrit_user = self.get_username_from_url(push_url)

660 return (

661 ('remote "gerrit"', "url", push_url),

662 ("gitreview", "username", gerrit_user),

663 )

664 return (('remote "gerrit"', "url", None),)

665

666

667class SubversionRepository(GitRepository):

668 name: ClassVar[StrOrPromise] = "Subversion"

669 default_branch: ClassVar[str] = "master"

670 push_label: ClassVar[StrOrPromise] = gettext_lazy(

671 "This will commit changes to the Subversion repository."

672 )

673

674 _version: ClassVar[str | None] = None

675

676 needs_push_url: ClassVar[bool] = False

677

678 def __init__(

679 self,

680 path: str,

681 *,

682 branch: str | None = None,

683 component: Component | None = None,

684 local: bool = False,

685 skip_init: bool = False,

686 ) -> None:

687 super().__init__(

688 path, branch=branch, component=component, local=local, skip_init=skip_init

689 )

690 self._fetch_revision: str | None = None

691

692 @classmethod

693 def global_setup(cls) -> None:

694 """Perform global settings."""

695 dirname = os.path.join(data_dir("home"), ".subversion")

696 filename = os.path.join(dirname, "config")

697 if not os.path.exists(dirname): 697 ↛ 699line 697 didn't jump to line 699 because the condition on line 697 was always true

698 os.makedirs(dirname)

699 config = RawConfigParser()

700 config.read(filename)

701 section = "auth"

702 option = "password-stores"

703 value = ""

704 if not config.has_section(section): 704 ↛ 706line 704 didn't jump to line 706 because the condition on line 704 was always true

705 config.add_section(section)

706 if config.has_option(section, option) and config.get(section, option) == value: 706 ↛ 707line 706 didn't jump to line 707 because the condition on line 706 was never true

707 return

708 config.set(section, option, value)

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

710 config.write(handle)

711

712 @classmethod

713 def _get_version(cls):

714 """Return VCS program version."""

715 return cls._popen(["svn", "--version"], merge_err=False).split()[2]

716

717 @classmethod

718 def is_stdlayout(cls, url):

719 output = cls._popen(["svn", "ls", url], fullcmd=True).splitlines()

720 return "trunk/" in output

721

722 @classmethod

723 def get_last_repo_revision(cls, url):

724 output = cls._popen(

725 ["svn", "log", url, "--limit=1", "--xml"],

726 fullcmd=True,

727 raw=True,

728 merge_err=False,

729 )

730 tree = parse_xml(output)

731 entry = tree.find("logentry")

732 if entry is not None:

733 return entry.get("revision")

734 return None

735

736 @classmethod

737 def get_remote_args(cls, source, target):

738 result = ["--prefix=origin/", "--", source, target]

739 if cls.is_stdlayout(source): 739 ↛ anywhereline 739 didn't jump anywhere: it always raised an exception.

740 result.insert(0, "--stdlayout")

741 revision = cls.get_last_repo_revision(source + "/trunk/")

742 else:

743 revision = cls.get_last_repo_revision(source)

744 if revision:

745 revision = f"--revision={revision}:HEAD"

746

747 return result, revision

748

749 def configure_remote(

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

751 ) -> None:

752 """

753 Initialize the git-svn repository.

754

755 This does not support switching remote as it's quite complex:

756 https://git.wiki.kernel.org/index.php/GitSvnSwitch

757

758 The git svn init errors in case the URL is not matching.

759 """

760 try:

761 existing = self.get_config("svn-remote.svn.url")

762 except RepositoryError:

763 existing = None

764 if existing: 764 ↛ 766line 764 didn't jump to line 766 because the condition on line 764 was never true

765 # The URL is root of the repository, while we get full path

766 if not pull_url.startswith(existing):

767 raise RepositoryError(-1, "Can not switch subversion URL")

768 return

769 args, self._fetch_revision = self.get_remote_args(pull_url, self.path)

770 self.execute(["svn", "init", *args])

771

772 def update_remote(self) -> None:

773 """Update remote repository."""

774 if self._fetch_revision:

775 self.execute(["svn", "fetch", self._fetch_revision])

776 self._fetch_revision = None

777 else:

778 self.execute(["svn", "fetch", "--parent"])

779 self.clean_revision_cache()

780

781 @classmethod

782 def _clone(

783 cls,

784 source: str,

785 target: str,

786 branch: str, # noqa: ARG003

787 ) -> None:

788 """Clone svn repository with git-svn."""

789 args, revision = cls.get_remote_args(source, target)

790 if revision:

791 args.insert(0, revision)

792 cls._popen(["svn", "clone", *args])

793

794 def merge(

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

796 ) -> None:

797 """

798 Rebases.

799

800 Git-svn does not support merge.

801 """

802 self.rebase(abort)

803 self.clean_revision_cache()

804

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

806 """

807 Rebase remote branch or reverts the rebase.

808

809 Git-svn does not support merge.

810 """

811 if abort:

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

813 else:

814 self.execute(["svn", "rebase"])

815 self.clean_revision_cache()

816

817 @cached_property

818 def last_remote_revision(self) -> str:

819 """Return last remote revision."""

820 return self.execute(

821 ["log", "-n", "1", "--format=format:%H", self.get_remote_branch_name()],

822 needs_lock=False,

823 merge_err=False,

824 )

825

826 def get_remote_branch_name(self, branch: str | None = None) -> str:

827 """

828 Return the remote branch name.

829

830 trunk if local branch is master, local branch otherwise.

831 """

832 if branch is None:

833 branch = self.branch

834 if branch == self.default_branch:

835 fetch = self.get_config("svn-remote.svn.fetch")

836 if "origin/trunk" in fetch:

837 return "origin/trunk"

838 if "origin/git-svn" in fetch:

839 return "origin/git-svn"

840 return f"origin/{branch}"

841

842 def list_remote_branches(self) -> list[str]:

843 return []

844

845 def push(self, branch: str) -> None:

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

847 self.execute(["svn", "dcommit", self.branch])

848

849

850class GitForcePushRepository(GitRepository):

851 name: ClassVar[StrOrPromise] = gettext_lazy("Git with force push")

852 _cmd_push: ClassVar[list[str]] = ["push", "--force"]

853 identifier: ClassVar[str] = "git-force-push"

854 push_label: ClassVar[StrOrPromise] = gettext_lazy(

855 "This will force push changes to the upstream repository."

856 )

857

858

859class GitMergeRequestBase(GitForcePushRepository):

860 needs_push_url: ClassVar[bool] = False

861 pushes_to_different_location: ClassVar[bool] = True

862 identifier: ClassVar[str]

863 API_TEMPLATE: ClassVar[str]

864 REQUIRED_CONFIG: ClassVar[set[str]] = {"username", "token"}

865 OPTIONAL_CONFIG: ClassVar[set[str]] = {"scheme"}

866

867 def merge(

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

869 ) -> None:

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

871 # This reverts merge behavior of pure git backend

872 # as we're expecting there will be an additional merge

873 # commit created from the merge request.

874 if abort:

875 self.execute(["merge", "--abort"])

876 # Needed for compatibility with original merge code

877 self.execute(["checkout", self.branch])

878 else:

879 cmd = ["merge"]

880 if no_ff:

881 cmd.append("--no-ff")

882 cmd.extend(self.get_gpg_sign_args())

883 cmd.append(self.get_remote_branch_name())

884 self.execute(cmd)

885 self.clean_revision_cache()

886

887 def parse_repo_url(

888 self, repo: str | None = None

889 ) -> tuple[str | None, str | None, str | None, str, str, str]:

890 if repo is None:

891 repo = self.component.push or self.component.repo

892 parsed = urlparse(repo)

893 host = parsed.hostname

894 scheme: str | None = parsed.scheme

895 username: str | None = parsed.username

896 password: str | None = parsed.password

897 if not host:

898 if ":" in repo:

899 # Assume SSH URL

900 host, path = repo.split(":")

901 host = host.split("@")[-1]

902 scheme = None

903 else:

904 host = repo

905 path = ""

906 scheme = None

907 else:

908 path = parsed.path

909 parts = path.split(":")[-1].rstrip("/").split("/")

910 last_part = parts[-1]

911 last_part = last_part.removesuffix(".git")

912 slug_parts = [last_part]

913 owner = ""

914 for part in parts[:-1]:

915 if not part:

916 continue

917 if not owner:

918 owner = part

919 continue

920 slug_parts.insert(-1, part)

921 slug = "/".join(slug_parts)

922 return (scheme, username, password, host, owner, slug)

923

924 def format_url(

925 self, scheme: str, hostname: str, owner: str, slug: str, **extra: str

926 ) -> str:

927 return self.API_TEMPLATE.format(

928 host=hostname,

929 owner=owner,

930 slug=slug,

931 owner_url=urllib.parse.quote_plus(owner),

932 slug_url=urllib.parse.quote_plus(slug),

933 scheme=scheme,

934 **extra,

935 )

936

937 @classmethod

938 def validate_configuration(cls) -> list[str]:

939 if not cls.is_configured():

940 return []

941 result = []

942 credentials_name = cls.get_credentials_name()

943 for key, config in cls.get_credentials_configuration().items():

944 if not isinstance(key, str) or key.lower().startswith(

945 ("http://", "https://")

946 ):

947 replacement = (

948 key.lower().removeprefix("https://").removeprefix("http://")

949 )

950 result.append(

951 f'{credentials_name} should include only hostname, use "{replacement}" instead of "{key}"'

952 )

953 result.extend(

954 f"{credentials_name}[{key}]: Missing required configuration: {current}"

955 for current in cls.REQUIRED_CONFIG

956 if current not in config

957 )

958 extra = set(config) - cls.OPTIONAL_CONFIG - cls.REQUIRED_CONFIG

959 if extra:

960 result.append(

961 f"{credentials_name}[{key}]: Unsupported configuration: {','.join(extra)}"

962 )

963 return result

964

965 @classmethod

966 def get_credentials_name(cls) -> str:

967 return f"{cls.identifier.upper()}_CREDENTIALS"

968

969 @classmethod

970 def get_credentials_configuration(cls):

971 return getattr(settings, cls.get_credentials_name())

972

973 def get_credentials(self) -> GitCredentials:

974 scheme, username, password, host, owner, slug = self.parse_repo_url()

975 hostname = self.format_api_host(host).lower()

976 credentials = self.get_credentials_by_hostname(hostname)

977

978 # Scheme override

979 if "scheme" in credentials:

980 scheme = credentials["scheme"]

981

982 # Fallback to https

983 if not scheme or scheme == "ssh":

984 scheme = "https"

985

986 # Credentials from URL

987 push_scheme = scheme

988 if not username or not password:

989 push_scheme = "ssh"

990 username = credentials["username"]

991 password = credentials["token"]

992

993 return {

994 "url": self.format_url(scheme, hostname, owner, slug),

995 "owner": owner,

996 "slug": slug,

997 "hostname": hostname,

998 "username": username,

999 "token": password,

1000 "scheme": scheme,

1001 "push_scheme": push_scheme,

1002 }

1003

1004 def get_credentials_by_hostname(self, hostname: str) -> dict[str, str]:

1005 configuration = self.get_credentials_configuration()

1006 try:

1007 credentials = configuration[hostname]

1008 except KeyError as exc:

1009 raise RepositoryError(

1010 0, f"{self.name} API access for {hostname} is not configured"

1011 ) from exc

1012 return credentials

1013

1014 @classmethod

1015 def is_configured(cls) -> bool:

1016 return bool(cls.get_credentials_configuration())

1017

1018 def push_to_fork(

1019 self, credentials: GitCredentials, local_branch: str, fork_branch: str

1020 ) -> None:

1021 """Push given local branch to branch in forked repository."""

1022 self.execute(

1023 [

1024 "push",

1025 "--force",

1026 credentials["username"],

1027 f"{local_branch}:{fork_branch}",

1028 ]

1029 )

1030

1031 def configure_fork_remote(

1032 self, ssh_url: str, http_url: str, credentials: GitCredentials

1033 ) -> None:

1034 """Configure fork remote repository."""

1035 remote_name = credentials["username"]

1036 if credentials["push_scheme"] == "ssh":

1037 push_url = ssh_url

1038 else:

1039 push_url = self.authenticate_url(http_url, credentials)

1040 self.log(

1041 f"Configuring fork remote '{remote_name}': {push_url}", level=logging.INFO

1042 )

1043 self.config_update(

1044 # Push url

1045 (f'remote "{remote_name}"', "pushurl", push_url),

1046 )

1047

1048 def should_use_fork(self, branch: str | None = None) -> bool:

1049 return not branch or branch == self.branch

1050

1051 def get_remote_branch_name(self, branch: str | None = None) -> str:

1052 remote = "origin"

1053 if branch is not None and self.should_use_fork(branch):

1054 credentials = self.get_credentials()

1055 remote = credentials["username"]

1056 return f"{remote}/{self.branch if branch is None else branch}"

1057

1058 def fork(self, credentials: GitCredentials) -> None:

1059 """Create fork of original repository if one doesn't exist yet."""

1060 remotes = self.execute(["remote"]).splitlines()

1061 if credentials["username"] not in remotes:

1062 self.create_fork(credentials)

1063

1064 def push(self, branch: str) -> None:

1065 """

1066 Fork repository on GitHub and push changes.

1067

1068 Pushes changes to *-weblate branch on fork and creates pull request against

1069 original repository.

1070 """

1071 credentials = self.get_credentials()

1072 if not self.should_use_fork(branch):

1073 fork_remote = "origin"

1074 fork_branch = branch

1075 super().push(branch)

1076 else:

1077 fork_remote = credentials["username"]

1078 self.fork(credentials)

1079 if self.component is not None:

1080 fork_branch = (

1081 f"weblate-{self.component.project.slug}-{self.component.slug}"

1082 )

1083 else:

1084 fork_branch = f"weblate-{self.branch}"

1085 self.push_to_fork(credentials, self.branch, fork_branch)

1086 self.create_pull_request(credentials, self.branch, fork_remote, fork_branch)

1087

1088 def authenticate_url(self, url: str, credentials: GitCredentials) -> str:

1089 """Inject credentials into URL."""

1090 parsed_url = urlparse(url)

1091 return urlunparse(

1092 (

1093 parsed_url[0],

1094 "{}:{}@{}".format(

1095 urllib.parse.quote(credentials["username"], safe=""),

1096 urllib.parse.quote(credentials["token"], safe=""),

1097 parsed_url[1],

1098 ),

1099 *parsed_url[2:],

1100 )

1101 )

1102

1103 def create_fork(self, credentials: GitCredentials) -> None:

1104 raise NotImplementedError

1105

1106 def get_fork_failed_message(

1107 self, error: str, credentials: GitCredentials, response: requests.Response

1108 ) -> str:

1109 hostname = credentials["hostname"]

1110 username = credentials["username"]

1111 try:

1112 data = response.json()

1113 except JSONDecodeError:

1114 data = response.text

1115 self.log(

1116 f"Creating fork via {response.url} failed ({response.status_code}): {data!r}",

1117 level=logging.WARNING,

1118 )

1119 if response.status_code == 404:

1120 error = f"Repository not found. Check whether exists and user '{username}' has access to it."

1121 if error.strip():

1122 return f"Could not fork repository at {hostname}: {error}"

1123 return f"Could not fork repository at {hostname}"

1124

1125 def create_pull_request(

1126 self,

1127 credentials: GitCredentials,

1128 origin_branch: str,

1129 fork_remote: str,

1130 fork_branch: str,

1131 ) -> None:

1132 raise NotImplementedError

1133

1134 def get_merge_message(self):

1135 lines = render_template(

1136 self.component.pull_message.strip(), component=self.component

1137 ).splitlines()

1138 return lines[0], "\n".join(lines[1:]).strip()

1139

1140 def format_api_host(self, host):

1141 return host

1142

1143 def get_headers(self, credentials: GitCredentials) -> dict[str, str]:

1144 return {

1145 "Accept": "application/json",

1146 "Authorization": f"token {credentials['token']}",

1147 }

1148

1149 def get_auth(

1150 self, credentials: GitCredentials

1151 ) -> tuple[str, str] | AuthBase | None:

1152 return None

1153

1154 def get_error_message(self, response_data: dict) -> str:

1155 """

1156 Log and parse all errors.

1157

1158 Sometimes GitHub returns the error

1159 messages in an errors list instead of the message. Sometimes, there

1160 is no errors list. Hence the different logics.

1161 """

1162 if not isinstance(response_data, dict):

1163 return ""

1164 errors = []

1165 for extra in ("message", "error", "error_description", "errors"):

1166 messages = response_data.get(extra)

1167 if messages is None:

1168 continue

1169 if isinstance(messages, str):

1170 errors.append(messages)

1171 elif isinstance(messages, list):

1172 for error in messages:

1173 if isinstance(error, str):

1174 line = error

1175 else:

1176 line = error.get("message", str(error))

1177 errors.append(line)

1178 elif isinstance(messages, dict):

1179 errors.extend(f"{key}: {error}" for key, error in messages.items())

1180 else:

1181 self.log(

1182 f"failed to parse HTTP response message: {messages!r}",

1183 level=logging.WARNING,

1184 )

1185

1186 if errors:

1187 for error in errors:

1188 self.log(error, level=logging.DEBUG)

1189

1190 return ", ".join(errors)

1191

1192 def should_retry(self, response, response_data) -> bool:

1193 retry_after = response.headers.get("Retry-After")

1194 if retry_after and retry_after.isdigit():

1195 # Cap sleeping to 60 seconds

1196 self.set_next_request_time(

1197 min(int(retry_after), 6 * max(settings.VCS_API_DELAY, 10))

1198 )

1199 return True

1200 return False

1201

1202 @cached_property

1203 def request_time_cache_key(self) -> str:

1204 vcs_id = self.get_identifier()

1205 return f"vcs:request-time:{vcs_id}"

1206

1207 def set_next_request_time(self, delay: int) -> None:

1208 cache.set(self.request_time_cache_key, time() + delay)

1209

1210 def request(

1211 self,

1212 method: str,

1213 credentials: GitCredentials,

1214 url: str,

1215 *,

1216 data: dict | None = None,

1217 params: dict | None = None,

1218 json: dict | None = None,

1219 retry: int = 0,

1220 ) -> tuple[dict, requests.Response, str]:

1221 do_retry = False

1222 vcs_id = self.get_identifier()

1223 self.log(f"HTTP {method} {url}")

1224 cache_id = self.request_time_cache_key

1225 lock = WeblateLock(

1226 lock_path=data_dir("home"),

1227 scope=f"vcs:api:{vcs_id}",

1228 key=0,

1229 slug=vcs_id,

1230 timeout=3 * max(settings.VCS_API_DELAY, 10),

1231 )

1232 try:

1233 with lock:

1234 next_api_time = cache.get(cache_id)

1235 now = time()

1236 if next_api_time is not None and now < next_api_time:

1237 with sentry_sdk.start_span(op="vcs.api_sleep", name=vcs_id):

1238 sleep(next_api_time - now)

1239 try:

1240 response = requests.request(

1241 method,

1242 url,

1243 headers=self.get_headers(credentials),

1244 data=data,

1245 params=params,

1246 json=json,

1247 auth=self.get_auth(credentials),

1248 timeout=5,

1249 )

1250 except (OSError, HTTPError) as error:

1251 report_error("Git API request")

1252 raise RepositoryError(0, str(error)) from error

1253

1254 # GitHub recommends a delay between 2 requests of at least 1s,

1255 # but in reality this hits secondary rate limits.

1256 self.set_next_request_time(settings.VCS_API_DELAY)

1257

1258 self.add_response_breadcrumb(response)

1259 try:

1260 response_data = response.json()

1261 except JSONDecodeError as error:

1262 report_error("GIT API request json decoding")

1263 self.raise_for_response(response)

1264 raise RepositoryError(0, str(error)) from error

1265

1266 if self.should_retry(response, response_data):

1267 do_retry = True

1268 except WeblateLockTimeoutError:

1269 do_retry = True

1270

1271 if do_retry:

1272 retry += 1

1273 if retry > 10:

1274 raise RepositoryError(0, "Too many retries")

1275 return self.request(

1276 method,

1277 credentials,

1278 url,

1279 data=data,

1280 params=params,

1281 json=json,

1282 retry=retry,

1283 )

1284

1285 return response_data, response, self.get_error_message(response_data)

1286

1287 def failed_pull_request(

1288 self, error: str, pr_url: str, response: requests.Response, data: dict

1289 ) -> NoReturn:

1290 status_code = response.status_code

1291 self.log(

1292 f"Creating pull request via {pr_url} failed ({status_code}): {data!r}",

1293 level=logging.WARNING,

1294 )

1295 report_error("Could not create pull request", message=True)

1296 raise RepositoryError(

1297 -1, f"Could not create pull request {status_code}: {error}"

1298 )

1299

1300 @classmethod

1301 def raise_for_response(cls, response: requests.Response) -> None:

1302 """

1303 Validate response status code.

1304

1305 Raises :class:`HTTPError`, if one occurred.

1306

1307 Some providers (Azure DevOps for instance) respond with codes in the 2XX range

1308 even though the response was an error. This method exists to let the

1309 inheritors override it if they have special cases.

1310 """

1311 response.raise_for_status()

1312

1313

1314class AzureDevOpsRepository(GitMergeRequestBase):

1315 name: ClassVar[StrOrPromise] = gettext_lazy("Azure DevOps pull request")

1316 identifier: ClassVar[str] = "azure_devops"

1317 _version: ClassVar[str | None] = None

1318 API_TEMPLATE: ClassVar[str] = (

1319 "{scheme}://{host}/{owner}/_apis/git/repositories/{slug}"

1320 )

1321 ORG_API_TEMPLATE: ClassVar[str] = (

1322 "{scheme}://{host}/{org}/_apis/Contribution/HierarchyQuery?api-version=5.0-preview.1"

1323 )

1324 REQUIRED_CONFIG: ClassVar[set[str]] = {"username", "token", "organization"}

1325 OPTIONAL_CONFIG: ClassVar[set[str]] = {"scheme", "workItemIds"}

1326 push_label: ClassVar[StrOrPromise] = gettext_lazy(

1327 "This will push changes and create a Azure DevOps pull request."

1328 )

1329

1330 @classmethod

1331 def raise_for_response(cls, response: requests.Response) -> None:

1332 super().raise_for_response(response)

1333

1334 # Azure DevOps returns 203 when the token is invalid

1335 if response.status_code == 203:

1336 raise RepositoryError(0, "Invalid token")

1337

1338 def fork(self, credentials: GitCredentials) -> None:

1339 remotes = self.execute(["remote"]).splitlines()

1340 if credentials["username"] not in remotes:

1341 self.create_fork(credentials)

1342 return

1343

1344 # If the fork was deleted, we just create another fork

1345 try:

1346 self.__get_forked_id(credentials, credentials["username"])

1347 except RepositoryError:

1348 self.create_fork(credentials)

1349

1350 def parse_repo_url(

1351 self, repo: str | None = None

1352 ) -> tuple[str | None, str | None, str | None, str, str, str]:

1353 if repo is None:

1354 repo = self.component.repo

1355

1356 scheme_regex = r"^[a-z]+:\/\/.*" # matches for example ssh://* and https://*

1357

1358 if not re.match(scheme_regex, repo):

1359 repo = f"ssh://{repo}" # assume all links without schema are ssh links

1360

1361 (scheme, username, password, host, owner, slug) = super().parse_repo_url(repo)

1362

1363 # ssh links are in a subdomain, the API link doesn't have that so remove it

1364 host = host.removeprefix("ssh.")

1365

1366 # https urls have /_git/ between owner and slug

1367 if "/_git/" in slug:

1368 parts = slug.split("/_git/")

1369 owner = owner + "/" + parts[0] # we want owner to be org/project

1370 slug = parts[1]

1371

1372 else:

1373 slug = slug.removeprefix("_git/")

1374 if "/" in slug:

1375 parts = slug.split("/")

1376

1377 owner = owner + "/" + parts[0] # we want owner to be org/project

1378 slug = parts[1]

1379

1380 return (scheme, username, password, host, owner, slug)

1381

1382 def get_headers(self, credentials: GitCredentials) -> dict[str, str]:

1383 headers = super().get_headers(credentials)

1384 headers["Accept"] = "application/json; api-version=7.0"

1385 return headers

1386

1387 def get_auth(

1388 self, credentials: GitCredentials

1389 ) -> tuple[str, str] | AuthBase | None:

1390 return ("", credentials["token"])

1391

1392 def create_fork(self, credentials: GitCredentials) -> None:

1393 # url without repository name

1394 fork_url = "/".join(list(credentials["url"].split("/")[0:-1]))

1395

1396 # Get parent repo info

1397 response_data, response, error = self.request(

1398 "get", credentials, credentials["url"]

1399 )

1400

1401 if "project" not in response_data:

1402 report_error("Could not fork repository", message=True)

1403 raise RepositoryError(

1404 0, self.get_fork_failed_message(error, credentials, response)

1405 )

1406

1407 found_fork = self.__find_fork(credentials)

1408

1409 if found_fork is not None:

1410 self.configure_fork_remote(

1411 found_fork["sshUrl"], found_fork["remoteUrl"], credentials

1412 )

1413 return

1414

1415 request = {

1416 "name": credentials["slug"],

1417 "project": {"id": response_data["project"]["id"]},

1418 "parentRepository": {

1419 "id": response_data["id"],

1420 "project": {"id": response_data["project"]["id"]},

1421 },

1422 }

1423

1424 # Create fork

1425 response_data, response, error = self.request(

1426 "post", credentials, fork_url, json=request

1427 )

1428

1429 if "TF400948" in error: # A Git repository with the name already exists

1430 fork_name = "{}-{}".format(

1431 credentials["slug"],

1432 random.randint(1000, 9999), # noqa: S311

1433 )

1434

1435 request["name"] = fork_name

1436 response_data, response, error = self.request(

1437 "post", credentials, fork_url, json=request

1438 )

1439

1440 if "sshUrl" not in response_data or "remoteUrl" not in response_data:

1441 report_error("Could not fork repository", message=True)

1442 raise RepositoryError(

1443 0, self.get_fork_failed_message(error, credentials, response)

1444 )

1445

1446 self.configure_fork_remote(

1447 response_data["sshUrl"], response_data["remoteUrl"], credentials

1448 )

1449

1450 def get_credentials(self) -> GitCredentials:

1451 super_credentials = super().get_credentials()

1452 hostname = super_credentials["hostname"]

1453 credentials = self.get_credentials_by_hostname(hostname)

1454

1455 super_credentials["organization"] = credentials["organization"]

1456 super_credentials["workItemIds"] = cast(

1457 "list[str]", credentials.get("workItemIds", [])

1458 )

1459

1460 return super_credentials

1461

1462 def create_pull_request(

1463 self,

1464 credentials: GitCredentials,

1465 origin_branch: str,

1466 fork_remote: str,

1467 fork_branch: str,

1468 retry_fork: bool = True,

1469 ) -> None:

1470 pr_url = "{}/pullrequests".format(credentials["url"])

1471 title, description = self.get_merge_message()

1472

1473 work_item_ids = self.get_credentials()["workItemIds"]

1474 work_item_refs = [{"id": str(ref)} for ref in work_item_ids]

1475

1476 request = {

1477 "sourceRefName": "refs/heads/" + fork_branch,

1478 "targetRefName": "refs/heads/" + origin_branch,

1479 "title": title,

1480 "description": description,

1481 "workItemRefs": work_item_refs,

1482 }

1483

1484 if fork_remote != "origin":

1485 forked_id = self.__get_forked_id(credentials, fork_remote)

1486 request["forkSource"] = {"repository": {"id": forked_id}}

1487

1488 response_data, response, error_message = self.request(

1489 "post", credentials, pr_url, json=request

1490 )

1491

1492 # Check for an error. If the error has a message saying A pull request already

1493 # exists, then we ignore that, else raise an error.

1494 if "url" not in response_data:

1495 # Gracefully handle pull request already exists

1496 if "TF401179" in error_message:

1497 return

1498

1499 self.failed_pull_request(error_message, pr_url, response, response_data)

1500

1501 def __get_forked_id(self, credentials: GitCredentials, remote: str) -> str:

1502 """

1503 Return ID of the forked Azure DevOps repository.

1504

1505 To send a PR to Azure DevOps via API with a fork, one needs to send

1506 request a request with the ID of the forked repository (unlike others,

1507 where the name is enough).

1508 """

1509 cmd = ["remote", "get-url", "--push", remote]

1510 fork_remotes = self.execute(cmd, needs_lock=False, merge_err=False).splitlines()

1511 (_scheme, _username, _password, hostname, owner, slug) = self.parse_repo_url(

1512 fork_remotes[0]

1513 )

1514 url = self.format_url("https", hostname, owner, slug)

1515

1516 # Get repo info

1517 response_data, _response, error = self.request("get", credentials, url)

1518

1519 if "id" not in response_data:

1520 raise RepositoryError(0, error)

1521

1522 return response_data["id"]

1523

1524 def __find_fork(self, credentials) -> dict | None:

1525 for fork in self.__get_forks(credentials):

1526 if "project" in fork and fork["project"]["name"] == credentials["username"]:

1527 return fork

1528

1529 return None

1530

1531 def __get_forks(self, credentials: GitCredentials) -> list:

1532 forks_url = "{}/forks/{}".format(credentials["url"], self.__get_org_id())

1533 response_data, response, error = self.request("get", credentials, forks_url)

1534

1535 if response.status_code != 200:

1536 report_error("Could not fork repository", message=True)

1537 raise RepositoryError(

1538 0, self.get_fork_failed_message(error, credentials, response)

1539 )

1540

1541 return response_data["value"]

1542

1543 def __get_org_id(self) -> str:

1544 credentials = self.get_credentials()

1545 org = credentials["owner"].split("/")[0] # format is "org/proj"

1546

1547 url = self.ORG_API_TEMPLATE.format(

1548 scheme=credentials["scheme"],

1549 host=credentials["hostname"],

1550 org=org,

1551 )

1552

1553 org_property = "ms.vss-features.my-organizations-data-provider"

1554 request = {

1555 "contributionIds": [org_property],

1556 "dataProviderContext": {"properties": {}},

1557 }

1558

1559 response_data, response, error_message = self.request(

1560 "post", credentials, url, json=request

1561 )

1562

1563 try:

1564 data_providers = response_data["dataProviders"]

1565 return data_providers[org_property]["organizations"][0]["id"]

1566 except (KeyError, IndexError) as error:

1567 report_error("Could not fork repository", message=True)

1568 raise RepositoryError(

1569 0, self.get_fork_failed_message(error_message, credentials, response)

1570 ) from error

1571

1572

1573class GithubRepository(GitMergeRequestBase):

1574 name: ClassVar[StrOrPromise] = gettext_lazy("GitHub pull request")

1575 identifier: ClassVar[str] = "github"

1576 _version: ClassVar[str | None] = None

1577 API_TEMPLATE: ClassVar[str] = "{scheme}://{host}/{suffix}repos/{owner}/{slug}"

1578 push_label: ClassVar[StrOrPromise] = gettext_lazy(

1579 "This will push changes and create a GitHub pull request."

1580 )

1581

1582 def format_api_host(self, host):

1583 if host == "github.com":

1584 return "api.github.com"

1585 return host

1586

1587 def format_url(

1588 self, scheme: str, hostname: str, owner: str, slug: str, **extra: str

1589 ) -> str:

1590 suffix = ""

1591 # In case the hostname of the repository does not point to "github.com" assume

1592 # that it is on a GitHub Enterprise server, which has uses a different base URL

1593 # for the API:

1594 if hostname != "api.github.com":

1595 suffix = "api/v3/"

1596 return super().format_url(scheme, hostname, owner, slug, suffix=suffix, **extra)

1597

1598 def get_headers(self, credentials: GitCredentials) -> dict[str, str]:

1599 headers = super().get_headers(credentials)

1600 headers["Accept"] = "application/vnd.github.v3+json"

1601 return headers

1602

1603 def should_retry(self, response, response_data) -> bool:

1604 if super().should_retry(response, response_data):

1605 return True

1606 # https://docs.github.com/rest/overview/resources-in-the-rest-api#secondary-rate-limits

1607 if (

1608 "message" in response_data

1609 and "You have exceeded a secondary rate limit" in response_data["message"]

1610 ):

1611 self.set_next_request_time(6 * max(settings.VCS_API_DELAY, 10))

1612 return True

1613 return False

1614

1615 def create_fork(self, credentials: GitCredentials) -> None:

1616 fork_url = "{}/forks".format(credentials["url"])

1617

1618 # GitHub API returns the entire data of the fork, in case the fork

1619 # already exists. Hence, this is perfectly handled, if the fork already

1620 # exists in the remote side.

1621 response_data, response, error = self.request("post", credentials, fork_url)

1622 if "ssh_url" not in response_data:

1623 report_error("Could not fork repository", message=True)

1624 raise RepositoryError(

1625 0, self.get_fork_failed_message(error, credentials, response)

1626 )

1627 self.configure_fork_remote(

1628 response_data["ssh_url"], response_data["clone_url"], credentials

1629 )

1630

1631 def create_pull_request(

1632 self,

1633 credentials: GitCredentials,

1634 origin_branch: str,

1635 fork_remote: str,

1636 fork_branch: str,

1637 retry_fork: bool = True,

1638 ) -> None:

1639 """

1640 Create pull request.

1641

1642 Use to merge branch in forked repository into branch of remote repository.

1643 """

1644 if fork_remote == "origin":

1645 head = fork_branch

1646 else:

1647 head = f"{fork_remote}:{fork_branch}"

1648 pr_url = "{}/pulls".format(credentials["url"])

1649 title, description = self.get_merge_message()

1650 request = {

1651 "head": head,

1652 "base": origin_branch,

1653 "title": title,

1654 "body": description,

1655 }

1656 response_data, response, error_message = self.request(

1657 "post", credentials, pr_url, json=request

1658 )

1659

1660 # Check for an error. If the error has a message saying A pull request already

1661 # exists, then we ignore that, else raise an error. Currently, since the API

1662 # doesn't return any other separate indication for a pull request existing

1663 # compared to other errors, checking message seems to be the only option

1664 if "url" not in response_data:

1665 # Gracefully handle pull request already exists or nothing to merge cases

1666 if (

1667 "A pull request already exists" in error_message

1668 or "No commits between " in error_message

1669 ):

1670 return

1671

1672 if "Validation Failed" in error_message:

1673 for error in response_data["errors"]:

1674 if error.get("field") == "head" and retry_fork:

1675 # This most likely indicates that Weblate repository has moved

1676 # and we should create a fresh fork.

1677 self.create_fork(credentials)

1678 self.create_pull_request(

1679 credentials,

1680 origin_branch,

1681 fork_remote,

1682 fork_branch,

1683 retry_fork=False,

1684 )

1685 return

1686

1687 self.failed_pull_request(error_message, pr_url, response, response_data)

1688

1689

1690class GiteaRepository(GitMergeRequestBase):

1691 name: ClassVar[StrOrPromise] = gettext_lazy("Gitea pull request")

1692 identifier: ClassVar[str] = "gitea"

1693 _version: ClassVar[str | None] = None

1694 API_TEMPLATE: ClassVar[str] = "{scheme}://{host}/api/v1/repos/{owner}/{slug}"

1695 push_label: ClassVar[StrOrPromise] = gettext_lazy(

1696 "This will push changes and create a Gitea pull request."

1697 )

1698

1699 def create_fork(self, credentials: GitCredentials) -> None:

1700 fork_url = "{}/forks".format(credentials["url"])

1701

1702 # Empty json body is required here, otherwise we'll get an

1703 # "Empty Content-Type" error from gitea.

1704 response_data, response, error = self.request(

1705 "post", credentials, fork_url, json={}

1706 )

1707 if (

1708 "message" in response_data

1709 and "repository is already forked by user" in error

1710 ) or response.status_code == 409:

1711 # we have to get the repository again if it is already forked

1712 response_data, response, error = self.request(

1713 "get", credentials, credentials["url"]

1714 )

1715 if "ssh_url" not in response_data:

1716 report_error("Could not fork repository", message=True)

1717 raise RepositoryError(

1718 0, self.get_fork_failed_message(error, credentials, response)

1719 )

1720 self.configure_fork_remote(

1721 response_data["ssh_url"], response_data["clone_url"], credentials

1722 )

1723

1724 def create_pull_request(

1725 self,

1726 credentials: GitCredentials,

1727 origin_branch: str,

1728 fork_remote: str,

1729 fork_branch: str,

1730 retry_fork: bool = True,

1731 ) -> None:

1732 """

1733 Create pull request.

1734

1735 Use to merge branch in forked repository into branch of remote repository.

1736 """

1737 if fork_remote == "origin":

1738 head = fork_branch

1739 else:

1740 head = f"{fork_remote}:{fork_branch}"

1741 pr_url = "{}/pulls".format(credentials["url"])

1742 title, description = self.get_merge_message()

1743 request = {

1744 "head": head,

1745 "base": origin_branch,

1746 "title": title,

1747 "body": description,

1748 }

1749 response_data, response, error_message = self.request(

1750 "post", credentials, pr_url, json=request

1751 )

1752

1753 # Check for an error. If the error has a message saying pull request already

1754 # exists, then we ignore that, else raise an error. Currently, since the API

1755 # doesn't return any other separate indication for a pull request existing

1756 # compared to other errors, checking message seems to be the only option

1757 if "url" not in response_data:

1758 # Gracefully handle pull request already exists

1759 if "pull request already exists for these targets" in error_message:

1760 return

1761

1762 self.failed_pull_request(error_message, pr_url, response, response_data)

1763

1764

1765class LocalRepository(GitRepository):

1766 """Local filesystem driver with no upstream repo."""

1767

1768 name = gettext_lazy("No remote repository")

1769 identifier = "local"

1770 default_branch = "main"

1771 supports_push = False

1772

1773 def configure_remote(

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

1775 ) -> None:

1776 return

1777

1778 def get_remote_branch_name(self, branch: str | None = None) -> str:

1779 return self.branch if branch is None else branch

1780

1781 def update_remote(self) -> None:

1782 return

1783

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

1785 return

1786

1787 def reset(self) -> None:

1788 return

1789

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

1791 return

1792

1793 def merge(

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

1795 ) -> None:

1796 return

1797

1798 def list_remote_branches(self):

1799 return []

1800

1801 @classmethod

1802 def get_remote_branch(cls, repo: str): # noqa: ARG003

1803 return cls.default_branch

1804

1805 @classmethod

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

1807 """Initialize the repository."""

1808 super().create_blank_repository(path)

1809 with open(os.path.join(path, "README.md"), "w") as handle:

1810 handle.write("Translations repository created by Weblate\n")

1811 handle.write("==========================================\n")

1812 handle.write("\n")

1813 handle.write("See https://weblate.org/ for more info.\n")

1814 cls._popen(["add", "README.md"], cwd=path)

1815 cls._popen(["commit", "--message", "Repository created by Weblate"], cwd=path)

1816

1817 @classmethod

1818 def _clone(

1819 cls,

1820 source: str, # noqa: ARG003

1821 target: str,

1822 branch: str, # noqa: ARG003

1823 ) -> None:

1824 if not os.path.exists(target): 1824 ↛ 1826line 1824 didn't jump to line 1826 because the condition on line 1824 was always true

1825 os.makedirs(target)

1826 cls.create_blank_repository(target)

1827

1828 @cached_property

1829 def last_remote_revision(self) -> str:

1830 return self.last_revision

1831

1832 @classmethod

1833 @contextmanager

1834 def build_local_repo(cls, target: str, commit_message: str):

1835 repo = cls(target)

1836 with repo.lock:

1837 # Create empty repo

1838 if os.path.exists(target): 1838 ↛ 1840line 1838 didn't jump to line 1840 because the condition on line 1838 was always true

1839 remove_tree(target)

1840 cls._clone("local:", target, cls.default_branch)

1841 # Populate files

1842 yield repo

1843 # Add to repository

1844 repo.execute(["add", target])

1845 if repo.needs_commit():

1846 repo.commit(commit_message)

1847

1848 @classmethod

1849 def from_zip(cls, target: str, zipfile: BinaryIO) -> Self:

1850 with cls.build_local_repo(target, "ZIP file uploaded into Weblate") as repo:

1851 # Extract zip file content, ignoring some files

1852 zipobj = ZipFile(zipfile)

1853 names = [name for name in zipobj.namelist() if not is_excluded(name)]

1854 zipobj.extractall(path=target, members=names)

1855 return repo

1856

1857 @classmethod

1858 def from_files(cls, target: str, files: dict[str, bytes]) -> Self:

1859 with cls.build_local_repo(target, "Started translation using Weblate") as repo:

1860 # Create files

1861 for name, content in files.items(): 1861 ↛ 1862line 1861 didn't jump to line 1862 because the loop on line 1861 never started

1862 fullname = os.path.join(target, name)

1863 dirname = os.path.dirname(fullname)

1864 if not os.path.exists(dirname):

1865 os.makedirs(dirname)

1866 with open(fullname, "wb") as handle:

1867 handle.write(content)

1868 return repo

1869

1870

1871class GitLabRepository(GitMergeRequestBase):

1872 name: ClassVar[StrOrPromise] = gettext_lazy("GitLab merge request")

1873 identifier: ClassVar[str] = "gitlab"

1874 _version: ClassVar[str | None] = None

1875 API_TEMPLATE: ClassVar[str] = (

1876 "{scheme}://{host}/api/v4/projects/{owner_url}%2F{slug_url}"

1877 )

1878 push_label: ClassVar[StrOrPromise] = gettext_lazy(

1879 "This will push changes and create a GitLab merge request."

1880 )

1881

1882 def get_fork_path(self, repo: str) -> str:

1883 _scheme, _username, _password, _host, owner, slug = self.parse_repo_url(repo)

1884 return urllib.parse.quote(f"{owner}/{slug}", safe="")

1885

1886 def get_forked_url(self, credentials: GitCredentials) -> str:

1887 """

1888 Return GitLab API URL for the forked repository.

1889

1890 To send a MR to GitLab via API, one needs to send request to the

1891 API URL of the forked repository along with the target project ID

1892 (unlike GitHub where the PR is sent to the target project's API URL).

1893 """

1894 target_path = credentials["url"].rsplit("/", 1)[-1]

1895 cmd = ["remote", "get-url", "--push", credentials["username"]]

1896 fork_remotes = self.execute(cmd, needs_lock=False, merge_err=False).splitlines()

1897 fork_path = self.get_fork_path(fork_remotes[0])

1898 return credentials["url"].replace(target_path, fork_path)

1899

1900 def get_headers(self, credentials: GitCredentials) -> dict[str, str]:

1901 headers = super().get_headers(credentials)

1902 headers["Authorization"] = f"Bearer {credentials['token']}"

1903 return headers

1904

1905 def get_target_project_id(self, credentials: GitCredentials):

1906 response_data, _response, error = self.request(

1907 "get", credentials, credentials["url"]

1908 )

1909 if "id" not in response_data:

1910 report_error("Could not get project", message=True)

1911 raise RepositoryError(0, f"Could not get project: {error}")

1912 return response_data["id"]

1913

1914 def configure_fork_features(

1915 self, credentials: GitCredentials, forked_url: str

1916 ) -> None:

1917 """

1918 Disable features in fork.

1919

1920 GitLab initializes a lot of the features in the fork

1921 that are not desirable, such as merge requests, issues, etc.

1922 This function is intended to disable all such features by

1923 editing the forked repo.

1924 """

1925 access_level_dict = {

1926 "issues_access_level": "disabled",

1927 "forking_access_level": "disabled",

1928 "builds_access_level": "enabled",

1929 "wiki_access_level": "disabled",

1930 "snippets_access_level": "disabled",

1931 "pages_access_level": "disabled",

1932 }

1933 response_data, _response, error = self.request(

1934 "put", credentials, forked_url, json=access_level_dict

1935 )

1936 if "web_url" not in response_data:

1937 report_error("Could not modify fork", message=True)

1938 raise RepositoryError(0, f"Could not modify fork {error}")

1939

1940 def create_fork(self, credentials: GitCredentials) -> None:

1941 get_fork_url = "{}/forks?owned=True".format(credentials["url"])

1942 fork_url = "{}/fork".format(credentials["url"])

1943 forked_repo = None

1944

1945 # Check if a fork already exists owned by the current user.

1946 # If fork already exists, set that fork as remote.

1947 # Else, create a new fork

1948 response_data, response, error = self.request("get", credentials, get_fork_url)

1949 for fork in response_data:

1950 # Since owned=True returns forks from both the user's repo and the forks

1951 # in all the groups owned by the user, hence we need the below logic

1952 # to find the fork within the user repo and not the groups

1953 if "owner" in fork and fork["owner"]["username"] == credentials["username"]:

1954 forked_repo = fork

1955

1956 if forked_repo is None:

1957 forked_repo, response, error = self.request("post", credentials, fork_url)

1958 # If a repo with the name of the fork already exist, append numeric

1959 # as suffix to name and path to use that as repo name and path.

1960 if (

1961 "ssh_url_to_repo" not in response_data

1962 and "has already been taken" in error

1963 ):

1964 fork_name = "{}-{}".format(

1965 credentials["url"].split("%2F")[-1],

1966 random.randint(1000, 9999), # noqa: S311

1967 )

1968 forked_repo, response, error = self.request(

1969 "post",

1970 credentials,

1971 fork_url,

1972 json={"name": fork_name, "path": fork_name},

1973 )

1974

1975 if (

1976 "ssh_url_to_repo" not in forked_repo

1977 or "http_url_to_repo" not in forked_repo

1978 ):

1979 report_error("Could not fork repository", message=True)

1980 raise RepositoryError(

1981 0, self.get_fork_failed_message(error, credentials, response)

1982 )

1983

1984 self.configure_fork_features(credentials, forked_repo["_links"]["self"])

1985

1986 self.configure_fork_remote(

1987 forked_repo["ssh_url_to_repo"], forked_repo["http_url_to_repo"], credentials

1988 )

1989

1990 def create_pull_request(

1991 self,

1992 credentials: GitCredentials,

1993 origin_branch: str,

1994 fork_remote: str,

1995 fork_branch: str,

1996 ) -> None:

1997 """

1998 Create pull request.

1999

2000 Use to merge branch in forked repository into branch of remote repository.

2001 """

2002 target_project_id = None

2003 pr_url = "{}/merge_requests".format(credentials["url"])

2004 if fork_remote != "origin":

2005 # GitLab MR works a little differently from GitHub. The MR needs

2006 # to be sent with the fork's API URL along with a parameter mentioning

2007 # the target project ID

2008 target_project_id = self.get_target_project_id(credentials)

2009 pr_url = f"{self.get_forked_url(credentials)}/merge_requests"

2010

2011 title, description = self.get_merge_message()

2012 request = {

2013 "source_branch": fork_branch,

2014 "target_branch": origin_branch,

2015 "title": title,

2016 "description": description,

2017 "target_project_id": target_project_id,

2018 }

2019 response_data, response, error = self.request(

2020 "post", credentials, pr_url, data=request

2021 )

2022

2023 if "web_url" not in response_data and response.status_code != 409:

2024 self.failed_pull_request(error, pr_url, response, response_data)

2025

2026

2027class PagureRepository(GitMergeRequestBase):

2028 name: ClassVar[StrOrPromise] = gettext_lazy("Pagure merge request")

2029 identifier: ClassVar[str] = "pagure"

2030 _version: ClassVar[str | None] = None

2031 API_TEMPLATE: ClassVar[str] = "{scheme}://{host}/api/0"

2032 push_label: ClassVar[StrOrPromise] = gettext_lazy(

2033 "This will push changes and create a Pagure merge request."

2034 )

2035

2036 def create_fork(self, credentials: GitCredentials) -> None:

2037 fork_url = "{}/fork".format(credentials["url"])

2038

2039 base_params = {

2040 "repo": credentials["slug"],

2041 "wait": True,

2042 }

2043 params: list[dict]

2044

2045 if credentials["owner"]:

2046 # We have no info on whether the URL part is namespace

2047 # or username, try both

2048 params = [

2049 {"namespace": credentials["owner"]},

2050 {"username": credentials["owner"]},

2051 ]

2052 else:

2053 params = [{}]

2054

2055 for param in params:

2056 param.update(base_params)

2057 _response_data, response, error = self.request(

2058 "post", credentials, fork_url, data=param

2059 )

2060 if '" cloned to "' in error or "already exists" in error:

2061 break

2062

2063 if '" cloned to "' not in error and "already exists" not in error:

2064 report_error("Could not fork repository", message=True)

2065 raise RepositoryError(

2066 0, self.get_fork_failed_message(error, credentials, response)

2067 )

2068

2069 self.configure_fork_remote(

2070 "ssh://git@{hostname}/forks/{username}/{slug}.git".format(**credentials),

2071 "https://{hostname}/forks/{username}/{slug}.git".format(**credentials),

2072 credentials,

2073 )

2074

2075 def create_pull_request(

2076 self,

2077 credentials: GitCredentials,

2078 origin_branch: str,

2079 fork_remote: str,

2080 fork_branch: str,

2081 ) -> None:

2082 """

2083 Create pull request.

2084

2085 Use to merge a branch in the forked repository

2086 into a branch of the remote repository.

2087 """

2088 if credentials["owner"]:

2089 pr_list_url = "{url}/{owner}/{slug}/pull-requests".format(**credentials)

2090 pr_create_url = "{url}/{owner}/{slug}/pull-request/new".format(

2091 **credentials

2092 )

2093 else:

2094 pr_list_url = "{url}/{slug}/pull-requests".format(**credentials)

2095 pr_create_url = "{url}/{slug}/pull-request/new".format(**credentials)

2096

2097 # List existing pull requests

2098 response_data, _response, error_message = self.request(

2099 "get", credentials, pr_list_url, params={"author": credentials["username"]}

2100 )

2101 if error_message:

2102 report_error("Pull request listing failed", message=True)

2103 raise RepositoryError(0, f"Pull request listing failed: {error_message}")

2104

2105 if response_data["total_requests"] > 0:

2106 # Open pull request from us is already there

2107 return

2108

2109 title, description = self.get_merge_message()

2110 request = {

2111 "branch_from": fork_branch,

2112 "branch_to": origin_branch,

2113 "title": title,

2114 "initial_comment": description,

2115 }

2116 if fork_remote != "origin":

2117 request["repo_from"] = credentials["slug"]

2118 request["repo_from_username"] = credentials["username"]

2119

2120 response_data, response, error_message = self.request(

2121 "post", credentials, pr_create_url, data=request

2122 )

2123

2124 if "id" not in response_data:

2125 self.failed_pull_request(

2126 error_message, pr_create_url, response, response_data

2127 )

2128

2129

2130class BitbucketServerRepository(GitMergeRequestBase):

2131 # Translators: Bitbucket Data Center is a product name, it differs from Bitbucked Cloud

2132 name: ClassVar[StrOrPromise] = gettext_lazy("Bitbucket Data Center pull request")

2133 identifier: ClassVar[str] = "bitbucketserver"

2134 _version: ClassVar[str | None] = None

2135 API_TEMPLATE: ClassVar[str] = (

2136 "{scheme}://{host}/rest/api/1.0/projects/{owner}/repos/{slug}"

2137 )

2138 push_label: ClassVar[StrOrPromise] = gettext_lazy(

2139 "This will push changes and create a Bitbucket Data Center pull request."

2140 )

2141

2142 def __init__(

2143 self,

2144 path: str,

2145 *,

2146 branch: str | None = None,

2147 component: Component | None = None,

2148 local: bool = False,

2149 skip_init: bool = False,

2150 ) -> None:

2151 super().__init__(

2152 path, branch=branch, component=component, local=local, skip_init=skip_init

2153 )

2154 self.bb_fork: dict = {}

2155

2156 def get_headers(self, credentials: GitCredentials) -> dict[str, str]:

2157 headers = super().get_headers(credentials)

2158 headers["Authorization"] = f"Bearer {credentials['token']}"

2159 return headers

2160

2161 def create_fork(self, credentials: GitCredentials) -> None:

2162 bb_fork, response, error_message = self.request(

2163 "post", credentials, credentials["url"], json={}

2164 )

2165 self.bb_fork = bb_fork

2166

2167 # If A fork already exists, get forks from origin and find the user's fork.

2168 # Since Bitbucket uses projectKey which can be any string (that is not

2169 # at all related to its name) we need to compare user's forks against

2170 # remote.

2171 if "This repository URL is already taken." in error_message:

2172 page = 0

2173 self.bb_fork = {}

2174 forks_url = f"{credentials['url']}/forks"

2175 while True:

2176 forks, response, error_message = self.request(

2177 "get", credentials, forks_url, params={"limit": 1000, "start": page}

2178 )

2179 if "values" in forks:

2180 for f in forks["values"]:

2181 fork_slug = f["origin"]["slug"]

2182 fork_project_key = f["origin"]["project"]["key"]

2183 if (

2184 fork_slug == credentials["slug"]

2185 and fork_project_key.upper() == credentials["owner"].upper()

2186 ):

2187 self.bb_fork = f

2188 break

2189

2190 if self.bb_fork or forks["isLastPage"] or not forks["values"]:

2191 break

2192

2193 page += 1

2194

2195 ssh_url = http_url = None

2196 if "links" in self.bb_fork:

2197 clone_urls = self.bb_fork["links"]["clone"]

2198 for link in clone_urls:

2199 if link["name"] == "ssh":

2200 ssh_url = link["href"]

2201 if link["name"] == "http":

2202 http_url = link["href"]

2203

2204 if not ssh_url or not http_url:

2205 report_error("Could not fork repository", message=True)

2206 raise RepositoryError(

2207 0, self.get_fork_failed_message(error_message, credentials, response)

2208 )

2209

2210 self.configure_fork_remote(ssh_url, http_url, credentials)

2211

2212 def get_default_reviewers(self, credentials: GitCredentials, fork_branch: str):

2213 target_repo, _response, error_message = self.request(

2214 "get", credentials, credentials["url"]

2215 )

2216 if "id" not in target_repo:

2217 return []

2218

2219 url = credentials["url"].replace("/rest/api/", "/rest/default-reviewers/")

2220 url = f"{url}/reviewers"

2221 params = {

2222 "targetRepoId": target_repo["id"],

2223 "sourceRepoId": self.bb_fork["id"],

2224 "targetRefId": self.branch,

2225 "sourceRefId": fork_branch,

2226 }

2227 users, _response, error_message = self.request(

2228 "get", credentials, url, params=params

2229 )

2230 if error_message or not users:

2231 return []

2232

2233 return [{"user": user} for user in users]

2234

2235 def create_pull_request(

2236 self,

2237 credentials: GitCredentials,

2238 origin_branch: str,

2239 fork_remote: str,

2240 fork_branch: str,

2241 ) -> None:

2242 # Make sure there's always a fork reference

2243 if not self.bb_fork:

2244 self.create_fork(credentials)

2245

2246 pr_url = f"{credentials['url']}/pull-requests"

2247 title, description = self.get_merge_message()

2248 request_body = {

2249 "title": title,

2250 "description": description,

2251 "toRef": {

2252 "id": f"refs/heads/{origin_branch}",

2253 "repository": {

2254 "slug": credentials["slug"],

2255 "project": {

2256 "key": credentials["owner"].upper(),

2257 },

2258 },

2259 },

2260 "fromRef": {

2261 "id": f"refs/heads/{fork_branch}",

2262 "repository": {

2263 "slug": self.bb_fork["slug"],

2264 "project": {"key": self.bb_fork["project"]["key"].upper()},

2265 },

2266 },

2267 "reviewers": self.get_default_reviewers(credentials, fork_branch),

2268 }

2269

2270 response_data, response, error_message = self.request(

2271 "post", credentials, pr_url, json=request_body

2272 )

2273

2274 """

2275 Bitbucket Data Center will return an error if a PR already exists.

2276 The push method in the parent class pushes changes to the correct

2277 fork or branch, and always calls this create_pull_request method after.

2278 If the PR exists already just do nothing because Bitbucket will

2279 auto-update the PR if the from ref is updated.

2280 """

2281 if "id" not in response_data:

2282 pr_exist_message = (

2283 "Only one pull request may be open for a given source and target branch"

2284 )

2285 if pr_exist_message in error_message:

2286 return

2287 self.failed_pull_request(error_message, pr_url, response, response_data)

2288

2289

2290class BitbucketCloudRepository(GitMergeRequestBase):

2291 """Bitbucket Cloud repository implementation."""

2292

2293 name: ClassVar[StrOrPromise] = gettext_lazy("Bitbucket Cloud merge request")

2294 identifier: ClassVar[str] = "bitbucketcloud"

2295 _version: ClassVar[str | None] = None

2296 API_TEMPLATE: ClassVar[str] = (

2297 "{scheme}://api.{host}/2.0/repositories/{owner}/{slug}"

2298 )

2299 REQUIRED_CONFIG: ClassVar[set[str]] = {"username", "token", "workspace"}

2300

2301 def get_credentials(self) -> GitCredentials:

2302 """Return credentials for Bitbucket Cloud."""

2303 super_credentials = super().get_credentials()

2304 credentials = self.get_credentials_by_hostname(super_credentials["hostname"])

2305 super_credentials["workspace"] = credentials["workspace"]

2306 return super_credentials

2307

2308 def get_auth(self, credentials: GitCredentials) -> tuple[str, str]:

2309 """Return Bitbucket Cloud authentication App Password credentials."""

2310 return credentials["username"], credentials["token"]

2311

2312 def get_headers(self, credentials: GitCredentials) -> dict[str, str]:

2313 """Return HTTP headers for Bitbucket Cloud API requests."""

2314 return {

2315 "Accept": "application/json",

2316 }

2317

2318 def get_default_reviewers_uuids(self, credentials: GitCredentials) -> list[str]:

2319 """Get a list of uuids of default reviewers for a repository."""

2320 list_reviewers_url = "{}/default-reviewers".format(credentials["url"])

2321 try:

2322 reviewers = self.build_full_paginated_result(

2323 credentials, list_reviewers_url, "Reviewers listing error: "

2324 )

2325 except RepositoryError:

2326 return []

2327

2328 return [reviewer["uuid"] for reviewer in reviewers]

2329

2330 def build_full_paginated_result(

2331 self,

2332 credentials: GitCredentials,

2333 url: str,

2334 error_message: str,

2335 ) -> list[Any]:

2336 """

2337 Build result from paginated endpoint of Bitbucket Cloud.

2338

2339 This method assumes that the result is an array of objects

2340 and that the request method is "GET"

2341 """

2342 result: list[Any] = []

2343 next_url: str | None = url

2344

2345 while next_url:

2346 response_data, _response, error = self.request("get", credentials, next_url)

2347

2348 if error:

2349 report_error("Could not paginate", message=True)

2350 raise RepositoryError(

2351 0,

2352 f"{error_message} {error}",

2353 )

2354

2355 result.extend(response_data["values"])

2356 next_url = response_data.get("next")

2357 return result

2358

2359 def create_pull_request(

2360 self,

2361 credentials: GitCredentials,

2362 origin_branch: str,

2363 fork_remote: str,

2364 fork_branch: str,

2365 ) -> None:

2366 """

2367 Create pull request on Bitbucket Cloud.

2368

2369 Returns the PR data is PR already exists

2370 """

2371 pr_url = "{}/pullrequests".format(credentials["url"])

2372 title, description = self.get_merge_message()

2373

2374 payload = {

2375 "title": title,

2376 "description": description,

2377 "source": {

2378 "branch": {"name": fork_branch},

2379 "repository": {"type": "repository", "name": fork_remote},

2380 },

2381 "destination": {

2382 "branch": {"name": origin_branch},

2383 },

2384 "reviewers": [

2385 {"uuid": reviewer}

2386 for reviewer in self.get_default_reviewers_uuids(credentials)

2387 ],

2388 }

2389

2390 response_data, response, error = self.request(

2391 "post", credentials, pr_url, json=payload

2392 )

2393 # Bitbucket Cloud handles Pull request already exists

2394 # and just returns its data

2395

2396 if response_data["type"] == "error":

2397 # gracefully handle nothing to merge case

2398 if "There are no changes to be pulled" in error:

2399 return

2400

2401 self.failed_pull_request(error, pr_url, response, response_data)

2402

2403 def create_fork(self, credentials: GitCredentials) -> None:

2404 """

2405 Create a fork of the given repository.

2406

2407 This method will first check if a fork already exists.

2408 If it does, it will use that fork.

2409 If not, it will create a new fork.

2410 """

2411 fork_url = "{}/forks".format(credentials["url"])

2412 payload: dict[str, str | dict[str, str]] = {

2413 "workspace": {"slug": credentials["workspace"]}

2414 }

2415

2416 forked_repo: dict | None = None

2417

2418 # list forks of users, if fork already exist, set remote_fork as fork

2419 for fork in self.list_repo_forks(credentials):

2420 if fork["owner"]["username"] == credentials["workspace"]:

2421 forked_repo = fork

2422

2423 if forked_repo is None:

2424 response_data, response, error = self.request(

2425 "post", credentials, fork_url, json=payload

2426 )

2427

2428 # if name is already taken, append an numeric value at the end

2429 if "already has a repository with this name" in error:

2430 fork_name = "{}-{}".format(

2431 credentials["slug"],

2432 random.randint(1000, 9999), # noqa: S311

2433 )

2434 payload["name"] = fork_name

2435

2436 response_data, response, error = self.request(

2437 "post", credentials, fork_url, json=payload

2438 )

2439

2440 if response_data["type"] == "error":

2441 report_error("Could not fork repository", message=True)

2442 raise RepositoryError(

2443 0, self.get_fork_failed_message(error, credentials, response)

2444 )

2445

2446 forked_repo = response_data

2447

2448 ssh_url_to_repo = next(

2449 (

2450 link["href"]

2451 for link in forked_repo["links"]["clone"]

2452 if link["name"] == "ssh"

2453 ),

2454 "",

2455 )

2456 http_url_to_repo = next(

2457 (

2458 link["href"]

2459 for link in forked_repo["links"]["clone"]

2460 if link["name"] == "https"

2461 ),

2462 "",

2463 )

2464

2465 self.configure_fork_remote(ssh_url_to_repo, http_url_to_repo, credentials)

2466

2467 def list_repo_forks(self, credentials: GitCredentials) -> list[dict[str, Any]]:

2468 """List all forks of a repository."""

2469 list_forks_url = "{}/forks".format(credentials["url"])

2470 return self.build_full_paginated_result(

2471 credentials, list_forks_url, "Forks listing error: "

2472 )