Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/repo/fun.py: 28%

1# This module is part of GitPython and is released under the

2# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/

4"""General repository-related functions."""

6from __future__ import annotations

8__all__ = [

9 "rev_parse",

10 "is_git_dir",

11 "touch",

12 "find_submodule_git_dir",

13 "name_to_object",

14 "short_to_long",

15 "deref_tag",

16 "to_commit",

17 "find_worktree_git_dir",

18]

20import os

21import os.path as osp

22from pathlib import Path

23import re

24import stat

25from string import digits

27from gitdb.exc import BadName, BadObject

29from git.cmd import Git

30from git.exc import WorkTreeRepositoryUnsupported

31from git.objects import Object

32from git.objects.util import parse_date

33from git.refs import SymbolicReference

34from git.util import cygpath, bin_to_hex, hex_to_bin

36# Typing ----------------------------------------------------------------------

38from typing import Iterator, Optional, TYPE_CHECKING, Tuple, Union, cast, overload

40from git.types import AnyGitObject, Literal, PathLike

42if TYPE_CHECKING:

43 from git.db import GitCmdObjectDB

44 from git.objects import Commit

45 from git.refs.reference import Reference

46 from git.refs.log import RefLog, RefLogEntry

47 from git.refs.tag import Tag

49 from .base import Repo

51# ----------------------------------------------------------------------------

54def touch(filename: str) -> str:

55 with open(filename, "ab"):

56 pass

57 return filename

60def is_git_dir(d: PathLike) -> bool:

61 """This is taken from the git setup.c:is_git_directory function.

63 .. note::

64 This function recognizes repositories using reftable through their

65 compatibility files, but GitPython's direct reference access does not support

66 reftable.

68 :raise git.exc.WorkTreeRepositoryUnsupported:

69 If it sees a worktree directory. It's quite hacky to do that here, but at least

70 clearly indicates that we don't support it. There is the unlikely danger to

71 throw if we see directories which just look like a worktree dir, but are none.

72 """

73 if osp.isdir(d):

74 headref = osp.join(d, "HEAD")

75 if osp.islink(headref):

76 try:

77 valid_head = os.readlink(headref).startswith("refs/")

78 except OSError:

79 valid_head = False

80 else:

81 try:

82 with open(headref, "rb") as fp:

83 head = fp.read(256)

84 except OSError:

85 valid_head = False

86 else:

87 valid_head = (head.startswith(b"ref:") and head[4:].lstrip().startswith(b"refs/")) or bool(

88 re.match(rb"(?:[0-9A-Fa-f]{64}|[0-9A-Fa-f]{40})", head)

89 )

91 common_dir = os.getenv("GIT_COMMON_DIR")

92 if common_dir == "":

93 return False

94 if common_dir is None:

95 common_dir_file = Path(d) / "commondir"

96 try:

97 common_dir = os.fsdecode(common_dir_file.read_bytes()).rstrip("\r\n")

98 except FileNotFoundError:

99 if osp.lexists(common_dir_file):

100 return False

101 common_dir = os.fspath(d)

102 except (OSError, UnicodeError):

103 return False

104 else:

105 if not common_dir:

106 return False

107 try:

108 common_dir = osp.realpath(osp.join(d, common_dir))

109 except (OSError, ValueError):

110 return False

111

112 object_dir = os.getenv("GIT_OBJECT_DIRECTORY")

113 if object_dir is None:

114 object_dir = osp.join(common_dir, "objects")

115 if valid_head and osp.isdir(object_dir) and osp.isdir(osp.join(common_dir, "refs")):

116 return True

117 if osp.isfile(osp.join(d, "gitdir")) and osp.isfile(osp.join(d, "commondir")) and osp.isfile(headref):

118 raise WorkTreeRepositoryUnsupported(d)

119 return False

120

121

122def find_worktree_git_dir(dotgit: PathLike) -> Optional[str]:

123 """Search for a gitdir for this worktree."""

124 try:

125 statbuf = os.stat(dotgit)

126 except (FileNotFoundError, NotADirectoryError):

127 return None

128 if not stat.S_ISREG(statbuf.st_mode) or statbuf.st_size > (1 << 20):

129 return None

130

131 try:

132 with open(dotgit, "rb") as fp:

