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

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 _fs_to_tree_path,

399 blob_from_path_and_stat,

400 index_entry_from_directory,

401 index_entry_from_stat,

402 )

403

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

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

406 for fs_path in fs_paths:

407 if not isinstance(fs_path, bytes):

408 fs_path = os.fsencode(fs_path)

409 if os.path.isabs(fs_path):

410 raise ValueError(

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

412 "repository root, not absolute"

413 )

414 tree_path = _fs_to_tree_path(fs_path)

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

416 try:

417 st = os.lstat(full_path)

418 except (FileNotFoundError, NotADirectoryError):

419 # File no longer exists

420 try:

421 del index[tree_path]

422 except KeyError:

423 pass # already removed

424 else:

425 if stat.S_ISDIR(st.st_mode):

426 entry = index_entry_from_directory(st, full_path)

427 if entry:

428 index[tree_path] = entry

429 else:

430 try:

431 del index[tree_path]

432 except KeyError:

433 pass

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

435 try:

436 del index[tree_path]

437 except KeyError:

438 pass

439 else:

440 blob = blob_from_path_and_stat(full_path, st)

441 blob = blob_normalizer.checkin_normalize(blob, fs_path)

442 self._repo.object_store.add_object(blob)

443 index[tree_path] = index_entry_from_stat(st, blob.id)

444 index.write()

445

446 def unstage(

447 self,

448 fs_paths: Sequence[str],

449 config: Config | None = None,

450 ) -> None:

451 """Unstage specific file in the index.

452

453 Args:

454 fs_paths: a list of files to unstage,

455 relative to the repository path.

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

457 ``self._repo.get_config_stack()``.

458 """

459 if config is None:

460 config = self._repo.get_config_stack()

461 from .index import IndexEntry, _fs_to_tree_path

462

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

464 try:

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

466 except KeyError:

467 # no head mean no commit in the repo

468 for fs_path in fs_paths:

469 tree_path = _fs_to_tree_path(fs_path)

470 del index[tree_path]

471 index.write()

472 return

473 else:

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

475 tree_id = commit.tree

476

477 for fs_path in fs_paths:

478 tree_path = _fs_to_tree_path(fs_path)

479 try:

480 tree = self._repo.object_store[tree_id]

481 assert isinstance(tree, Tree)

482 tree_entry = tree.lookup_path(

483 self._repo.object_store.__getitem__, tree_path

484 )

485 except KeyError:

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

487 # remove index entry

488 try:

489 del index[tree_path]

490 continue

491 except KeyError as exc:

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

493

494 st = None

495 try:

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

497 except FileNotFoundError:

498 pass

499

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

501 assert isinstance(blob_obj, Blob)

502 blob_size = len(blob_obj.data)

503

504 index_entry = IndexEntry(

505 ctime=(commit.commit_time, 0),

506 mtime=(commit.commit_time, 0),

507 dev=st.st_dev if st else 0,

508 ino=st.st_ino if st else 0,

509 mode=tree_entry[0],

510 uid=st.st_uid if st else 0,

511 gid=st.st_gid if st else 0,

512 size=blob_size,

513 sha=tree_entry[1],

514 flags=0,

515 extended_flags=0,

516 )

517

518 index[tree_path] = index_entry

519 index.write()

520

521 def commit(

522 self,

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

524 committer: bytes | None = None,

525 author: bytes | None = None,

526 commit_timestamp: float | None = None,

527 commit_timezone: int | None = None,

528 author_timestamp: float | None = None,

529 author_timezone: int | None = None,

530 tree: ObjectID | None = None,

531 encoding: bytes | None = None,

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

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

534 no_verify: bool = False,

535 sign: bool | None = None,

536 signoff: bool | None = None,

537 config: Config | None = None,

538 ) -> ObjectID:

