Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/worktree.py: 25%

1# worktree.py -- Working tree operations for Git repositories

4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later

5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU

6# General Public License as published by the Free Software Foundation; version 2.0

7# or (at your option) any later version. You can redistribute it and/or

8# modify it under the terms of either of these two licenses.

10# Unless required by applicable law or agreed to in writing, software

11# distributed under the License is distributed on an "AS IS" BASIS,

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

13# See the License for the specific language governing permissions and

14# limitations under the License.

15#

16# You should have received a copy of the licenses; if not, see

17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License

18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache

19# License, Version 2.0.

20#

22"""Working tree operations for Git repositories."""

24from __future__ import annotations

26__all__ = [

27 "WorkTree",

28 "WorkTreeContainer",

29 "WorkTreeInfo",

30 "add_worktree",

31 "list_worktrees",

32 "lock_worktree",

33 "move_worktree",

34 "prune_worktrees",

35 "read_worktree_lock_reason",

36 "remove_worktree",

37 "repair_worktree",

38 "temporary_worktree",

39 "unlock_worktree",

40]

42import builtins

43import os

44import shutil

45import stat

46import tempfile

47import time

48import warnings

49from collections.abc import Callable, Iterable, Iterator, Sequence

50from contextlib import contextmanager

51from pathlib import Path

52from typing import TYPE_CHECKING, Any

54from .errors import CommitError, HookError

55from .objects import Blob, Commit, ObjectID, Tag, Tree

57if TYPE_CHECKING:

58 from .config import Config

59from .refs import SYMREF, Ref, local_branch_name

60from .repo import (

61 GITDIR,

62 WORKTREES,

63 Repo,

64 check_user_identity,

65 get_user_identity,

66)

67from .trailers import add_trailer_to_message

70def _should_use_relative_paths(

71 repo: Repo,

72 relative_paths: bool | None,

73 existing_path: bytes | None = None,

74) -> bool:

75 """Determine whether to use relative paths for gitdir references.

77 Args:

78 repo: The repository

79 relative_paths: Explicit preference (True/False) or None to check config

80 existing_path: Optional existing path to check format (for preserving format)

82 Returns:

83 True if relative paths should be used, False otherwise

84 """

85 if relative_paths is not None:

86 return relative_paths

88 # Check config

89 config = repo.get_config()

90 try:

91 use_relative = config.get_boolean(

92 (b"worktree",), b"useRelativePaths", default=False

93 )

94 if use_relative:

95 return True

96 except KeyError:

97 pass

99 # Preserve existing format if available

100 if existing_path is not None:

101 return not os.path.isabs(os.fsdecode(existing_path))

102

103 return False

104

105

106def _compute_gitdir_path(

107 repo: Repo,

108 gitdir_file: str,

109 worktree_control_dir: str,

110 use_relative: bool,

111) -> str:

112 """Compute the gitdir path and enable extension if needed.

113

114 Args:

115 repo: The repository

116 gitdir_file: Absolute path to the .git file

117 worktree_control_dir: Absolute path to the worktree control directory

118 use_relative: Whether to use relative paths

119

120 Returns:

121 The path to write (relative or absolute)

122 """

123 if use_relative:

124 from .repo import _enable_relative_worktrees_extension

125

126 _enable_relative_worktrees_extension(repo)

127 return os.path.relpath(gitdir_file, worktree_control_dir)

128 else:

129 return gitdir_file

130

131

132class WorkTreeInfo:

133 """Information about a single worktree.

134

135 Attributes:

136 path: Path to the worktree

137 head: Current HEAD commit SHA

138 branch: Current branch (if not detached)

139 bare: Whether this is a bare repository

140 detached: Whether HEAD is detached

141 locked: Whether the worktree is locked

142 prunable: Whether the worktree can be pruned

143 lock_reason: Reason for locking (if locked)

144 """

145

146 def __init__(

147 self,

148 path: str,

149 head: bytes | None = None,

150 branch: Ref | None = None,

151 bare: bool = False,

152 detached: bool = False,

153 locked: bool = False,

154 prunable: bool = False,

155 lock_reason: str | None = None,

156 ):

157 """Initialize WorkTreeInfo.

158

159 Args:

160 path: Path to the worktree

161 head: Current HEAD commit SHA

162 branch: Current branch (if not detached)

163 bare: Whether this is a bare repository

164 detached: Whether HEAD is detached

165 locked: Whether the worktree is locked

166 prunable: Whether the worktree can be pruned

167 lock_reason: Reason for locking (if locked)

168 """

169 self.path = path

170 self.head = head

171 self.branch = branch

172 self.bare = bare

173 self.detached = detached

174 self.locked = locked

175 self.prunable = prunable

176 self.lock_reason = lock_reason

177

178 def __repr__(self) -> str:

179 """Return string representation of WorkTreeInfo."""

180 return f"WorkTreeInfo(path={self.path!r}, branch={self.branch!r}, detached={self.detached})"

181

182 def __eq__(self, other: object) -> bool:

183 """Check equality with another WorkTreeInfo."""

184 if not isinstance(other, WorkTreeInfo):

185 return NotImplemented

186 return (

187 self.path == other.path

188 and self.head == other.head

189 and self.branch == other.branch

190 and self.bare == other.bare

191 and self.detached == other.detached

192 and self.locked == other.locked

193 and self.prunable == other.prunable

194 and self.lock_reason == other.lock_reason

195 )

196

197 def open(self) -> WorkTree:

198 """Open this worktree as a WorkTree.

199

200 Returns:

201 WorkTree object for this worktree

202

203 Raises:

204 NotGitRepository: If the worktree path is invalid

205 """

206 from .repo import Repo

207

208 repo = Repo(self.path)

209 return WorkTree(repo, self.path)

210

211

212class WorkTreeContainer:

213 """Container for managing multiple working trees.

214

215 This class manages worktrees for a repository, similar to how

216 RefsContainer manages references.

217 """

218

219 def __init__(self, repo: Repo) -> None:

220 """Initialize a WorkTreeContainer for the given repository.

221

222 Args:

223 repo: The repository this container belongs to

224 """

225 self._repo = repo

226

227 def list(self) -> list[WorkTreeInfo]:

228 """List all worktrees for this repository.

229

230 Returns:

231 A list of WorkTreeInfo objects

232 """

233 return list_worktrees(self._repo)

234

235 def add(

236 self,

237 path: str | bytes | os.PathLike[str],

238 branch: str | bytes | None = None,

239 commit: ObjectID | None = None,

240 force: bool = False,

241 detach: bool = False,

242 exist_ok: bool = False,

243 relative_paths: bool | None = None,

244 ) -> Repo:

245 """Add a new worktree.

246

247 Args:

248 path: Path where the new worktree should be created

249 branch: Branch to checkout in the new worktree

250 commit: Specific commit to checkout (results in detached HEAD)

251 force: Force creation even if branch is already checked out elsewhere

252 detach: Detach HEAD in the new worktree

253 exist_ok: If True, do not raise an error if the directory already exists

254 relative_paths: If True, use relative paths for gitdir references.

255 If None, check worktree.useRelativePaths config (defaults to False)

256

257 Returns:

258 The newly created worktree repository

259 """

260 return add_worktree(

261 self._repo,

262 path,

263 branch=branch,

264 commit=commit,

265 force=force,

266 detach=detach,

267 exist_ok=exist_ok,

268 relative_paths=relative_paths,

269 )

270

271 def remove(self, path: str | bytes | os.PathLike[str], force: bool = False) -> None:

272 """Remove a worktree.

273

274 Args:

275 path: Path to the worktree to remove

276 force: Force removal even if there are local changes

277 """

278 remove_worktree(self._repo, path, force=force)

279

280 def prune(

281 self, expire: int | None = None, dry_run: bool = False

282 ) -> builtins.list[str]:

283 """Prune worktree administrative files for missing worktrees.

284

285 Args:

286 expire: Only prune worktrees older than this many seconds

287 dry_run: Don't actually remove anything, just report what would be removed

288

289 Returns:

290 List of pruned worktree identifiers

291 """

292 return prune_worktrees(self._repo, expire=expire, dry_run=dry_run)

293

294 def move(

295 self,

296 old_path: str | bytes | os.PathLike[str],

297 new_path: str | bytes | os.PathLike[str],

298 relative_paths: bool | None = None,

299 ) -> None:

300 """Move a worktree to a new location.

301

302 Args:

303 old_path: Current path of the worktree

304 new_path: New path for the worktree

305 relative_paths: If True, use relative paths for gitdir references.

306 If None, check worktree.useRelativePaths config or preserve existing format

307 """

308 move_worktree(self._repo, old_path, new_path, relative_paths=relative_paths)

309

310 def lock(

311 self, path: str | bytes | os.PathLike[str], reason: str | None = None

312 ) -> None:

313 """Lock a worktree to prevent it from being pruned.

314

315 Args:

316 path: Path to the worktree to lock

317 reason: Optional reason for locking

318 """

319 lock_worktree(self._repo, path, reason=reason)

320

321 def unlock(self, path: str | bytes | os.PathLike[str]) -> None:

322 """Unlock a worktree.

323

324 Args:

325 path: Path to the worktree to unlock

326 """

327 unlock_worktree(self._repo, path)

328

329 def repair(

330 self,

331 paths: Sequence[str | bytes | os.PathLike[str]] | None = None,

332 relative_paths: bool | None = None,

333 ) -> builtins.list[str]:

334 """Repair worktree administrative files.

335

336 Args:

337 paths: Optional list of worktree paths to repair. If None, repairs

338 connections from the main repository to all linked worktrees.

339 relative_paths: If True, use relative paths for gitdir references.

340 If None, check worktree.useRelativePaths config or preserve existing format

341

342 Returns:

343 List of repaired worktree paths

344 """

345 return repair_worktree(self._repo, paths=paths, relative_paths=relative_paths)

346

347 def __iter__(self) -> Iterator[WorkTreeInfo]:

348 """Iterate over all worktrees."""

349 yield from self.list()

350

351

352class WorkTree:

353 """Working tree operations for a Git repository.

354

355 This class provides methods for working with the working tree,

356 such as staging files, committing changes, and resetting the index.

357 """

358

359 def __init__(self, repo: Repo, path: str | bytes | os.PathLike[str]) -> None:

360 """Initialize a WorkTree for the given repository.

361

362 Args:

363 repo: The repository this working tree belongs to

364 path: Path to the working tree directory

365 """

366 self._repo = repo

367 raw_path = os.fspath(path)

368 if isinstance(raw_path, bytes):

369 self.path: str = os.fsdecode(raw_path)

370 else:

371 self.path = raw_path

372 self.path = os.path.abspath(self.path)

373

374 def stage(

375 self,

376 fs_paths: str

377 | bytes

378 | os.PathLike[str]

379 | Iterable[str | bytes | os.PathLike[str]],

380 config: Config | None = None,

381 ) -> None:

382 """Stage a set of paths.

383

384 Args:

385 fs_paths: List of paths, relative to the repository path

386 config: Repository configuration. If None, falls back to

387 ``self._repo.get_config_stack()``.

388 """

389 if config is None:

390 config = self._repo.get_config_stack()

391 root_path_bytes = os.fsencode(self.path)

392

393 if isinstance(fs_paths, str | bytes | os.PathLike):

394 fs_paths = [fs_paths]

395 fs_paths = list(fs_paths)

396

397 from .index import (

398 EXTENDED_FLAG_INTEND_TO_ADD,

399 FLAG_EXTENDED,

400 _fs_to_tree_path,

401 blob_from_path_and_stat,

402 cleanup_mode,

403 index_entry_from_directory,

404 index_entry_from_stat,

405 unchanged_entry_flags,

406 )

407

408 index = self._repo.open_index(config=config)

409 blob_normalizer = self._repo.get_blob_normalizer(config=config)

410 for fs_path in fs_paths:

411 if not isinstance(fs_path, bytes):

412 fs_path = os.fsencode(fs_path)

413 if os.path.isabs(fs_path):

414 raise ValueError(

415 f"path {fs_path!r} should be relative to "

416 "repository root, not absolute"

417 )

418 tree_path = _fs_to_tree_path(fs_path)

419 full_path = os.path.join(root_path_bytes, fs_path)

420 try:

421 st = os.lstat(full_path)

422 except (FileNotFoundError, NotADirectoryError):

423 # File no longer exists

424 try:

425 del index[tree_path]

426 except KeyError:

427 pass # already removed

428 else:

429 if stat.S_ISDIR(st.st_mode):

430 entry = index_entry_from_directory(st, full_path)

431 if entry:

432 index[tree_path] = entry

433 else:

434 try:

435 del index[tree_path]

436 except KeyError:

437 pass

438 elif not stat.S_ISREG(st.st_mode) and not stat.S_ISLNK(st.st_mode):

439 try:

440 del index[tree_path]

441 except KeyError:

442 pass

443 else:

444 blob = blob_from_path_and_stat(full_path, st)

445 blob = blob_normalizer.checkin_normalize(blob, fs_path)

446 self._repo.object_store.add_object(blob)

447 blob_id = ObjectID(blob.get_id(self._repo.object_format))

448 flags, extended_flags = unchanged_entry_flags(

449 index, tree_path, blob_id, cleanup_mode(st.st_mode)

450 )

451 # Staging the content ends any intent-to-add.