133 content_bytes = fp.read(statbuf.st_size)

134 if len(content_bytes) != statbuf.st_size:

135 return None

136 content = os.fsdecode(content_bytes).rstrip("\r\n")

137 except (OSError, UnicodeError):

138 return None

139 return content[8:] if len(content) >= 9 and content.startswith("gitdir: ") else None

140

141

142def find_submodule_git_dir(d: PathLike) -> Optional[PathLike]:

143 """Search for a submodule repo."""

144 if is_git_dir(d):

145 return d

146

147 path = find_worktree_git_dir(d)

148 if path is None:

149 return None

150

151 if Git.is_cygwin():

152 # Cygwin creates submodules prefixed with `/cygdrive/...`.

153 # Cygwin git understands Cygwin paths much better than Windows ones.

154 # Also the Cygwin tests are assuming Cygwin paths.

155 path = cygpath(path)

156 if not osp.isabs(path):

157 path = osp.normpath(osp.join(osp.dirname(d), path))

158 return path if is_git_dir(path) else None

159

160

161def short_to_long(odb: "GitCmdObjectDB", hexsha: str) -> Optional[bytes]:

162 """

163 :return:

164 Long hexadecimal sha1 from the given less than 40 byte hexsha, or ``None`` if no

165 candidate could be found.

166

167 :param hexsha:

168 hexsha with less than 40 bytes.

169 """

170 try:

171 return bin_to_hex(odb.partial_to_complete_sha_hex(hexsha))

172 except BadObject:

173 return None

174 # END exception handling

175

176

177def _describe_to_long(repo: "Repo", name: str) -> Optional[bytes]:

178 """Resolve git-describe style names to the abbreviated object they contain."""

179 match = re.match(r"^.+-\d+-g([0-9A-Fa-f]{4,40})(?:-dirty)?$", name)

180 if match is None:

181 match = re.match(r"^.+-g([0-9A-Fa-f]{4,40})(?:-dirty)?$", name)

182 if match is None:

183 match = re.match(r"^([0-9A-Fa-f]{4,40})-dirty$", name)

184 if match is None:

185 return None

186 # END handle match

187

188 hexsha = match.group(1)

189 if len(hexsha) == 40:

190 return hexsha.encode("ascii")

191 return short_to_long(repo.odb, hexsha)

192

193

194@overload

195def name_to_object(repo: "Repo", name: str, return_ref: Literal[False] = ...) -> AnyGitObject: ...

196

197

198@overload

199def name_to_object(repo: "Repo", name: str, return_ref: Literal[True]) -> Union[AnyGitObject, SymbolicReference]: ...

200

201

202def name_to_object(repo: "Repo", name: str, return_ref: bool = False) -> Union[AnyGitObject, SymbolicReference]:

203 """

204 :return:

205 Object specified by the given name - hexshas (short and long) as well as

206 references are supported.

207

208 :param return_ref:

209 If ``True``, and name specifies a reference, we will return the reference

210 instead of the object. Otherwise it will raise :exc:`~gitdb.exc.BadObject` or

211 :exc:`~gitdb.exc.BadName`.

212 """

213 hexsha: Union[None, str, bytes] = None

214

215 # Is it a hexsha? Try the most common ones, which is 7 to 40.

216 if repo.re_hexsha_shortened.match(name):

217 if len(name) != 40:

218 # Find long sha for short sha.

219 hexsha = short_to_long(repo.odb, name)

220 else:

221 hexsha = name

222 # END handle short shas

223 # END find sha if it matches

224

225 # If we couldn't find an object for what seemed to be a short hexsha, try to find it

226 # as reference anyway, it could be named 'aaa' for instance.

227 if hexsha is None:

228 for base in (

229 "%s",

230 "refs/%s",

231 "refs/tags/%s",

232 "refs/heads/%s",

233 "refs/remotes/%s",

234 "refs/remotes/%s/HEAD",

235 ):

236 try:

237 hexsha = SymbolicReference.dereference_recursive(repo, base % name)

238 if return_ref:

239 return SymbolicReference(repo, base % name)

240 # END handle symbolic ref

241 break

242 except ValueError:

243 pass

244 # END for each base

245 # END handle hexsha

246

247 if hexsha is None:

248 hexsha = _describe_to_long(repo, name)