539 """Create a new commit.

540

541 If not specified, committer and author default to

542 get_user_identity(..., 'COMMITTER')

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

544

545 Args:

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

547 and returns bytes)

548 committer: Committer fullname

549 author: Author fullname

550 commit_timestamp: Commit timestamp (defaults to now)

551 commit_timezone: Commit timestamp timezone (defaults to GMT)

552 author_timestamp: Author timestamp (defaults to commit

553 timestamp)

554 author_timezone: Author timestamp timezone

555 (defaults to commit timestamp timezone)

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

557 current index will be committed).

558 encoding: Encoding

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

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

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

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

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

564 pass True to use default GPG key,

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

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

567 If None, uses format.signoff config.

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

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

570 ``self._repo.get_config_stack()``.

571

572 Returns:

573 New commit SHA1

574 """

575 try:

576 if not no_verify:

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

578 except HookError as exc:

579 raise CommitError(exc) from exc

580 except KeyError: # no hook defined, silent fallthrough

581 pass

582

583 if config is None:

584 config = self._repo.get_config_stack()

585

586 c = Commit()

587 if tree is None:

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

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

590 else:

591 if len(tree) != 40:

592 raise ValueError("tree must be a 40-byte hex sha string")

593 c.tree = tree

594

595 if merge_heads is None:

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

597 if committer is None:

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

599 check_user_identity(committer)

600 c.committer = committer

601 if commit_timestamp is None:

602 # FIXME: Support GIT_COMMITTER_DATE environment variable

603 commit_timestamp = time.time()

604 c.commit_time = int(commit_timestamp)

605 if commit_timezone is None:

606 # FIXME: Use current user timezone rather than UTC

607 commit_timezone = 0

608 c.commit_timezone = commit_timezone

609 if author is None:

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

611 c.author = author

612 check_user_identity(author)

613 if author_timestamp is None:

614 # FIXME: Support GIT_AUTHOR_DATE environment variable

615 author_timestamp = commit_timestamp

616 c.author_time = int(author_timestamp)

617 if author_timezone is None:

618 author_timezone = commit_timezone

619 c.author_timezone = author_timezone

620 if encoding is None:

621 try:

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

623 except KeyError:

624 pass # No dice

625 if encoding is not None:

626 c.encoding = encoding

627 # Store original message (might be callable)

628 original_message = message

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

630

631 # Check if we should sign the commit

632 if sign is None:

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

634 try:

635 should_sign = config.get_boolean(

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

637 )

638 except KeyError:

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

640 else:

641 should_sign = sign

642

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

644 keyid = None

645 if should_sign:

646 try:

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

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

649 except KeyError:

650 keyid = None

651

652 if ref is None:

653 # Create a dangling commit

654 c.parents = list(merge_heads)

655 else:

656 try:

657 old_head = self._repo.refs[ref]

658 c.parents = [old_head, *merge_heads]

659 except KeyError:

660 c.parents = list(merge_heads)

661

662 # Handle message after parents are set

663 if callable(original_message):

664 message = original_message(self._repo, c)

665 if message is None:

666 raise ValueError("Message callback returned None")

667 else:

668 message = original_message

669

670 if message is None:

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

672 raise ValueError("No commit message specified")

673

674 # Handle signoff

675 should_signoff = signoff

676 if should_signoff is None:

677 # Check format.signOff configuration

678 try:

679 should_signoff = config.get_boolean(

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

681 )

682 except KeyError:

683 should_signoff = False

684

685 if should_signoff:

686 # Add Signed-off-by trailer

687 # Get the committer identity for the signoff

688 signoff_identity = committer

689 if isinstance(message, bytes):

690 message_bytes = message

691 else:

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

693

694 message_bytes = add_trailer_to_message(

695 message_bytes,

696 "Signed-off-by",

697 signoff_identity.decode("utf-8")

698 if isinstance(signoff_identity, bytes)

699 else signoff_identity,

700 separator=":",

701 where="end",

702 if_exists="addIfDifferentNeighbor",

703 if_missing="add",

704 )

705 message = message_bytes

706

707 try:

708 if no_verify:

709 c.message = message

710 else:

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