452 extended_flags &= ~EXTENDED_FLAG_INTEND_TO_ADD

453 if not extended_flags:

454 flags &= ~FLAG_EXTENDED

455 index[tree_path] = index_entry_from_stat(

456 st, blob_id, flags=flags, extended_flags=extended_flags

457 )

458 index.write()

459

460 def unstage(

461 self,

462 fs_paths: Sequence[str],

463 config: Config | None = None,

464 ) -> None:

465 """Unstage specific file in the index.

466

467 Args:

468 fs_paths: a list of files to unstage,

469 relative to the repository path.

470 config: Repository configuration. If None, falls back to

471 ``self._repo.get_config_stack()``.

472 """

473 if config is None:

474 config = self._repo.get_config_stack()

475 from .index import IndexEntry, _fs_to_tree_path

476

477 index = self._repo.open_index(config=config)

478 try:

479 commit = self._repo[Ref(b"HEAD")]

480 except KeyError:

481 # no head mean no commit in the repo

482 for fs_path in fs_paths:

483 tree_path = _fs_to_tree_path(fs_path)

484 del index[tree_path]

485 index.write()

486 return

487 else:

488 assert isinstance(commit, Commit), "HEAD must be a commit"

489 tree_id = commit.tree

490

491 for fs_path in fs_paths:

492 tree_path = _fs_to_tree_path(fs_path)

493 try:

494 tree = self._repo.object_store[tree_id]

495 assert isinstance(tree, Tree)

496 tree_entry = tree.lookup_path(

497 self._repo.object_store.__getitem__, tree_path

498 )

499 except KeyError:

500 # if tree_entry didn't exist, this file was being added, so

501 # remove index entry

502 try:

503 del index[tree_path]

504 continue

505 except KeyError as exc:

506 raise KeyError(f"file '{tree_path.decode()}' not in index") from exc

507

508 st = None

509 try:

510 st = os.lstat(os.path.join(self.path, fs_path))

511 except FileNotFoundError:

512 pass

513

514 blob_obj = self._repo[tree_entry[1]]

515 assert isinstance(blob_obj, Blob)

516 blob_size = len(blob_obj.data)

517

518 index_entry = IndexEntry(

519 ctime=(commit.commit_time, 0),

520 mtime=(commit.commit_time, 0),

521 dev=st.st_dev if st else 0,

522 ino=st.st_ino if st else 0,

523 mode=tree_entry[0],

524 uid=st.st_uid if st else 0,

525 gid=st.st_gid if st else 0,

526 size=blob_size,

527 sha=tree_entry[1],

528 flags=0,

529 extended_flags=0,

530 )

531

532 index[tree_path] = index_entry

533 index.write()

534

535 def commit(

536 self,

537 message: str | bytes | Callable[[Any, Commit], bytes] | None = None,

538 committer: bytes | None = None,

539 author: bytes | None = None,

540 commit_timestamp: float | None = None,

541 commit_timezone: int | None = None,

542 author_timestamp: float | None = None,

543 author_timezone: int | None = None,

544 tree: ObjectID | None = None,

545 encoding: bytes | None = None,

546 ref: Ref | None = Ref(b"HEAD"),

547 merge_heads: Sequence[ObjectID] | None = None,

548 no_verify: bool = False,

549 sign: bool | None = None,

550 signoff: bool | None = None,

551 config: Config | None = None,

552 ) -> ObjectID:

553 """Create a new commit.

554

555 If not specified, committer and author default to

556 get_user_identity(..., 'COMMITTER')

557 and get_user_identity(..., 'AUTHOR') respectively.

558

559 Args:

560 message: Commit message (bytes or callable that takes (repo, commit)

561 and returns bytes)

562 committer: Committer fullname

563 author: Author fullname

564 commit_timestamp: Commit timestamp (defaults to now)

565 commit_timezone: Commit timestamp timezone (defaults to GMT)

566 author_timestamp: Author timestamp (defaults to commit

567 timestamp)

568 author_timezone: Author timestamp timezone

569 (defaults to commit timestamp timezone)

570 tree: SHA1 of the tree root to use (if not specified the

571 current index will be committed).

572 encoding: Encoding

573 ref: Optional ref to commit to (defaults to current branch).

574 If None, creates a dangling commit without updating any ref.

575 merge_heads: Merge heads (defaults to .git/MERGE_HEAD)

576 no_verify: Skip pre-commit and commit-msg hooks

577 sign: GPG Sign the commit (bool, defaults to False,

578 pass True to use default GPG key,

579 pass a str containing Key ID to use a specific GPG key)

580 signoff: Add Signed-off-by line (DCO) to commit message.

581 If None, uses format.signoff config.

582 config: Configuration to consult for committer/author identity and

583 other commit-time settings. If None, falls back to

584 ``self._repo.get_config_stack()``.

585

586 Returns:

587 New commit SHA1

588 """

589 try:

590 if not no_verify:

591 self._repo.hooks["pre-commit"].execute()

592 except HookError as exc:

593 raise CommitError(exc) from exc

594 except KeyError: # no hook defined, silent fallthrough

595 pass

596

597 if config is None:

598 config = self._repo.get_config_stack()

599

600 object_format = self._repo.object_format

601 c = Commit()

602 if tree is None:

603 index = self._repo.open_index(config=config)

604 c.tree = index.commit(self._repo.object_store)

605 else:

606 if len(tree) != object_format.hex_length:

607 raise ValueError(

608 f"tree must be a {object_format.hex_length}-byte hex sha string"

609 )

610 c.tree = tree

611

612 if merge_heads is None:

613 merge_heads = self._repo._read_heads("MERGE_HEAD")

614 if committer is None:

615 committer = get_user_identity(config, kind="COMMITTER")

616 check_user_identity(committer)

617 c.committer = committer

618 if commit_timestamp is None:

619 # FIXME: Support GIT_COMMITTER_DATE environment variable

620 commit_timestamp = time.time()

621 c.commit_time = int(commit_timestamp)

622 if commit_timezone is None:

623 # FIXME: Use current user timezone rather than UTC

624 commit_timezone = 0

625 c.commit_timezone = commit_timezone

626 if author is None:

627 author = get_user_identity(config, kind="AUTHOR")

628 c.author = author

629 check_user_identity(author)

630 if author_timestamp is None:

631 # FIXME: Support GIT_AUTHOR_DATE environment variable

632 author_timestamp = commit_timestamp

633 c.author_time = int(author_timestamp)

634 if author_timezone is None:

635 author_timezone = commit_timezone

636 c.author_timezone = author_timezone

637 if encoding is None:

638 try:

639 encoding = config.get(("i18n",), "commitEncoding")

640 except KeyError:

641 pass # No dice

642 if encoding is not None:

643 c.encoding = encoding

644 # Store original message (might be callable)

645 original_message = message

