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

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 GitCommandError, 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 Optional, TYPE_CHECKING, Union, cast, overload

40from git.types import AnyGitObject, Literal, PathLike

42if TYPE_CHECKING:

43 from gitdb.db import CompoundDB, LooseObjectDB

44 from git.objects import Commit

45 from git.refs.reference import Reference

46 from git.refs.log import RefLog, RefLogEntry

48 from .base import Repo

50# ----------------------------------------------------------------------------

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

54 with open(filename, "ab"):

55 pass

56 return filename

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

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

62 .. note::

63 This function recognizes repositories using reftable through their

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

65 reftable.

67 :raise git.exc.WorkTreeRepositoryUnsupported:

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

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

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

71 """

72 if osp.isdir(d):

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

74 if osp.islink(headref):

75 try:

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

77 except OSError:

78 valid_head = False

79 else:

80 try:

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

82 head = fp.read(256)

83 except OSError:

84 valid_head = False

85 else:

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

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

88 )

90 common_dir = os.getenv("GIT_COMMON_DIR")

91 if common_dir == "":

92 return False

93 if common_dir is None:

94 common_dir_file = Path(d) / "commondir"

95 try:

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

97 except FileNotFoundError:

98 if osp.lexists(common_dir_file):

99 return False

100 common_dir = os.fspath(d)

101 except (OSError, UnicodeError):

102 return False

103 else:

104 if not common_dir:

105 return False

106 try:

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

108 except (OSError, ValueError):

109 return False

110

111 object_dir = os.getenv("GIT_OBJECT_DIRECTORY")

112 if object_dir is None:

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

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

115 return True

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

117 raise WorkTreeRepositoryUnsupported(d)

118 return False

119

120

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

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

123 try:

124 statbuf = os.stat(dotgit)

125 except (FileNotFoundError, NotADirectoryError):

126 return None

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

128 return None

129

130 try:

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

132 content_bytes = fp.read(statbuf.st_size)

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

134 return None

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

136 except (OSError, UnicodeError):

137 return None

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

139

140

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

142 """Search for a submodule repo."""

143 if is_git_dir(d):

144 return d

145

146 path = find_worktree_git_dir(d)

147 if path is None:

148 return None

149

150 if Git.is_cygwin():

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

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

153 # Also the Cygwin tests are assuming Cygwin paths.

154 path = cygpath(path, expand_vars=False)

155 if not osp.isabs(path):

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

157 return path if is_git_dir(path) else None

158

159

160def short_to_long(odb: Union["CompoundDB", "LooseObjectDB"], hexsha: str) -> Optional[bytes]:

161 """

162 :return:

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

164 candidate could be found.

165

166 :param hexsha:

167 hexsha with less than 40 bytes.

168 """

169 try:

170 return bin_to_hex(odb.partial_to_complete_sha_hex(hexsha))

171 except BadObject:

172 return None

173 # END exception handling

174

175

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

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

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

179 if match is None:

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

181 if match is None:

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

183 if match is None:

184 return None

185 # END handle match

186

187 hexsha = match.group(1)

188 if len(hexsha) == 40:

189 return hexsha.encode("ascii")

190 return short_to_long(repo.odb, hexsha)

191

192

193@overload

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

195

196

197@overload

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

199

200

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

202 """

203 :return:

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

205 references are supported.

206

207 :param return_ref:

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

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

210 :exc:`~gitdb.exc.BadName`.