712 if c.message is None:

713 c.message = message

714 except HookError as exc:

715 raise CommitError(exc) from exc

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

717 c.message = message

718

719 if ref is None:

720 # Create a dangling commit

721 if should_sign:

722 from dulwich.signature import get_signature_vendor

723

724 vendor = get_signature_vendor(config=config)

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

726 self._repo.object_store.add_object(c)

727 else:

728 try:

729 old_head = self._repo.refs[ref]

730 if should_sign:

731 from dulwich.signature import get_signature_vendor

732

733 vendor = get_signature_vendor(config=config)

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

735 self._repo.object_store.add_object(c)

736 message_bytes = (

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

738 )

739 ok = self._repo.refs.set_if_equals(

740 ref,

741 old_head,

742 c.id,

743 message=b"commit: " + message_bytes,

744 committer=committer,

745 timestamp=int(commit_timestamp)

746 if commit_timestamp is not None

747 else None,

748 timezone=commit_timezone,

749 )

750 except KeyError:

751 c.parents = list(merge_heads)

752 if should_sign:

753 from dulwich.signature import get_signature_vendor

754

755 vendor = get_signature_vendor(config=config)

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

757 self._repo.object_store.add_object(c)

758 message_bytes = (

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

760 )

761 ok = self._repo.refs.add_if_new(

762 ref,

763 c.id,

764 message=b"commit: " + message_bytes,

765 committer=committer,

766 timestamp=int(commit_timestamp)

767 if commit_timestamp is not None

768 else None,

769 timezone=commit_timezone,

770 )

771 if not ok:

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

773 # commit and all its objects as garbage.

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

775

776 self._repo._del_named_file("MERGE_HEAD")

777

778 try:

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

780 except HookError as e: # silent failure

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

782 except KeyError: # no hook defined, silent fallthrough

783 pass

784

785 # Trigger auto GC if needed

786 from .gc import maybe_auto_gc

787

788 maybe_auto_gc(self._repo)

789

790 return c.id

791

792 def reset_index(

793 self,

794 tree: ObjectID | None = None,

795 config: Config | None = None,

796 ) -> None:

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

798

799 Args:

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

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

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

803 """

804 if config is None:

805 config = self._repo.get_config_stack()

806 stacked_config = config

807 from .index import (

808 build_index_from_tree,

809 get_path_element_validator,

810 symlink,

811 )

812

813 if tree is None:

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

815 if isinstance(head, Tag):

816 _cls, obj = head.object

817 head = self._repo.get_object(obj)

818 from .objects import Commit

819

820 assert isinstance(head, Commit)

821 tree = head.tree

822 assert tree is not None

823 config = self._repo.get_config()

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

825 validate_path_element = get_path_element_validator(config)

826 if config.get_boolean(b"core", b"symlinks", True):

827 symlink_fn = symlink

828 else:

829

830 def symlink_fn( # type: ignore[misc,unused-ignore]

831 src: str | bytes,

832 dst: str | bytes,

833 target_is_directory: bool = False,

834 *,

835 dir_fd: int | None = None,

836 ) -> None:

837 with open(dst, "w" + ("b" if isinstance(src, bytes) else "")) as f:

838 f.write(src)

839

840 blob_normalizer = self._repo.get_blob_normalizer(config=stacked_config)

841 return build_index_from_tree(

842 self.path,

843 self._repo.index_path(),

844 self._repo.object_store,

845 tree,

846 honor_filemode=honor_filemode,

847 validate_path_element=validate_path_element,

848 symlink_fn=symlink_fn, # type: ignore[arg-type,unused-ignore]

849 blob_normalizer=blob_normalizer,

850 )

851

852 def _sparse_checkout_file_path(self) -> str:

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

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

855

856 def configure_for_cone_mode(self) -> None:

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

858 config = self._repo.get_config()

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

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

861 config.write_to_path()

862

863 def infer_cone_mode(self) -> bool:

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

865 config = self._repo.get_config()

866 try:

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

868 return sc_cone == b"true"

869 except KeyError:

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

871 return False

872

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

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

875

876 Returns:

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

878 """