646 message = None # Will be set later after parents are set

647

648 # Check if we should sign the commit

649 if sign is None:

650 # Check commit.gpgSign configuration when sign is not explicitly set

651 try:

652 should_sign = config.get_boolean(

653 (b"commit",), b"gpgsign", default=False

654 )

655 except KeyError:

656 should_sign = False # Default to not signing if no config

657 else:

658 should_sign = sign

659

660 # Get the signing key from config if signing is enabled

661 keyid = None

662 if should_sign:

663 try:

664 keyid_bytes = config.get((b"user",), b"signingkey")

665 keyid = keyid_bytes.decode() if keyid_bytes else None

666 except KeyError:

667 keyid = None

668

669 if ref is None:

670 # Create a dangling commit

671 c.parents = list(merge_heads)

672 else:

673 try:

674 old_head = self._repo.refs[ref]

675 c.parents = [old_head, *merge_heads]

676 except KeyError:

677 c.parents = list(merge_heads)

678

679 # Handle message after parents are set

680 if callable(original_message):

681 message = original_message(self._repo, c)

682 if message is None:

683 raise ValueError("Message callback returned None")

684 else:

685 message = original_message

686

687 if message is None:

688 # FIXME: Try to read commit message from .git/MERGE_MSG

689 raise ValueError("No commit message specified")

690

691 # Handle signoff

692 should_signoff = signoff

693 if should_signoff is None:

694 # Check format.signOff configuration

695 try:

696 should_signoff = config.get_boolean(

697 (b"format",), b"signoff", default=False

698 )

699 except KeyError:

700 should_signoff = False

701

702 if should_signoff:

703 # Add Signed-off-by trailer

704 # Get the committer identity for the signoff

705 signoff_identity = committer

706 if isinstance(message, bytes):

707 message_bytes = message

708 else:

709 message_bytes = message.encode("utf-8")

710

711 message_bytes = add_trailer_to_message(

712 message_bytes,

713 "Signed-off-by",

714 signoff_identity.decode("utf-8")

715 if isinstance(signoff_identity, bytes)

716 else signoff_identity,

717 separator=":",

718 where="end",

719 if_exists="addIfDifferentNeighbor",

720 if_missing="add",

721 )

722 message = message_bytes

723

724 try:

725 if no_verify:

726 c.message = message

727 else:

728 c.message = self._repo.hooks["commit-msg"].execute(message)

729 if c.message is None:

730 c.message = message

731 except HookError as exc:

732 raise CommitError(exc) from exc

733 except KeyError: # no hook defined, message not modified

734 c.message = message

735

736 if ref is None:

737 # Create a dangling commit

738 if should_sign:

739 from dulwich.signature import get_signature_vendor

740

741 vendor = get_signature_vendor(config=config)

742 c.gpgsig = vendor.sign(c.as_raw_string(), keyid=keyid)

743 self._repo.object_store.add_object(c)

744 commit_id = ObjectID(c.get_id(object_format))

745 else:

746 try:

747 old_head = self._repo.refs[ref]

748 if should_sign:

749 from dulwich.signature import get_signature_vendor

750

751 vendor = get_signature_vendor(config=config)

752 c.gpgsig = vendor.sign(c.as_raw_string(), keyid=keyid)

753 self._repo.object_store.add_object(c)

754 commit_id = ObjectID(c.get_id(object_format))

755 message_bytes = (

756 message.encode() if isinstance(message, str) else message

757 )

758 ok = self._repo.refs.set_if_equals(

759 ref,

760 old_head,

761 commit_id,

762 message=b"commit: " + message_bytes,

763 committer=committer,

764 timestamp=int(commit_timestamp)

765 if commit_timestamp is not None

766 else None,

767 timezone=commit_timezone,

768 )

769 except KeyError:

770 c.parents = list(merge_heads)

771 if should_sign:

772 from dulwich.signature import get_signature_vendor

773

774 vendor = get_signature_vendor(config=config)

775 c.gpgsig = vendor.sign(c.as_raw_string(), keyid=keyid)

776 self._repo.object_store.add_object(c)

777 commit_id = ObjectID(c.get_id(object_format))

778 message_bytes = (

779 message.encode() if isinstance(message, str) else message

780 )

781 ok = self._repo.refs.add_if_new(

782 ref,

783 commit_id,

784 message=b"commit: " + message_bytes,

785 committer=committer,

786 timestamp=int(commit_timestamp)

787 if commit_timestamp is not None

788 else None,

789 timezone=commit_timezone,

790 )

791 if not ok:

792 # Fail if the atomic compare-and-swap failed, leaving the

793 # commit and all its objects as garbage.

794 raise CommitError(f"{ref!r} changed during commit")

795

796 self._repo._del_named_file("MERGE_HEAD")

797

798 try:

799 self._repo.hooks["post-commit"].execute()

800 except HookError as e: # silent failure

801 warnings.warn(f"post-commit hook failed: {e}", UserWarning)

802 except KeyError: # no hook defined, silent fallthrough

803 pass

804

805 # Trigger auto GC if needed

806 from .gc import maybe_auto_gc

807

808 maybe_auto_gc(self._repo)

809

810 return commit_id

811

812 def reset_index(

813 self,

814 tree: ObjectID | None = None,

815 config: Config | None = None,

816 ) -> None:

817 """Reset the index back to a specific tree.

818

819 Args:

820 tree: Tree SHA to reset to, None for current HEAD tree.

821 config: Stacked configuration used for filter setup. If None,

822 falls back to ``self._repo.get_config_stack()``.

823 """

824 if config is None:

825 config = self._repo.get_config_stack()

826 stacked_config = config

827 from .index import (

828 build_index_from_tree,

829 get_path_element_validator,

830 get_symlink_fn,

831 )

832

833 if tree is None:

834 head = self._repo[Ref(b"HEAD")]

835 if isinstance(head, Tag):

836 _cls, obj = head.object

837 head = self._repo.get_object(obj)

838 from .objects import Commit

839

840 assert isinstance(head, Commit)

841 tree = head.tree

842 assert tree is not None

843 config = self._repo.get_config()

844 honor_filemode = config.get_boolean(b"core", b"filemode", os.name != "nt")

845 validate_path_element = get_path_element_validator(config)

846 symlink_fn = get_symlink_fn(config)

847

848 blob_normalizer = self._repo.get_blob_normalizer(

849 config=stacked_config, tree=tree

850 )

851 return build_index_from_tree(

852 self.path,

853 self._repo.index_path(),

854 self._repo.object_store,

855 tree,

856 honor_filemode=honor_filemode,

857 validate_path_element=validate_path_element,

858 symlink_fn=symlink_fn,

859 blob_normalizer=blob_normalizer,

860 object_format=self._repo.object_format,

861 )