249 # END handle describe output

250

251 # Didn't find any ref, this is an error.

252 if return_ref:

253 raise BadObject("Couldn't find reference named %r" % name)

254 # END handle return ref

255

256 # Tried everything ? fail.

257 if hexsha is None:

258 raise BadName(name)

259 # END assert hexsha was found

260

261 return Object.new_from_sha(repo, hex_to_bin(hexsha))

262

263

264def deref_tag(tag: "Tag") -> AnyGitObject:

265 """Recursively dereference a tag and return the resulting object."""

266 while True:

267 try:

268 tag = tag.object

269 except AttributeError:

270 break

271 # END dereference tag

272 return tag

273

274

275def to_commit(obj: Object) -> "Commit":

276 """Convert the given object to a commit if possible and return it."""

277 if obj.type == "tag":

278 obj = deref_tag(obj)

279

280 if obj.type != "commit":

281 raise ValueError("Cannot convert object %r to type commit" % obj)

282 # END verify type

283 return obj

284

285

286def _object_from_hexsha(repo: "Repo", hexsha: str) -> AnyGitObject:

287 return Object.new_from_sha(repo, hex_to_bin(hexsha))

288

289

290def _current_reflog_ref(repo: "Repo") -> SymbolicReference:

291 try:

292 return repo.head.ref

293 except TypeError:

294 return repo.head

295 # END handle detached head

296

297

298def _common_reflog_path(repo: "Repo", ref: SymbolicReference) -> Optional[str]:

299 if repo.common_dir == repo.git_dir:

300 return None

301 # END handle normal repository

302 return SymbolicReference._get_validated_path(osp.join(repo.common_dir, "logs"), ref.path)

303

304

305def _ref_log(repo: "Repo", ref: SymbolicReference) -> "RefLog":

306 try:

307 return ref.log()

308 except FileNotFoundError:

309 common_path = _common_reflog_path(repo, ref)

310 if common_path and osp.isfile(common_path):

311 from git.refs.log import RefLog

312

313 return RefLog.from_file(common_path)

314 # END handle linked-worktree branch logs

315 try:

316 if ref.path == repo.head.ref.path:

317 return repo.head.log()

318 # END handle linked-worktree current branch logs

319 except TypeError:

320 pass

321 # END handle detached head

322 raise

323 # END handle missing branch log

324

325

326def _ref_log_entry(repo: "Repo", ref: SymbolicReference, index: int) -> "RefLogEntry":

327 try:

328 return ref.log_entry(index)

329 except FileNotFoundError:

330 common_path = _common_reflog_path(repo, ref)

331 if common_path and osp.isfile(common_path):

332 from git.refs.log import RefLog

333

334 return RefLog.entry_at(common_path, index)

335 # END handle linked-worktree branch logs

336 try:

337 if ref.path == repo.head.ref.path:

338 return repo.head.log_entry(index)

339 # END handle linked-worktree current branch logs

340 except TypeError:

341 pass

342 # END handle detached head

343 raise

344 # END handle missing branch log

345

346

347def _find_reflog_entry_by_date(repo: "Repo", ref: SymbolicReference, spec: str) -> str:

348 try:

349 timestamp, _offset = parse_date(spec)

350 except ValueError as e:

351 raise NotImplementedError("Support for additional @{...} modes not implemented") from e

352 # END handle unsupported dates

353 log = _ref_log(repo, ref)

354 if not log:

355 raise IndexError("Invalid revlog date: %s" % spec)

356 # END handle empty log

357

358 for entry in reversed(log):

359 if entry.time[0] <= timestamp:

360 return entry.newhexsha

361 # END found candidate

362 # END for each entry

363 return log[0].newhexsha

364

365

366def _previous_checked_out_branch(repo: "Repo", nth: int) -> AnyGitObject:

367 if nth <= 0:

368 raise ValueError("Invalid previous checkout selector: -%i" % nth)

369 # END handle invalid input

370

371 seen = 0

372 for entry in reversed(_ref_log(repo, repo.head)):

373 message = entry.message or ""

374 prefix = "checkout: moving from "

375 if not message.startswith(prefix):

376 continue

377 # END skip non-checkouts

378

379 previous_branch = message[len(prefix) :].split(" to ", 1)[0]