879 path = self._sparse_checkout_file_path()

880 try:

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

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

883 except FileNotFoundError:

884 return []

885

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

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

888

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

890

891 Args:

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

893 """

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

895 os.makedirs(info_dir, exist_ok=True)

896

897 path = self._sparse_checkout_file_path()

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

899 for pat in patterns:

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

901

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

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

904

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

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

907 Never add the same line twice.

908 """

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

910 if dirs:

911 for d in dirs:

912 d = d.strip("/")

913 line = f"/{d}/"

914 if d and line not in patterns:

915 patterns.append(line)

916 self.set_sparse_checkout_patterns(patterns)

917

918

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

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

921

922 Args:

923 worktree_path: Path to the worktree's administrative directory

924

925 Returns:

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

927 """

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

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

930 return None

931

932 try:

933 with open(locked_path) as f:

934 return f.read().strip()

935 except (FileNotFoundError, PermissionError):

936 return None

937

938

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

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

941

942 Args:

943 repo: The repository to list worktrees for

944

945 Returns:

946 A list of WorkTreeInfo objects

947 """

948 worktrees = []

949

950 # Add main worktree

951 main_wt_info = WorkTreeInfo(

952 path=repo.path,

953 head=repo.head(),

954 bare=repo.bare,

955 detached=False,

956 locked=False,

957 prunable=False,

958 )

959

960 # Get branch info for main worktree

961 try:

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

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

964 if head_contents.startswith(SYMREF):

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

966 main_wt_info.branch = ref_name

967 else:

968 main_wt_info.detached = True

969 main_wt_info.branch = None

970 except (FileNotFoundError, PermissionError):

971 main_wt_info.branch = None

972 main_wt_info.detached = True

973

974 worktrees.append(main_wt_info)

975

976 # List additional worktrees

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

978 if os.path.isdir(worktrees_dir):

979 for entry in os.listdir(worktrees_dir):

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

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

982 continue

983

984 wt_info = WorkTreeInfo(

985 path="", # Will be set below

986 bare=False,

987 detached=False,

988 locked=False,

989 prunable=False,

990 )

991

992 # Read gitdir to get actual worktree path

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

994 try:

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

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

997 # Convert relative path to absolute if needed

998 wt_path = os.fsdecode(gitdir_contents)

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

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

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

1002 except (FileNotFoundError, PermissionError):

1003 # Worktree directory is missing, skip it

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

1005 continue

1006

1007 # Check if worktree path exists

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

1009 wt_info.prunable = True

1010

1011 # Read HEAD

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

1013 try:

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

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

1016 if head_contents.startswith(SYMREF):

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

1018 wt_info.branch = ref_name

1019 # Resolve ref to get commit sha

1020 try:

1021 wt_info.head = repo.refs[ref_name]

1022 except KeyError:

1023 wt_info.head = None

1024 else:

1025 wt_info.detached = True

1026 wt_info.branch = None

1027 wt_info.head = head_contents

1028 except (FileNotFoundError, PermissionError):

1029 wt_info.head = None

1030 wt_info.branch = None

1031

1032 # Check if locked

1033 lock_reason = read_worktree_lock_reason(worktree_path)

1034 if lock_reason is not None:

1035 wt_info.locked = True

1036 wt_info.lock_reason = lock_reason

1037

1038 worktrees.append(wt_info)

1039

1040 return worktrees

1041

1042

1043def add_worktree(

1044 repo: Repo,

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

1046 branch: str | bytes | None = None,

1047 commit: ObjectID | None = None,

1048 force: bool = False,

1049 detach: bool = False,

1050 exist_ok: bool = False,

1051 relative_paths: bool | None = None,

1052) -> Repo:

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

1054

1055 Args:

1056 repo: The main repository

1057 path: Path where the new worktree should be created

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

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

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