862

863 def _sparse_checkout_file_path(self) -> str:

864 """Return the path of the sparse-checkout file in this repo's control dir."""

865 return os.path.join(self._repo.controldir(), "info", "sparse-checkout")

866

867 def configure_for_cone_mode(self) -> None:

868 """Ensure the repository is configured for cone-mode sparse-checkout."""

869 config = self._repo.get_config()

870 config.set((b"core",), b"sparseCheckout", b"true")

871 config.set((b"core",), b"sparseCheckoutCone", b"true")

872 config.write_to_path()

873

874 def infer_cone_mode(self) -> bool:

875 """Return True if 'core.sparseCheckoutCone' is set to 'true' in config, else False."""

876 config = self._repo.get_config()

877 try:

878 sc_cone = config.get((b"core",), b"sparseCheckoutCone")

879 return sc_cone == b"true"

880 except KeyError:

881 # If core.sparseCheckoutCone is not set, default to False

882 return False

883

884 def get_sparse_checkout_patterns(self) -> list[str]:

885 """Return a list of sparse-checkout patterns from info/sparse-checkout.

886

887 Returns:

888 A list of patterns. Returns an empty list if the file is missing.

889 """

890 path = self._sparse_checkout_file_path()

891 try:

892 with open(path, encoding="utf-8") as f:

893 return [line.strip() for line in f if line.strip()]

894 except FileNotFoundError:

895 return []

896

897 def set_sparse_checkout_patterns(self, patterns: Sequence[str]) -> None:

898 """Write the given sparse-checkout patterns into info/sparse-checkout.

899

900 Creates the info/ directory if it does not exist.

901

902 Args:

903 patterns: A list of gitignore-style patterns to store.

904 """

905 info_dir = os.path.join(self._repo.controldir(), "info")

906 os.makedirs(info_dir, exist_ok=True)

907

908 path = self._sparse_checkout_file_path()

909 with open(path, "w", encoding="utf-8") as f:

910 for pat in patterns:

911 f.write(pat + "\n")

912

913 def set_cone_mode_patterns(self, dirs: Sequence[str] | None = None) -> None:

914 """Write the given cone-mode directory patterns into info/sparse-checkout.

915

916 For each directory to include, add an inclusion line that "undoes" the prior

917 ``!/*/`` 'exclude' that re-includes that directory and everything under it.

918 Never add the same line twice.

919 """

920 patterns = ["/*", "!/*/"]

921 if dirs:

922 for d in dirs:

923 d = d.strip("/")

924 line = f"/{d}/"

925 if d and line not in patterns:

926 patterns.append(line)

927 self.set_sparse_checkout_patterns(patterns)

928

929

930def read_worktree_lock_reason(worktree_path: str) -> str | None:

931 """Read the lock reason for a worktree.

932

933 Args:

934 worktree_path: Path to the worktree's administrative directory

935

936 Returns:

937 The lock reason if the worktree is locked, None otherwise

938 """

939 locked_path = os.path.join(worktree_path, "locked")

940 if not os.path.exists(locked_path):

941 return None

942

943 try:

944 with open(locked_path) as f:

945 return f.read().strip()

946 except (FileNotFoundError, PermissionError):

947 return None

948

949

950def list_worktrees(repo: Repo) -> list[WorkTreeInfo]:

951 """List all worktrees for the given repository.

952

953 Args:

954 repo: The repository to list worktrees for

955

956 Returns:

957 A list of WorkTreeInfo objects

958 """

959 worktrees = []

960

961 # Add main worktree

962 main_wt_info = WorkTreeInfo(

963 path=repo.path,

964 head=repo.head(),

965 bare=repo.bare,

966 detached=False,

967 locked=False,

968 prunable=False,

969 )

970

971 # Get branch info for main worktree

972 try:

973 with open(os.path.join(repo.controldir(), "HEAD"), "rb") as f:

974 head_contents = f.read().strip()

975 if head_contents.startswith(SYMREF):

976 ref_name = Ref(head_contents[len(SYMREF) :].strip())

977 main_wt_info.branch = ref_name

978 else:

979 main_wt_info.detached = True

980 main_wt_info.branch = None

981 except (FileNotFoundError, PermissionError):

982 main_wt_info.branch = None

983 main_wt_info.detached = True

984

985 worktrees.append(main_wt_info)

986

987 # List additional worktrees

988 worktrees_dir = os.path.join(repo.controldir(), WORKTREES)

989 if os.path.isdir(worktrees_dir):

990 for entry in os.listdir(worktrees_dir):

991 worktree_path = os.path.join(worktrees_dir, entry)

992 if not os.path.isdir(worktree_path):

993 continue

994

995 wt_info = WorkTreeInfo(

996 path="", # Will be set below

997 bare=False,

998 detached=False,

999 locked=False,

1000 prunable=False,

1001 )

1002

1003 # Read gitdir to get actual worktree path

1004 gitdir_path = os.path.join(worktree_path, GITDIR)

1005 try:

1006 with open(gitdir_path, "rb") as f:

1007 gitdir_contents = f.read().strip()

1008 # Convert relative path to absolute if needed

1009 wt_path = os.fsdecode(gitdir_contents)

1010 if not os.path.isabs(wt_path):

1011 wt_path = os.path.abspath(os.path.join(worktree_path, wt_path))

1012 wt_info.path = os.path.dirname(wt_path) # Remove .git suffix

1013 except (FileNotFoundError, PermissionError):

1014 # Worktree directory is missing, skip it

1015 # TODO: Consider adding these as prunable worktrees with a placeholder path

1016 continue

1017

1018 # Check if worktree path exists

1019 if wt_info.path and not os.path.exists(wt_info.path):

1020 wt_info.prunable = True

1021

1022 # Read HEAD

1023 head_path = os.path.join(worktree_path, "HEAD")

1024 try:

1025 with open(head_path, "rb") as f:

1026 head_contents = f.read().strip()

1027 if head_contents.startswith(SYMREF):

1028 ref_name = Ref(head_contents[len(SYMREF) :].strip())

1029 wt_info.branch = ref_name

1030 # Resolve ref to get commit sha

1031 try:

1032 wt_info.head = repo.refs[ref_name]

1033 except KeyError:

1034 wt_info.head = None

1035 else:

1036 wt_info.detached = True

1037 wt_info.branch = None

1038 wt_info.head = head_contents

1039 except (FileNotFoundError, PermissionError):

1040 wt_info.head = None

1041 wt_info.branch = None

1042

1043 # Check if locked

1044 lock_reason = read_worktree_lock_reason(worktree_path)

1045 if lock_reason is not None:

1046 wt_info.locked = True

1047 wt_info.lock_reason = lock_reason

1048

1049 worktrees.append(wt_info)

1050

1051 return worktrees

1052