211 """

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

213

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

215 if repo.re_hexsha_shortened.match(name):

216 if len(name) != 40:

217 # Find long sha for short sha.

218 hexsha = short_to_long(repo.odb, name)

219 else:

220 hexsha = name

221 # END handle short shas

222 # END find sha if it matches

223

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

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

226 if hexsha is None:

227 for base in (

228 "%s",

229 "refs/%s",

230 "refs/tags/%s",

231 "refs/heads/%s",

232 "refs/remotes/%s",

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

234 ):

235 try:

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

237 if return_ref:

238 return SymbolicReference(repo, base % name)

239 # END handle symbolic ref

240 break

241 except ValueError:

242 pass

243 # END for each base

244 # END handle hexsha

245

246 if hexsha is None:

247 hexsha = _describe_to_long(repo, name)

248 # END handle describe output

249

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

251 if return_ref:

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

253 # END handle return ref

254

255 # Tried everything ? fail.

256 if hexsha is None:

257 raise BadName(name)

258 # END assert hexsha was found

259

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

261

262

263def deref_tag(tag: AnyGitObject) -> AnyGitObject:

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

265 while True:

266 try:

267 tag = tag.object

268 except AttributeError:

269 break

270 # END dereference tag

271 return tag

272

273

274def to_commit(obj: AnyGitObject) -> "Commit":

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

276 if obj.type == "tag":

277 obj = deref_tag(obj)

278

279 if obj.type != "commit":

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

281 # END verify type

282 return obj

283

284

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

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

287

288

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

290 try:

291 return repo.head.ref

292 except TypeError:

293 return repo.head

294 # END handle detached head

295

296

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

298 if repo.common_dir == repo.git_dir:

299 return None

300 # END handle normal repository

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

302

303

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

305 try:

306 return ref.log()

307 except FileNotFoundError:

308 common_path = _common_reflog_path(repo, ref)

309 if common_path and osp.isfile(common_path):

310 from git.refs.log import RefLog

311

312 return RefLog.from_file(common_path)

313 # END handle linked-worktree branch logs

314 try:

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

316 return repo.head.log()

317 # END handle linked-worktree current branch logs

318 except TypeError:

319 pass

320 # END handle detached head

321 raise

322 # END handle missing branch log

323

324

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

326 try:

327 return ref.log_entry(index)

328 except FileNotFoundError:

329 common_path = _common_reflog_path(repo, ref)

330 if common_path and osp.isfile(common_path):

331 from git.refs.log import RefLog

332

333 return RefLog.entry_at(common_path, index)

334 # END handle linked-worktree branch logs

335 try:

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

337 return repo.head.log_entry(index)

338 # END handle linked-worktree current branch logs

339 except TypeError:

340 pass

341 # END handle detached head

342 raise

343 # END handle missing branch log

344

345

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

347 try:

348 timestamp, _offset = parse_date(spec)

349 except ValueError as e:

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

351 # END handle unsupported dates

352 log = _ref_log(repo, ref)

353 if not log:

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

355 # END handle empty log

356

357 for entry in reversed(log):

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

359 return entry.newhexsha

360 # END found candidate

361 # END for each entry

362 return log[0].newhexsha

363

364

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

366 if nth <= 0:

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

368 # END handle invalid input

369

370 seen = 0

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

372 message = entry.message or ""

373 prefix = "checkout: moving from "

374 if not message.startswith(prefix):

375 continue

376 # END skip non-checkouts

377

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

379 seen += 1

380 if seen == nth:

381 return name_to_object(repo, previous_branch)

382 # END found selector

383 # END for each entry

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

385

386

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

388 from git.refs.head import Head

389

390 if ref is None:

391 try:

392 head = repo.active_branch

393 except TypeError as e:

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

395 elif isinstance(ref, Head):

396 head = ref

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

398 head = Head(repo, ref.path)

399 else:

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

401 # END handle head

402

403 tracking_branch = head.tracking_branch()

404 if tracking_branch is None:

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

406 # END handle missing upstream

407 return tracking_branch.commit

408

409

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

411 if content.startswith("+"):

412 content = content[1:]

413 # END handle explicit positive sign

414

415 if content.startswith("-"):

416 if ref is not None:

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

418 if content == "-0":

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

420 # END handle invalid negative zero

421 try:

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

423 except ValueError as e:

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

425 # END handle previous checkout branch

426

427 content_lower = content.lower()

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

429 return _tracking_branch_object(repo, ref)

430 # END handle sibling branches

431

432 ref = ref or _current_reflog_ref(repo)

433 try:

434 entry_no = int(content)

435 except ValueError:

436 hexsha = _find_reflog_entry_by_date(repo, ref, content)

437 else:

438 if entry_no >= 100000000:

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

440 elif entry_no == 0:

441 return ref.commit

442 else:

443 try:

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

445 except IndexError as e:

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

447 # END handle index out of bound

448 hexsha = entry.newhexsha

449 # END handle offset or date-like timestamp

450 # END handle content

451 return _object_from_hexsha(repo, hexsha)

452

453

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

455 depth = 1

456 escaped = False

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

458 char = rev[idx]

459 if escaped:

460 escaped = False

461 elif char == "\\":

462 escaped = True

463 elif char == "{":

464 depth += 1

465 elif char == "}":

466 depth -= 1

467 if depth == 0:

468 return idx

469 # END found end

470 # END handle char

471 # END for each char

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

473

474

475def _find_commit_by_message(repo: "Repo", rev: Optional[AnyGitObject], pattern: str) -> AnyGitObject:

476 if not pattern:

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

478 if pattern.startswith("!") and not pattern.startswith(("!-", "!!")):

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

480

481 # Git's native regular expressions avoid Python's exponential backtracking,

482 # and its history walk does not deserialize every visited commit in Python.

483 search = ":/" + pattern if rev is None else "%s^{/%s}" % (to_commit(rev).hexsha, pattern)

484 try:

485 # The fixed prefix prevents options without requiring Git 2.30's

486 # --end-of-options support in rev-parse.

487 hexsha = repo.git.rev_parse("--verify", search)

488 except GitCommandError as e:

489 # Git does not distinguish invalid regexes from searches with no match.

490 raise BadName("No commit found matching message pattern %r" % pattern) from e

491 return _object_from_hexsha(repo, hexsha)

492

493

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

495 if not spec:

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

497 # END handle empty lookup

498

499 stage = 0

500 path = spec

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

502 stage = int(spec[0])

503 path = spec[2:]

504 # END handle stage

505

506 try:

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

508 except KeyError as e:

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

510

511

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

513 if obj.type != "tree":

514 obj = to_commit(obj).tree

515 # END get tree

516 if not path:

517 return obj

518 return obj[path]

519

520

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

522 if output_type.startswith("/"):

523 return _find_commit_by_message(repo, obj, output_type[1:])

524 if output_type == "":

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

526 if output_type == "object":

527 return obj

528 if output_type == "commit":

529 return to_commit(obj)

530 if output_type == "tree":

531 return to_commit(obj).tree if obj.type != "tree" else obj

532 if output_type == "blob":

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

534 if obj.type == output_type:

535 return obj

536 # END handle matching type

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

538 if output_type == "tag":

539 if obj.type == output_type:

540 return obj

541 # END handle matching type

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

543 # END handle known types

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

545

546

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

548 for idx, char in enumerate(rev):

549 if char in "^~:":

550 return idx

551 if char == "@":

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

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

554 return idx

555 if next_char == "{":

556 return idx

557 # END handle reflog selector

558 # END handle at symbol

559 # END for each char

560 return None

561

562

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

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

565

566 :return:

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

568

569 This may be any type of git object:

570

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

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

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

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

575

576 :param rev:

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

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

579

580 Commit message searches use Git's native extended regular expressions.

581 Invalid search expressions and searches without a match both raise

582 :exc:`~gitdb.exc.BadName`.

583

584 :raise gitdb.exc.BadObject:

585 If the given revision could not be found.

586

587 :raise ValueError:

588 If `rev` couldn't be parsed.

589

590 :raise IndexError:

591 If an invalid reflog index is specified.

592 """

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

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