1061 detach: Detach HEAD in the new worktree

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

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

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

1065

1066 Returns:

1067 The newly created worktree repository

1068

1069 Raises:

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

1071 """

1072 from .repo import Repo as RepoClass

1073

1074 path = os.fspath(path)

1075 if isinstance(path, bytes):

1076 path = os.fsdecode(path)

1077

1078 # Determine whether to use relative paths

1079 use_relative = _should_use_relative_paths(repo, relative_paths)

1080

1081 # Check if path already exists

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

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

1084

1085 # Normalize branch name

1086 if branch is not None:

1087 if isinstance(branch, str):

1088 branch = branch.encode()

1089 branch = local_branch_name(branch)

1090

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

1092 if branch and not force:

1093 for wt in list_worktrees(repo):

1094 if wt.branch == branch:

1095 raise ValueError(

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

1097 )

1098

1099 # Determine what to checkout

1100 if commit is not None:

1101 checkout_ref = commit

1102 detach = True

1103 elif branch is not None:

1104 # Check if branch exists

1105 try:

1106 checkout_ref = repo.refs[branch]

1107 except KeyError:

1108 if commit is None:

1109 # Create new branch from HEAD

1110 checkout_ref = repo.head()

1111 repo.refs[branch] = checkout_ref

1112 else:

1113 # Create new branch from specified commit

1114 checkout_ref = commit

1115 repo.refs[branch] = checkout_ref

1116 else:

1117 # Default to current HEAD

1118 checkout_ref = repo.head()

1119 detach = True

1120

1121 # Create the worktree directory

1122 os.makedirs(path, exist_ok=exist_ok)

1123

1124 # Initialize the worktree

1125 identifier = os.path.basename(path)

1126 wt_repo = RepoClass._init_new_working_directory(

1127 path, repo, identifier=identifier, relative_paths=use_relative

1128 )

1129

1130 # Set HEAD appropriately

1131 if detach:

1132 # Detached HEAD - write SHA directly to HEAD

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

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

1135 else:

1136 # Point to branch

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

1138 from dulwich.refs import HEADREF

1139

1140 wt_repo.refs.set_symbolic_ref(HEADREF, branch)

1141

1142 # Reset index to match HEAD

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

1144

1145 return wt_repo

1146

1147

1148def remove_worktree(

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

1150) -> None:

1151 """Remove a worktree.

1152

1153 Args:

1154 repo: The main repository

1155 path: Path to the worktree to remove

1156 force: Force removal even if there are local changes

1157

1158 Raises:

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

1160 """

1161 path = os.fspath(path)

1162 if isinstance(path, bytes):

1163 path = os.fsdecode(path)

1164

1165 # Don't allow removing the main worktree

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

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

1168

1169 # Find the worktree

1170 worktree_found = False

1171 worktree_id = None

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

1173

1174 if os.path.isdir(worktrees_dir):

1175 for entry in os.listdir(worktrees_dir):

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

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

1178

1179 try:

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

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

1182 wt_path = os.fsdecode(gitdir_contents)

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

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

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

1186

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

1188 worktree_found = True

1189 worktree_id = entry

1190 break

1191 except (FileNotFoundError, PermissionError):

1192 continue

1193

1194 if not worktree_found:

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

1196

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

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

1199

1200 # Check if locked

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

1202 if not force:

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

1204

1205 # Check for local changes if not forcing

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

1207 # TODO: Check for uncommitted changes in the worktree

1208 pass

1209

1210 # Remove the working directory

1211 if os.path.exists(path):

1212 shutil.rmtree(path)

1213

1214 # Remove the administrative files

1215 shutil.rmtree(worktree_control_dir)

1216

1217

1218def prune_worktrees(

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

1220) -> list[str]:

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

1222

1223 Args:

1224 repo: The main repository

1225 expire: Only prune worktrees older than this many seconds

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

1227

1228 Returns:

1229 List of pruned worktree identifiers

1230 """

1231 pruned: list[str] = []

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