380 seen += 1

381 if seen == nth:

382 return name_to_object(repo, previous_branch)

383 # END found selector

384 # END for each entry

385 raise IndexError("Invalid previous checkout selector: -%i" % nth)

386

387

388def _tracking_branch_object(repo: "Repo", ref: Optional[SymbolicReference]) -> AnyGitObject:

389 from git.refs.head import Head

390

391 if ref is None:

392 try:

393 head = repo.active_branch

394 except TypeError as e:

395 raise BadName("@{upstream}") from e

396 elif isinstance(ref, Head):

397 head = ref

398 elif os.fspath(ref.path).startswith("refs/heads/"):

399 head = Head(repo, ref.path)

400 else:

401 raise BadName("%s@{upstream}" % ref.name)

402 # END handle head

403

404 tracking_branch = head.tracking_branch()

405 if tracking_branch is None:

406 raise BadName("%s@{upstream}" % head.name)

407 # END handle missing upstream

408 return tracking_branch.commit

409

410

411def _apply_reflog(repo: "Repo", ref: Optional[SymbolicReference], content: str) -> AnyGitObject:

412 if content.startswith("+"):

413 content = content[1:]

414 # END handle explicit positive sign

415

416 if content.startswith("-"):

417 if ref is not None:

418 raise ValueError("Previous checkout selectors do not take an explicit ref")

419 if content == "-0":

420 raise ValueError("Negative zero is invalid in reflog selector")

421 # END handle invalid negative zero

422 try:

423 return _previous_checked_out_branch(repo, int(content[1:]))

424 except ValueError as e:

425 raise ValueError("Invalid previous checkout selector: %s" % content) from e

426 # END handle previous checkout branch

427

428 content_lower = content.lower()

429 if content_lower in ("u", "upstream", "push"):

430 return _tracking_branch_object(repo, ref)

431 # END handle sibling branches

432

433 ref = ref or _current_reflog_ref(repo)

434 try:

435 entry_no = int(content)

436 except ValueError:

437 hexsha = _find_reflog_entry_by_date(repo, ref, content)

438 else:

439 if entry_no >= 100000000:

440 hexsha = _find_reflog_entry_by_date(repo, ref, "%s +0000" % entry_no)

441 elif entry_no == 0:

442 return ref.commit

443 else:

444 try:

445 entry = _ref_log_entry(repo, ref, -(entry_no + 1))

446 except IndexError as e:

447 raise IndexError("Invalid revlog index: %i" % entry_no) from e

448 # END handle index out of bound

449 hexsha = entry.newhexsha

450 # END handle offset or date-like timestamp

451 # END handle content

452 return _object_from_hexsha(repo, hexsha)

453

454

455def _find_closing_brace(rev: str, start: int) -> int:

456 depth = 1

457 escaped = False

458 for idx in range(start + 1, len(rev)):

459 char = rev[idx]

460 if escaped:

461 escaped = False

462 elif char == "\\":

463 escaped = True

464 elif char == "{":

465 depth += 1

466 elif char == "}":

467 depth -= 1

468 if depth == 0:

469 return idx

470 # END found end

471 # END handle char

472 # END for each char

473 raise ValueError("Missing closing brace to define type in %s" % rev)

474

475

476def _parse_search(pattern: str) -> Tuple[str, bool]:

477 if not pattern:

478 raise ValueError("Revision search requires a pattern")

479 # END handle empty pattern

480

481 if pattern.startswith("!-"):

482 return pattern[2:], True

483 if pattern.startswith("!!"):

484 return pattern[1:], False

485 if pattern.startswith("!"):

486 raise ValueError("Need one character after /!, typically -")

487 return pattern, False

488

489

490def _unescape_braced_regex(pattern: str) -> str:

491 out = []

492 idx = 0

493 while idx < len(pattern):

494 char = pattern[idx]

495 if char == "\\" and idx + 1 < len(pattern):

496 next_char = pattern[idx + 1]

497 if next_char in "{}\\":

498 out.append(next_char)

499 else:

500 out.append(char)

501 out.append(next_char)

502 # END handle escaped char

503 idx += 2

504 continue

505 # END handle backslash

506 out.append(char)

507 idx += 1

508 # END for each char

509 return "".join(out)

510

511