595 if rev.startswith(":"):

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

597 # END handle top-level colon modes

598

599 obj: Optional[AnyGitObject] = None

600 ref = None

601 lr = len(rev)

602 first_token = _first_rev_token(rev)

603 if first_token is None:

604 return name_to_object(repo, rev)

605 # END handle plain name

606

607 if first_token == 0:

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

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

610 # END handle invalid leading token

611 ref = _current_reflog_ref(repo)

612 obj = ref.commit

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

614 else:

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

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

617 obj = ref.commit

618 else:

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

620 # END handle anchor

621 start = first_token

622 # END initialize anchor

623

624 while start < lr:

625 token = rev[start]

626

627 if token == "@":

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

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

630 # END handle invalid @

631 end = _find_closing_brace(rev, start + 1)

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

633 ref = None

634 start = end + 1

635 continue

636 # END handle reflog

637

638 if token == ":":

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

640 # END handle path

641

642 start += 1

643

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

645 end = _find_closing_brace(rev, start)

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

647 ref = None

648 start = end + 1

649 continue

650 # END parse type

651

652 num = 0

653 found_digit = False

654 while start < lr:

655 if rev[start] in digits:

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

657 start += 1

658 found_digit = True

659 else:

660 break

661 # END handle number

662 # END number parse loop

663

664 if not found_digit:

665 num = 1

666 # END set default num

667

668 try:

669 if token == "~":

670 obj = to_commit(obj)

671 for _ in range(num):

672 obj = obj.parents[0]

673 # END for each history item to walk

674 elif token == "^":

675 obj = to_commit(obj)

676 if num == 0:

677 pass

678 else:

679 obj = obj.parents[num - 1]

680 # END handle parent

681 else:

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

683 # END end handle tag

684 except (IndexError, AttributeError) as e:

685 raise BadName(

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

687 ) from e

688 # END exception handling

689 # END parse loop

690

691 if obj is None:

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

693

694 return obj