1233

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

1235 return pruned

1236

1237 current_time = time.time()

1238

1239 for entry in os.listdir(worktrees_dir):

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

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

1242 continue

1243

1244 # Skip locked worktrees

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

1246 continue

1247

1248 should_prune = False

1249

1250 # Check if gitdir exists and points to valid location

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

1252 try:

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

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

1255 wt_path = os.fsdecode(gitdir_contents)

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

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

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

1259

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

1261 should_prune = True

1262 except (FileNotFoundError, PermissionError):

1263 should_prune = True

1264

1265 # Check expiry time if specified

1266 if should_prune and expire is not None:

1267 stat_info = os.stat(worktree_path)

1268 age = current_time - stat_info.st_mtime

1269 if age < expire:

1270 should_prune = False

1271

1272 if should_prune:

1273 pruned.append(entry)

1274 if not dry_run:

1275 shutil.rmtree(worktree_path)

1276

1277 return pruned

1278

1279

1280def lock_worktree(

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

1282) -> None:

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

1284

1285 Args:

1286 repo: The main repository

1287 path: Path to the worktree to lock

1288 reason: Optional reason for locking

1289 """

1290 worktree_id = _find_worktree_id(repo, path)

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

1292

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

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

1295 if reason:

1296 f.write(reason)

1297

1298

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

1300 """Unlock a worktree.

1301

1302 Args:

1303 repo: The main repository

1304 path: Path to the worktree to unlock

1305 """

1306 worktree_id = _find_worktree_id(repo, path)

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

1308

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

1310 if os.path.exists(lock_path):

1311 os.remove(lock_path)

1312

1313

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

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

1316

1317 Args:

1318 repo: The main repository

1319 path: Path to the worktree

1320

1321 Returns:

1322 The worktree identifier

1323

1324 Raises:

1325 ValueError: If the worktree is not found

1326 """

1327 path = os.fspath(path)

1328 if isinstance(path, bytes):

1329 path = os.fsdecode(path)

1330

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

1332

1333 if os.path.isdir(worktrees_dir):

1334 for entry in os.listdir(worktrees_dir):

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

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

1337

1338 try:

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

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

1341 wt_path = os.fsdecode(gitdir_contents)

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

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

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

1345

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

1347 return entry

1348 except (FileNotFoundError, PermissionError):

1349 continue

1350

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

1352

1353

1354def move_worktree(

1355 repo: Repo,

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

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

1358 relative_paths: bool | None = None,

1359) -> None:

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

1361

1362 Args:

1363 repo: The main repository

1364 old_path: Current path of the worktree

1365 new_path: New path for the worktree

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

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

1368

1369 Raises:

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

1371 """

1372 old_path = os.fspath(old_path)

1373 new_path = os.fspath(new_path)

1374 if isinstance(old_path, bytes):

1375 old_path = os.fsdecode(old_path)

1376 if isinstance(new_path, bytes):

1377 new_path = os.fsdecode(new_path)

1378

1379 # Don't allow moving the main worktree

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

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

1382

1383 # Check if new path already exists

1384 if os.path.exists(new_path):

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

1386

1387 # Find the worktree

1388 worktree_id = _find_worktree_id(repo, old_path)

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

1390

1391 # Read existing path to check format

1392 existing_path = None

1393 try:

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

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

1396 except (FileNotFoundError, PermissionError):

1397 pass

1398

1399 # Determine whether to use relative paths

1400 use_relative = _should_use_relative_paths(repo, relative_paths, existing_path)

1401

1402 # Move the actual worktree directory

1403 shutil.move(old_path, new_path)

1404

1405 # Update the gitdir file in the worktree

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

1407

1408 # Compute the path to write

1409 gitdir_path = _compute_gitdir_path(

1410 repo, gitdir_file_abs, worktree_control_dir, use_relative

1411 )

1412

1413 # Update the gitdir pointer in the control directory

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

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

1416

1417

1418def repair_worktree(

1419 repo: Repo,

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

1421 relative_paths: bool | None = None,

1422) -> list[str]:

1423 """Repair worktree administrative files.