1053

1054def add_worktree(

1055 repo: Repo,

1056 path: str | bytes | os.PathLike[str],

1057 branch: str | bytes | None = None,

1058 commit: ObjectID | None = None,

1059 force: bool = False,

1060 detach: bool = False,

1061 exist_ok: bool = False,

1062 relative_paths: bool | None = None,

1063) -> Repo:

1064 """Add a new worktree to the repository.

1065

1066 Args:

1067 repo: The main repository

1068 path: Path where the new worktree should be created

1069 branch: Branch to checkout in the new worktree (creates if doesn't exist)

1070 commit: Specific commit to checkout (results in detached HEAD)

1071 force: Force creation even if branch is already checked out elsewhere

1072 detach: Detach HEAD in the new worktree

1073 exist_ok: If True, do not raise an error if the directory already exists

1074 relative_paths: If True, use relative paths for gitdir references.

1075 If None, check worktree.useRelativePaths config (defaults to False)

1076

1077 Returns:

1078 The newly created worktree repository

1079

1080 Raises:

1081 ValueError: If the path already exists (and exist_ok is False) or branch is already checked out

1082 """

1083 from .repo import Repo as RepoClass

1084

1085 path = os.fspath(path)

1086 if isinstance(path, bytes):

1087 path = os.fsdecode(path)

1088

1089 # Determine whether to use relative paths

1090 use_relative = _should_use_relative_paths(repo, relative_paths)

1091

1092 # Check if path already exists

1093 if os.path.exists(path) and not exist_ok:

1094 raise ValueError(f"Path already exists: {path}")

1095

1096 # Normalize branch name

1097 if branch is not None:

1098 if isinstance(branch, str):

1099 branch = branch.encode()

1100 branch = local_branch_name(branch)

1101

1102 # Check if branch is already checked out in another worktree

1103 if branch and not force:

1104 for wt in list_worktrees(repo):

1105 if wt.branch == branch:

1106 raise ValueError(

1107 f"Branch {branch.decode()} is already checked out at {wt.path}"

1108 )

1109

1110 # Determine what to checkout

1111 if commit is not None:

1112 checkout_ref = commit

1113 detach = True

1114 elif branch is not None:

1115 # Check if branch exists

1116 try:

1117 checkout_ref = repo.refs[branch]

1118 except KeyError:

1119 if commit is None:

1120 # Create new branch from HEAD

1121 checkout_ref = repo.head()

1122 repo.refs[branch] = checkout_ref

1123 else:

1124 # Create new branch from specified commit

1125 checkout_ref = commit

1126 repo.refs[branch] = checkout_ref

1127 else:

1128 # Default to current HEAD

1129 checkout_ref = repo.head()

1130 detach = True

1131

1132 # Create the worktree directory

1133 os.makedirs(path, exist_ok=exist_ok)

1134

1135 # Initialize the worktree

1136 identifier = os.path.basename(path)

1137 wt_repo = RepoClass._init_new_working_directory(

1138 path, repo, identifier=identifier, relative_paths=use_relative

1139 )

1140

1141 # Set HEAD appropriately

1142 if detach:

1143 # Detached HEAD - write SHA directly to HEAD

1144 with open(os.path.join(wt_repo.controldir(), "HEAD"), "wb") as f:

1145 f.write(checkout_ref + b"\n")

1146 else:

1147 # Point to branch

1148 assert branch is not None # Should be guaranteed by logic above

1149 from dulwich.refs import HEADREF

1150

1151 wt_repo.refs.set_symbolic_ref(HEADREF, branch)

1152

1153 # Reset index to match HEAD

1154 wt_repo.get_worktree().reset_index(config=wt_repo.get_config_stack())

1155

1156 return wt_repo

1157

1158

1159def remove_worktree(

1160 repo: Repo, path: str | bytes | os.PathLike[str], force: bool = False

1161) -> None:

1162 """Remove a worktree.

1163

1164 Args:

1165 repo: The main repository

1166 path: Path to the worktree to remove

1167 force: Force removal even if there are local changes

1168

1169 Raises:

1170 ValueError: If the worktree doesn't exist, has local changes, or is locked

1171 """

1172 path = os.fspath(path)

1173 if isinstance(path, bytes):

1174 path = os.fsdecode(path)

1175

1176 # Don't allow removing the main worktree

1177 if os.path.abspath(path) == os.path.abspath(repo.path):

1178 raise ValueError("Cannot remove the main working tree")

1179

1180 # Find the worktree

1181 worktree_found = False

1182 worktree_id = None

1183 worktrees_dir = os.path.join(repo.controldir(), WORKTREES)

1184

1185 if os.path.isdir(worktrees_dir):

1186 for entry in os.listdir(worktrees_dir):

1187 worktree_path = os.path.join(worktrees_dir, entry)

1188 gitdir_path = os.path.join(worktree_path, GITDIR)

1189

1190 try:

1191 with open(gitdir_path, "rb") as f:

1192 gitdir_contents = f.read().strip()

1193 wt_path = os.fsdecode(gitdir_contents)

1194 if not os.path.isabs(wt_path):

1195 wt_path = os.path.abspath(os.path.join(worktree_path, wt_path))

1196 wt_dir = os.path.dirname(wt_path) # Remove .git suffix

1197

1198 if os.path.abspath(wt_dir) == os.path.abspath(path):

1199 worktree_found = True

1200 worktree_id = entry

1201 break

1202 except (FileNotFoundError, PermissionError):

1203 continue

1204

1205 if not worktree_found:

1206 raise ValueError(f"Worktree not found: {path}")

1207

1208 assert worktree_id is not None # Should be set if worktree_found is True

1209 worktree_control_dir = os.path.join(worktrees_dir, worktree_id)

1210

1211 # Check if locked

1212 if os.path.exists(os.path.join(worktree_control_dir, "locked")):

1213 if not force:

1214 raise ValueError(f"Worktree is locked: {path}")

1215

1216 # Check for local changes if not forcing

1217 if not force and os.path.exists(path):

1218 # TODO: Check for uncommitted changes in the worktree

1219 pass

1220

1221 # Remove the working directory

1222 if os.path.exists(path):

1223 shutil.rmtree(path)

1224

1225 # Remove the administrative files

1226 shutil.rmtree(worktree_control_dir)

1227

1228

1229def prune_worktrees(

1230 repo: Repo, expire: int | None = None, dry_run: bool = False

1231) -> list[str]:

1232 """Prune worktree administrative files for missing worktrees.

1233

1234 Args:

1235 repo: The main repository

1236 expire: Only prune worktrees older than this many seconds

1237 dry_run: Don't actually remove anything, just report what would be removed

1238

1239 Returns:

1240 List of pruned worktree identifiers

1241 """

1242 pruned: list[str] = []