512def _find_commit_by_message(

513 repo: "Repo", rev: Optional[AnyGitObject], pattern: str, braced: bool = False

514) -> AnyGitObject:

515 pattern, negated = _parse_search(_unescape_braced_regex(pattern) if braced else pattern)

516 try:

517 regex = re.compile(pattern)

518 except re.error as e:

519 raise ValueError("Invalid commit message regex %r" % pattern) from e

520 # END handle invalid regex

521 if rev is None:

522 commits = _all_ref_commits(repo)

523 else:

524 commits = _reachable_commits([to_commit(cast(Object, rev))])

525 # END handle starting point

526

527 for commit in commits:

528 message = commit.message

529 if isinstance(message, bytes):

530 message = message.decode(commit.encoding, "replace")

531 # END handle bytes message

532 matches = regex.search(message or "") is not None

533 if matches != negated:

534 return commit

535 # END found commit

536 # END for each commit

537 raise BadName("No commit found matching message pattern %r" % pattern)

538

539

540def _all_ref_commits(repo: "Repo") -> Iterator["Commit"]:

541 starts = []

542 for ref in repo.references:

543 try:

544 starts.append(to_commit(cast(Object, ref.object)))

545 except (BadName, ValueError):

546 pass

547 # END skip refs that do not point to commits

548 # END for each ref

549 try:

550 starts.append(repo.head.commit)

551 except ValueError:

552 pass

553 # END handle unborn head

554 return _reachable_commits(starts)

555

556

557def _reachable_commits(starts: list["Commit"]) -> Iterator["Commit"]:

558 seen = set()

559 pending = starts[:]

560 while pending:

561 pending.sort(key=lambda commit: commit.committed_date, reverse=True)

562 commit = pending.pop(0)

563 if commit.binsha in seen:

564 continue

565 # END skip seen commit

566 seen.add(commit.binsha)

567 yield commit

568 pending.extend(commit.parents)

569 # END while commits remain

570

571

572def _index_lookup(repo: "Repo", spec: str) -> AnyGitObject:

573 if not spec:

574 raise ValueError("':' must be followed by a path")

575 # END handle empty lookup

576

577 stage = 0

578 path = spec

579 if len(spec) >= 2 and spec[1] == ":" and spec[0] in "0123":

580 stage = int(spec[0])

581 path = spec[2:]

582 # END handle stage

583

584 try:

585 return repo.index.entries[(path, stage)].to_blob(repo)

586 except KeyError as e:

587 raise BadName("Path %r did not exist in the index at stage %i" % (path, stage)) from e

588

589

590def _tree_lookup(obj: AnyGitObject, path: str) -> AnyGitObject:

591 if obj.type != "tree":

592 obj = to_commit(cast(Object, obj)).tree

593 # END get tree

594 if not path:

595 return obj

596 return obj[path]

597

598

599def _peel(obj: AnyGitObject, output_type: str, repo: "Repo", rev: str) -> AnyGitObject:

600 if output_type.startswith("/"):

601 return _find_commit_by_message(repo, obj, output_type[1:], braced=True)

602 if output_type == "":

603 return deref_tag(obj) if obj.type == "tag" else obj

604 if output_type == "object":

605 return obj

606 if output_type == "commit":

607 return to_commit(cast(Object, obj))

608 if output_type == "tree":

609 return to_commit(cast(Object, obj)).tree if obj.type != "tree" else obj

610 if output_type == "blob":

611 obj = deref_tag(obj) if obj.type == "tag" else obj

612 if obj.type == output_type:

613 return obj

614 # END handle matching type

615 raise ValueError("Could not accommodate requested object type %r, got %s" % (output_type, obj.type))

616 if output_type == "tag":

617 if obj.type == output_type:

618 return obj

619 # END handle matching type

620 raise ValueError("Could not accommodate requested object type %r, got %s" % (output_type, obj.type))

621 # END handle known types

622 raise ValueError("Invalid output type: %s ( in %s )" % (output_type, rev))

623

624

625def _first_rev_token(rev: str) -> Optional[int]:

626 for idx, char in enumerate(rev):

627 if char in "^~:":

628 return idx

629 if char == "@":

630 next_char = rev[idx + 1] if idx + 1 < len(rev) else None