1424

1425 This repairs the connection between worktrees and the main repository

1426 when they have been moved or become corrupted.

1427

1428 Args:

1429 repo: The main repository

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

1431 connections from the main repository to all linked worktrees.

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

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

1434

1435 Returns:

1436 List of repaired worktree paths

1437

1438 Raises:

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

1440 """

1441 repaired: list[str] = []

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

1443

1444 if paths:

1445 # Repair specific worktrees

1446 for path in paths:

1447 path_str = os.fspath(path)

1448 if isinstance(path_str, bytes):

1449 path_str = os.fsdecode(path_str)

1450 path_str = os.path.abspath(path_str)

1451

1452 # Check if this is a linked worktree

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

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

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

1456

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

1458 try:

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

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

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

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

1463 else:

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

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

1466 raise ValueError(

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

1468 ) from e

1469

1470 # Make the path absolute if it's relative

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

1472 worktree_control_path = os.path.abspath(

1473 os.path.join(path_str, worktree_control_path)

1474 )

1475

1476 # Update the gitdir file in the worktree control directory

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

1478 if os.path.exists(gitdir_pointer):

1479 # Read existing path to check format

1480 existing_path = None

1481 try:

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

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

1484 except (FileNotFoundError, PermissionError):

1485 pass

1486

1487 # Determine which format to use for this worktree

1488 use_relative = _should_use_relative_paths(

1489 repo, relative_paths, existing_path

1490 )

1491

1492 # Compute the path to write

1493 gitdir_path_to_write = _compute_gitdir_path(

1494 repo, gitdir_file, worktree_control_path, use_relative

1495 )

1496

1497 # Update to point to the current location

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

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

1500 repaired.append(path_str)

1501 else:

1502 # Repair from main repository to all linked worktrees

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

1504 return repaired

1505

1506 for entry in os.listdir(worktrees_dir):

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

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

1509 continue

1510

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

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

1513 try:

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

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

1516 old_gitdir_location = os.fsdecode(gitdir_contents)

1517 except (FileNotFoundError, PermissionError):

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

1519 continue

1520

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

1522 old_worktree_path = os.path.dirname(old_gitdir_location)

1523

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

1525 if os.path.exists(old_gitdir_location):

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

1527 try:

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

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

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

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

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

1533 current_pointer = os.path.abspath(

1534 os.path.join(old_worktree_path, current_pointer)

1535 )

1536

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

1538 expected_pointer = worktree_control_path

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

1540 expected_pointer

1541 ):

1542 # Determine which format to use

1543 use_relative = _should_use_relative_paths(

1544 repo, relative_paths, gitdir_contents

1545 )

1546

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

1548 pointer_to_write = _compute_gitdir_path(

1549 repo,

1550 worktree_control_path,

1551 old_worktree_path,

1552 use_relative,

1553 )

1554

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

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

1557 wf.write(

1558 b"gitdir: "

1559 + os.fsencode(pointer_to_write)

1560 + b"\n"

1561 )

1562 repaired.append(old_worktree_path)

1563 except (PermissionError, UnicodeDecodeError):

1564 continue

1565

1566 return repaired

1567

1568

1569@contextmanager

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

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

1572

1573 Args:

1574 repo: Dulwich repository object

1575 prefix: Prefix for the temporary directory name

1576

1577 Yields:

1578 Worktree object

1579 """

1580 temp_dir = None

1581 worktree = None

1582

1583 try:

1584 # Create temporary directory

1585 temp_dir = tempfile.mkdtemp(prefix=prefix)

1586

1587 # Add worktree

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

1589

1590 yield worktree

1591

1592 finally:

1593 # Clean up worktree registration

1594 if worktree:

1595 repo.worktrees.remove(worktree.path)

1596

1597 # Clean up temporary directory

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

1599 shutil.rmtree(temp_dir)