1243 worktrees_dir = os.path.join(repo.controldir(), WORKTREES)

1244

1245 if not os.path.isdir(worktrees_dir):

1246 return pruned

1247

1248 current_time = time.time()

1249

1250 for entry in os.listdir(worktrees_dir):

1251 worktree_path = os.path.join(worktrees_dir, entry)

1252 if not os.path.isdir(worktree_path):

1253 continue

1254

1255 # Skip locked worktrees

1256 if os.path.exists(os.path.join(worktree_path, "locked")):

1257 continue

1258

1259 should_prune = False

1260

1261 # Check if gitdir exists and points to valid location

1262 gitdir_path = os.path.join(worktree_path, GITDIR)

1263 try:

1264 with open(gitdir_path, "rb") as f:

1265 gitdir_contents = f.read().strip()

1266 wt_path = os.fsdecode(gitdir_contents)

1267 if not os.path.isabs(wt_path):

1268 wt_path = os.path.abspath(os.path.join(worktree_path, wt_path))

1269 wt_dir = os.path.dirname(wt_path) # Remove .git suffix

1270

1271 if not os.path.exists(wt_dir):

1272 should_prune = True

1273 except (FileNotFoundError, PermissionError):

1274 should_prune = True

1275

1276 # Check expiry time if specified

1277 if should_prune and expire is not None:

1278 stat_info = os.stat(worktree_path)

1279 age = current_time - stat_info.st_mtime

1280 if age < expire:

1281 should_prune = False

1282

1283 if should_prune:

1284 pruned.append(entry)

1285 if not dry_run:

1286 shutil.rmtree(worktree_path)

1287

1288 return pruned

1289

1290

1291def lock_worktree(

1292 repo: Repo, path: str | bytes | os.PathLike[str], reason: str | None = None

1293) -> None:

1294 """Lock a worktree to prevent it from being pruned.

1295

1296 Args:

1297 repo: The main repository

1298 path: Path to the worktree to lock

1299 reason: Optional reason for locking

1300 """

1301 worktree_id = _find_worktree_id(repo, path)

1302 worktree_control_dir = os.path.join(repo.controldir(), WORKTREES, worktree_id)

1303

1304 lock_path = os.path.join(worktree_control_dir, "locked")

1305 with open(lock_path, "w") as f:

1306 if reason:

1307 f.write(reason)

1308

1309

1310def unlock_worktree(repo: Repo, path: str | bytes | os.PathLike[str]) -> None:

1311 """Unlock a worktree.

1312

1313 Args:

1314 repo: The main repository

1315 path: Path to the worktree to unlock

1316 """

1317 worktree_id = _find_worktree_id(repo, path)

1318 worktree_control_dir = os.path.join(repo.controldir(), WORKTREES, worktree_id)

1319

1320 lock_path = os.path.join(worktree_control_dir, "locked")

1321 if os.path.exists(lock_path):

1322 os.remove(lock_path)

1323

1324

1325def _find_worktree_id(repo: Repo, path: str | bytes | os.PathLike[str]) -> str:

1326 """Find the worktree identifier for the given path.

1327

1328 Args:

1329 repo: The main repository

1330 path: Path to the worktree

1331

1332 Returns:

1333 The worktree identifier

1334

1335 Raises:

1336 ValueError: If the worktree is not found

1337 """

1338 path = os.fspath(path)

1339 if isinstance(path, bytes):

1340 path = os.fsdecode(path)

1341

1342 worktrees_dir = os.path.join(repo.controldir(), WORKTREES)

1343

1344 if os.path.isdir(worktrees_dir):

1345 for entry in os.listdir(worktrees_dir):

1346 worktree_path = os.path.join(worktrees_dir, entry)

1347 gitdir_path = os.path.join(worktree_path, GITDIR)

1348

1349 try:

1350 with open(gitdir_path, "rb") as f:

1351 gitdir_contents = f.read().strip()

1352 wt_path = os.fsdecode(gitdir_contents)

1353 if not os.path.isabs(wt_path):

1354 wt_path = os.path.abspath(os.path.join(worktree_path, wt_path))

1355 wt_dir = os.path.dirname(wt_path) # Remove .git suffix

1356

1357 if os.path.abspath(wt_dir) == os.path.abspath(path):

1358 return entry

1359 except (FileNotFoundError, PermissionError):

1360 continue

1361

1362 raise ValueError(f"Worktree not found: {path}")

1363

1364

1365def move_worktree(

1366 repo: Repo,

1367 old_path: str | bytes | os.PathLike[str],

1368 new_path: str | bytes | os.PathLike[str],

1369 relative_paths: bool | None = None,

1370) -> None:

1371 """Move a worktree to a new location.

1372

1373 Args:

1374 repo: The main repository

1375 old_path: Current path of the worktree

1376 new_path: New path for the worktree

1377 relative_paths: If True, use relative paths for gitdir references.

1378 If None, check worktree.useRelativePaths config or preserve existing format

1379

1380 Raises:

1381 ValueError: If the worktree doesn't exist or new path already exists

1382 """

1383 old_path = os.fspath(old_path)

1384 new_path = os.fspath(new_path)

1385 if isinstance(old_path, bytes):

1386 old_path = os.fsdecode(old_path)

1387 if isinstance(new_path, bytes):

1388 new_path = os.fsdecode(new_path)

1389

1390 # Don't allow moving the main worktree

1391 if os.path.abspath(old_path) == os.path.abspath(repo.path):

1392 raise ValueError("Cannot move the main working tree")

1393

1394 # Check if new path already exists

1395 if os.path.exists(new_path):

1396 raise ValueError(f"Path already exists: {new_path}")

1397

1398 # Find the worktree

1399 worktree_id = _find_worktree_id(repo, old_path)

1400 worktree_control_dir = os.path.join(repo.controldir(), WORKTREES, worktree_id)

1401

1402 # Read existing path to check format

1403 existing_path = None

1404 try:

1405 with open(os.path.join(worktree_control_dir, GITDIR), "rb") as f:

1406 existing_path = f.read().strip()

1407 except (FileNotFoundError, PermissionError):

1408 pass

1409

1410 # Determine whether to use relative paths

1411 use_relative = _should_use_relative_paths(repo, relative_paths, existing_path)

1412

1413 # Move the actual worktree directory

1414 shutil.move(old_path, new_path)

1415

1416 # Update the gitdir file in the worktree

1417 gitdir_file_abs = os.path.abspath(os.path.join(new_path, ".git"))

1418

1419 # Compute the path to write

1420 gitdir_path = _compute_gitdir_path(

1421 repo, gitdir_file_abs, worktree_control_dir, use_relative

1422 )

1423

1424 # Update the gitdir pointer in the control directory

1425 with open(os.path.join(worktree_control_dir, GITDIR), "wb") as f:

1426 f.write(os.fsencode(gitdir_path) + b"\n")

1427

1428

1429def repair_worktree(

1430 repo: Repo,

1431 paths: Sequence[str | bytes | os.PathLike[str]] | None = None,

1432 relative_paths: bool | None = None,

1433) -> list[str]:

1434 """Repair worktree administrative files.

1435

1436 This repairs the connection between worktrees and the main repository

1437 when they have been moved or become corrupted.

1438

1439 Args:

1440 repo: The main repository

1441 paths: Optional list of worktree paths to repair. If None, repairs

1442 connections from the main repository to all linked worktrees.

1443 relative_paths: If True, use relative paths for gitdir references.

1444 If None, check worktree.useRelativePaths config or preserve existing format

1445

1446 Returns:

1447 List of repaired worktree paths

1448

1449 Raises:

1450 ValueError: If a specified path is not a valid worktree

1451 """

1452 repaired: list[str] = []

1453 worktrees_dir = os.path.join(repo.controldir(), WORKTREES)

1454

1455 if paths:

1456 # Repair specific worktrees

1457 for path in paths:

1458 path_str = os.fspath(path)

1459 if isinstance(path_str, bytes):

1460 path_str = os.fsdecode(path_str)

1461 path_str = os.path.abspath(path_str)

1462

1463 # Check if this is a linked worktree

1464 gitdir_file = os.path.join(path_str, ".git")

1465 if not os.path.exists(gitdir_file):

1466 raise ValueError(f"Not a valid worktree: {path_str}")

1467

1468 # Read the .git file to get the worktree control directory

1469 try:

1470 with open(gitdir_file, "rb") as f:

1471 gitdir_content = f.read().strip()

1472 if gitdir_content.startswith(b"gitdir: "):

1473 worktree_control_path = gitdir_content[8:].decode()

1474 else:

1475 raise ValueError(f"Invalid .git file in worktree: {path_str}")

1476 except (FileNotFoundError, PermissionError, UnicodeDecodeError) as e:

1477 raise ValueError(

1478 f"Cannot read .git file in worktree: {path_str}"

1479 ) from e

1480

1481 # Make the path absolute if it's relative

1482 if not os.path.isabs(worktree_control_path):

1483 worktree_control_path = os.path.abspath(

1484 os.path.join(path_str, worktree_control_path)

1485 )

1486

1487 # Update the gitdir file in the worktree control directory

1488 gitdir_pointer = os.path.join(worktree_control_path, GITDIR)

1489 if os.path.exists(gitdir_pointer):

1490 # Read existing path to check format

1491 existing_path = None

1492 try:

1493 with open(gitdir_pointer, "rb") as f:

1494 existing_path = f.read().strip()

1495 except (FileNotFoundError, PermissionError):

1496 pass

1497

1498 # Determine which format to use for this worktree

1499 use_relative = _should_use_relative_paths(

1500 repo, relative_paths, existing_path

1501 )

1502

1503 # Compute the path to write

1504 gitdir_path_to_write = _compute_gitdir_path(

1505 repo, gitdir_file, worktree_control_path, use_relative

1506 )

1507

1508 # Update to point to the current location

1509 with open(gitdir_pointer, "wb") as f:

1510 f.write(os.fsencode(gitdir_path_to_write) + b"\n")

1511 repaired.append(path_str)

1512 else:

1513 # Repair from main repository to all linked worktrees

1514 if not os.path.isdir(worktrees_dir):

1515 return repaired

1516

1517 for entry in os.listdir(worktrees_dir):

1518 worktree_control_path = os.path.join(worktrees_dir, entry)

1519 if not os.path.isdir(worktree_control_path):

1520 continue

1521

1522 # Read the gitdir file to find where the worktree thinks it is

1523 gitdir_path = os.path.join(worktree_control_path, GITDIR)

1524 try:

1525 with open(gitdir_path, "rb") as f:

1526 gitdir_contents = f.read().strip()

1527 old_gitdir_location = os.fsdecode(gitdir_contents)

1528 except (FileNotFoundError, PermissionError):

1529 # Can't repair if we can't read the gitdir file

1530 continue

1531

1532 # Get the worktree directory (remove .git suffix)

1533 old_worktree_path = os.path.dirname(old_gitdir_location)

1534

1535 # Check if the .git file exists at the old location

1536 if os.path.exists(old_gitdir_location):

1537 # Try to read and update the .git file to ensure it points back correctly

1538 try:

1539 with open(old_gitdir_location, "rb") as f:

1540 content = f.read().strip()

1541 if content.startswith(b"gitdir: "):

1542 current_pointer = content[8:].decode()

1543 if not os.path.isabs(current_pointer):

1544 current_pointer = os.path.abspath(

1545 os.path.join(old_worktree_path, current_pointer)

1546 )

1547

1548 # If it doesn't point to the right place, fix it

1549 expected_pointer = worktree_control_path

1550 if os.path.abspath(current_pointer) != os.path.abspath(

1551 expected_pointer

1552 ):

1553 # Determine which format to use

1554 use_relative = _should_use_relative_paths(

1555 repo, relative_paths, gitdir_contents

1556 )

1557

1558 # Compute the path to write (from worktree to control dir)

1559 pointer_to_write = _compute_gitdir_path(

1560 repo,

1561 worktree_control_path,

1562 old_worktree_path,

1563 use_relative,

1564 )

1565

1566 # Update the .git file to point to the correct location

1567 with open(old_gitdir_location, "wb") as wf:

1568 wf.write(

1569 b"gitdir: "

1570 + os.fsencode(pointer_to_write)

1571 + b"\n"

1572 )

1573 repaired.append(old_worktree_path)

1574 except (PermissionError, UnicodeDecodeError):

1575 continue

1576

1577 return repaired

1578

1579

1580@contextmanager

1581def temporary_worktree(repo: Repo, prefix: str = "tmp-worktree-") -> Iterator[Repo]:

1582 """Create a temporary worktree that is automatically cleaned up.

1583

1584 Args:

1585 repo: Dulwich repository object

1586 prefix: Prefix for the temporary directory name

1587

1588 Yields:

1589 Worktree object

1590 """

1591 temp_dir = None

1592 worktree = None

1593

1594 try:

1595 # Create temporary directory

1596 temp_dir = tempfile.mkdtemp(prefix=prefix)

1597

1598 # Add worktree

1599 worktree = repo.worktrees.add(temp_dir, exist_ok=True)

1600

1601 yield worktree

1602

1603 finally:

1604 # Clean up worktree registration

1605 if worktree:

1606 repo.worktrees.remove(worktree.path)

1607

1608 # Clean up temporary directory

1609 if temp_dir and Path(temp_dir).exists():

1610 shutil.rmtree(temp_dir)