631 if idx == 0 and next_char in (None, "^", "~", ":", "{"):

632 return idx

633 if next_char == "{":

634 return idx

635 # END handle reflog selector

636 # END handle at symbol

637 # END for each char

638 return None

639

640

641def rev_parse(repo: "Repo", rev: str) -> AnyGitObject:

642 """Parse a revision string. Like :manpage:`git-rev-parse(1)`.

643

644 :return:

645 `~git.objects.base.Object` at the given revision.

646

647 This may be any type of git object:

648

649 * :class:`Commit <git.objects.commit.Commit>`

650 * :class:`TagObject <git.objects.tag.TagObject>`

651 * :class:`Tree <git.objects.tree.Tree>`

652 * :class:`Blob <git.objects.blob.Blob>`

653

654 :param rev:

655 :manpage:`git-rev-parse(1)`-compatible revision specification as string.

656 Please see :manpage:`git-rev-parse(1)` for details.

657

658 :raise gitdb.exc.BadObject:

659 If the given revision could not be found.

660

661 :raise ValueError:

662 If `rev` couldn't be parsed.

663

664 :raise IndexError:

665 If an invalid reflog index is specified.

666 """

667 if rev.startswith(":/"):

668 return _find_commit_by_message(repo, None, rev[2:])

669 if rev.startswith(":"):

670 return _index_lookup(repo, rev[1:])

671 # END handle top-level colon modes

672

673 obj: Optional[AnyGitObject] = None

674 ref = None

675 lr = len(rev)

676 first_token = _first_rev_token(rev)

677 if first_token is None:

678 return name_to_object(repo, rev)

679 # END handle plain name

680

681 if first_token == 0:

682 if rev[0] != "@":

683 raise ValueError("Revision specifier must start with an object name: %s" % rev)

684 # END handle invalid leading token

685 ref = _current_reflog_ref(repo)

686 obj = ref.commit

687 start = 0 if rev.startswith("@{") else 1

688 else:

689 if rev[first_token] == "@":

690 ref = cast("Reference", name_to_object(repo, rev[:first_token], return_ref=True))

691 obj = ref.commit

692 else:

693 obj = name_to_object(repo, rev[:first_token])

694 # END handle anchor

695 start = first_token

696 # END initialize anchor

697

698 while start < lr:

699 token = rev[start]

700

701 if token == "@":

702 if start + 1 >= lr or rev[start + 1] != "{":

703 raise ValueError("Invalid @ token in revision specifier: %s" % rev)

704 # END handle invalid @

705 end = _find_closing_brace(rev, start + 1)

706 obj = _apply_reflog(repo, ref if first_token != 0 and start == first_token else None, rev[start + 2 : end])

707 ref = None

708 start = end + 1

709 continue

710 # END handle reflog

711

712 if token == ":":

713 return _tree_lookup(obj, rev[start + 1 :])

714 # END handle path

715

716 start += 1

717

718 if token == "^" and start < lr and rev[start] == "{":

719 end = _find_closing_brace(rev, start)

720 obj = _peel(obj, rev[start + 1 : end], repo, rev)

721 ref = None

722 start = end + 1

723 continue

724 # END parse type

725

726 num = 0

727 found_digit = False

728 while start < lr:

729 if rev[start] in digits:

730 num = num * 10 + int(rev[start])

731 start += 1

732 found_digit = True

733 else:

734 break

735 # END handle number

736 # END number parse loop

737

738 if not found_digit:

739 num = 1

740 # END set default num

741

742 try:

743 if token == "~":

744 obj = to_commit(obj)

745 for _ in range(num):

746 obj = obj.parents[0]

747 # END for each history item to walk

748 elif token == "^":

749 obj = to_commit(obj)

750 if num == 0:

751 pass

752 else:

753 obj = obj.parents[num - 1]

754 # END handle parent

755 else:

756 raise ValueError("Invalid token: %r" % token)

757 # END end handle tag

758 except (IndexError, AttributeError) as e:

759 raise BadName(

760 f"Invalid revision spec '{rev}' - not enough parent commits to reach '{token}{int(num)}'"

761 ) from e

762 # END exception handling

763 # END parse loop

764

765 if obj is None:

766 raise ValueError("Revision specifier could not be parsed: %s" % rev)

767

768 return obj