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(

444 st, ObjectID(blob.get_id(self._repo.object_format))

445 )

446 index.write()

447

448 def unstage(

449 self,

450 fs_paths: Sequence[str],

451 config: Config | None = None,

452 ) -> None:

453 """Unstage specific file in the index.

454

455 Args:

456 fs_paths: a list of files to unstage,

457 relative to the repository path.

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

459 ``self._repo.get_config_stack()``.

460 """

461 if config is None:

462 config = self._repo.get_config_stack()

463 from .index import IndexEntry, _fs_to_tree_path

464

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

466 try:

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

468 except KeyError:

469 # no head mean no commit in the repo

470 for fs_path in fs_paths:

471 tree_path = _fs_to_tree_path(fs_path)

472 del index[tree_path]

473 index.write()

474 return

475 else:

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

477 tree_id = commit.tree

478

479 for fs_path in fs_paths:

480 tree_path = _fs_to_tree_path(fs_path)

481 try:

482 tree = self._repo.object_store[tree_id]

483 assert isinstance(tree, Tree)

484 tree_entry = tree.lookup_path(

485 self._repo.object_store.__getitem__, tree_path

486 )

487 except KeyError:

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

489 # remove index entry

490 try:

491 del index[tree_path]

492 continue

493 except KeyError as exc:

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

495

496 st = None

497 try:

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

499 except FileNotFoundError:

500 pass

501

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

503 assert isinstance(blob_obj, Blob)

504 blob_size = len(blob_obj.data)

505

506 index_entry = IndexEntry(

507 ctime=(commit.commit_time, 0),

508 mtime=(commit.commit_time, 0),

509 dev=st.st_dev if st else 0,

510 ino=st.st_ino if st else 0,

511 mode=tree_entry[0],

512 uid=st.st_uid if st else 0,

513 gid=st.st_gid if st else 0,

514 size=blob_size,

515 sha=tree_entry[1],

516 flags=0,

517 extended_flags=0,

518 )

519

520 index[tree_path] = index_entry

521 index.write()

522

523 def commit(

524 self,

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

526 committer: bytes | None = None,

527 author: bytes | None = None,

528 commit_timestamp: float | None = None,

529 commit_timezone: int | None = None,

530 author_timestamp: float | None = None,

531 author_timezone: int | None = None,

532 tree: ObjectID | None = None,

533 encoding: bytes | None = None,

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

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

536 no_verify: bool = False,

537 sign: bool | None = None,

538 signoff: bool | None = None,

539 config: Config | None = None,

540 ) -> ObjectID:

541 """Create a new commit.

542

543 If not specified, committer and author default to

544 get_user_identity(..., 'COMMITTER')

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

546

547 Args:

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

549 and returns bytes)

550 committer: Committer fullname

551 author: Author fullname

552 commit_timestamp: Commit timestamp (defaults to now)

553 commit_timezone: Commit timestamp timezone (defaults to GMT)

554 author_timestamp: Author timestamp (defaults to commit

555 timestamp)

556 author_timezone: Author timestamp timezone

557 (defaults to commit timestamp timezone)

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

559 current index will be committed).

560 encoding: Encoding

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

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

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

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

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

566 pass True to use default GPG key,

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

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

569 If None, uses format.signoff config.

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

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

572 ``self._repo.get_config_stack()``.

573

574 Returns:

575 New commit SHA1

576 """

577 try:

578 if not no_verify:

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

580 except HookError as exc:

581 raise CommitError(exc) from exc

582 except KeyError: # no hook defined, silent fallthrough

583 pass

584

585 if config is None:

586 config = self._repo.get_config_stack()

587

588 object_format = self._repo.object_format

589 c = Commit()

590 if tree is None:

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

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

593 else:

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

595 raise ValueError(

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

597 )

598 c.tree = tree

599

600 if merge_heads is None:

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

602 if committer is None:

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

604 check_user_identity(committer)

605 c.committer = committer

606 if commit_timestamp is None:

607 # FIXME: Support GIT_COMMITTER_DATE environment variable

608 commit_timestamp = time.time()

609 c.commit_time = int(commit_timestamp)

610 if commit_timezone is None:

611 # FIXME: Use current user timezone rather than UTC

612 commit_timezone = 0

613 c.commit_timezone = commit_timezone

614 if author is None:

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

616 c.author = author

617 check_user_identity(author)

618 if author_timestamp is None:

619 # FIXME: Support GIT_AUTHOR_DATE environment variable

620 author_timestamp = commit_timestamp

621 c.author_time = int(author_timestamp)

622 if author_timezone is None:

623 author_timezone = commit_timezone

624 c.author_timezone = author_timezone

625 if encoding is None:

626 try:

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

628 except KeyError:

629 pass # No dice

630 if encoding is not None:

631 c.encoding = encoding

632 # Store original message (might be callable)

633 original_message = message

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

635

636 # Check if we should sign the commit

637 if sign is None:

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

639 try:

640 should_sign = config.get_boolean(

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

642 )

643 except KeyError:

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

645 else:

646 should_sign = sign

647

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

649 keyid = None

650 if should_sign:

651 try:

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

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

654 except KeyError:

655 keyid = None

656

657 if ref is None:

658 # Create a dangling commit

659 c.parents = list(merge_heads)

660 else:

661 try:

662 old_head = self._repo.refs[ref]

663 c.parents = [old_head, *merge_heads]

664 except KeyError:

665 c.parents = list(merge_heads)

666

667 # Handle message after parents are set

668 if callable(original_message):

669 message = original_message(self._repo, c)

670 if message is None:

671 raise ValueError("Message callback returned None")

672 else:

673 message = original_message

674

675 if message is None:

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

677 raise ValueError("No commit message specified")

678

679 # Handle signoff

680 should_signoff = signoff

681 if should_signoff is None:

682 # Check format.signOff configuration

683 try:

684 should_signoff = config.get_boolean(

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

686 )

687 except KeyError:

688 should_signoff = False

689

690 if should_signoff:

691 # Add Signed-off-by trailer

692 # Get the committer identity for the signoff

693 signoff_identity = committer

694 if isinstance(message, bytes):

695 message_bytes = message

696 else:

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

698

699 message_bytes = add_trailer_to_message(

700 message_bytes,

701 "Signed-off-by",

702 signoff_identity.decode("utf-8")

703 if isinstance(signoff_identity, bytes)

704 else signoff_identity,

705 separator=":",

706 where="end",

707 if_exists="addIfDifferentNeighbor",

708 if_missing="add",

709 )

710 message = message_bytes

711

712 try:

713 if no_verify:

714 c.message = message

715 else:

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

717 if c.message is None:

718 c.message = message

719 except HookError as exc:

720 raise CommitError(exc) from exc

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

722 c.message = message

723

724 if ref is None:

725 # Create a dangling commit

726 if should_sign:

727 from dulwich.signature import get_signature_vendor

728

729 vendor = get_signature_vendor(config=config)

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

731 self._repo.object_store.add_object(c)

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

733 else:

734 try:

735 old_head = self._repo.refs[ref]

736 if should_sign:

737 from dulwich.signature import get_signature_vendor

738

739 vendor = get_signature_vendor(config=config)

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

741 self._repo.object_store.add_object(c)

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

743 message_bytes = (

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

745 )

746 ok = self._repo.refs.set_if_equals(

747 ref,

748 old_head,

749 commit_id,

750 message=b"commit: " + message_bytes,

751 committer=committer,

752 timestamp=int(commit_timestamp)

753 if commit_timestamp is not None

754 else None,

755 timezone=commit_timezone,

756 )

757 except KeyError:

758 c.parents = list(merge_heads)

759 if should_sign:

760 from dulwich.signature import get_signature_vendor

761

762 vendor = get_signature_vendor(config=config)

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

764 self._repo.object_store.add_object(c)

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

766 message_bytes = (

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

768 )

769 ok = self._repo.refs.add_if_new(

770 ref,

771 commit_id,

772 message=b"commit: " + message_bytes,

773 committer=committer,

774 timestamp=int(commit_timestamp)

775 if commit_timestamp is not None

776 else None,

777 timezone=commit_timezone,

778 )

779 if not ok:

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

781 # commit and all its objects as garbage.

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

783

784 self._repo._del_named_file("MERGE_HEAD")

785

786 try:

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

788 except HookError as e: # silent failure

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

790 except KeyError: # no hook defined, silent fallthrough

791 pass

792

793 # Trigger auto GC if needed

794 from .gc import maybe_auto_gc

795

796 maybe_auto_gc(self._repo)

797

798 return commit_id

799

800 def reset_index(

801 self,

802 tree: ObjectID | None = None,

803 config: Config | None = None,

804 ) -> None:

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

806

807 Args:

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

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

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

811 """

812 if config is None:

813 config = self._repo.get_config_stack()

814 stacked_config = config

815 from .index import (

816 build_index_from_tree,

817 get_path_element_validator,

818 get_symlink_fn,

819 )

820

821 if tree is None:

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

823 if isinstance(head, Tag):

824 _cls, obj = head.object

825 head = self._repo.get_object(obj)

826 from .objects import Commit

827

828 assert isinstance(head, Commit)

829 tree = head.tree

830 assert tree is not None

831 config = self._repo.get_config()

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

833 validate_path_element = get_path_element_validator(config)

834 symlink_fn = get_symlink_fn(config)

835

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

837 return build_index_from_tree(

838 self.path,

839 self._repo.index_path(),

840 self._repo.object_store,

841 tree,

842 honor_filemode=honor_filemode,

843 validate_path_element=validate_path_element,

844 symlink_fn=symlink_fn,

845 blob_normalizer=blob_normalizer,

846 object_format=self._repo.object_format,

847 )

848

849 def _sparse_checkout_file_path(self) -> str:

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

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

852

853 def configure_for_cone_mode(self) -> None:

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

855 config = self._repo.get_config()

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

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

858 config.write_to_path()

859

860 def infer_cone_mode(self) -> bool:

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

862 config = self._repo.get_config()

863 try:

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

865 return sc_cone == b"true"

866 except KeyError:

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

868 return False

869

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

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

872

873 Returns:

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

875 """

876 path = self._sparse_checkout_file_path()

877 try:

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

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

880 except FileNotFoundError:

881 return []

882

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

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

885

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

887

888 Args:

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

890 """

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

892 os.makedirs(info_dir, exist_ok=True)

893

894 path = self._sparse_checkout_file_path()

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

896 for pat in patterns:

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

898

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

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

901

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

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

904 Never add the same line twice.

905 """

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

907 if dirs:

908 for d in dirs:

909 d = d.strip("/")

910 line = f"/{d}/"

911 if d and line not in patterns:

912 patterns.append(line)

913 self.set_sparse_checkout_patterns(patterns)

914

915

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

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

918

919 Args:

920 worktree_path: Path to the worktree's administrative directory

921

922 Returns:

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

924 """

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

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

927 return None

928

929 try:

930 with open(locked_path) as f:

931 return f.read().strip()

932 except (FileNotFoundError, PermissionError):

933 return None

934

935

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

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

938

939 Args:

940 repo: The repository to list worktrees for

941

942 Returns:

943 A list of WorkTreeInfo objects

944 """

945 worktrees = []

946

947 # Add main worktree

948 main_wt_info = WorkTreeInfo(

949 path=repo.path,

950 head=repo.head(),

951 bare=repo.bare,

952 detached=False,

953 locked=False,

954 prunable=False,

955 )

956

957 # Get branch info for main worktree

958 try:

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

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

961 if head_contents.startswith(SYMREF):

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

963 main_wt_info.branch = ref_name

964 else:

965 main_wt_info.detached = True

966 main_wt_info.branch = None

967 except (FileNotFoundError, PermissionError):

968 main_wt_info.branch = None

969 main_wt_info.detached = True

970

971 worktrees.append(main_wt_info)

972

973 # List additional worktrees

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

975 if os.path.isdir(worktrees_dir):

976 for entry in os.listdir(worktrees_dir):

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

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

979 continue

980

981 wt_info = WorkTreeInfo(

982 path="", # Will be set below

983 bare=False,

984 detached=False,

985 locked=False,

986 prunable=False,

987 )

988

989 # Read gitdir to get actual worktree path

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

991 try:

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

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

994 # Convert relative path to absolute if needed

995 wt_path = os.fsdecode(gitdir_contents)

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

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

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

999 except (FileNotFoundError, PermissionError):

1000 # Worktree directory is missing, skip it

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

1002 continue

1003

1004 # Check if worktree path exists

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

1006 wt_info.prunable = True

1007

1008 # Read HEAD

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

1010 try:

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

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

1013 if head_contents.startswith(SYMREF):

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

1015 wt_info.branch = ref_name

1016 # Resolve ref to get commit sha

1017 try:

1018 wt_info.head = repo.refs[ref_name]

1019 except KeyError:

1020 wt_info.head = None

1021 else:

1022 wt_info.detached = True

1023 wt_info.branch = None

1024 wt_info.head = head_contents

1025 except (FileNotFoundError, PermissionError):

1026 wt_info.head = None

1027 wt_info.branch = None

1028

1029 # Check if locked

1030 lock_reason = read_worktree_lock_reason(worktree_path)

1031 if lock_reason is not None:

1032 wt_info.locked = True

1033 wt_info.lock_reason = lock_reason

1034

1035 worktrees.append(wt_info)

1036

1037 return worktrees

1038

1039

1040def add_worktree(

1041 repo: Repo,

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

1043 branch: str | bytes | None = None,

1044 commit: ObjectID | None = None,

1045 force: bool = False,

1046 detach: bool = False,

1047 exist_ok: bool = False,

1048 relative_paths: bool | None = None,

1049) -> Repo:

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

1051

1052 Args:

1053 repo: The main repository

1054 path: Path where the new worktree should be created

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

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

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

1058 detach: Detach HEAD in the new worktree

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

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

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

1062

1063 Returns:

1064 The newly created worktree repository

1065

1066 Raises:

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

1068 """

1069 from .repo import Repo as RepoClass

1070

1071 path = os.fspath(path)

1072 if isinstance(path, bytes):

1073 path = os.fsdecode(path)

1074

1075 # Determine whether to use relative paths

1076 use_relative = _should_use_relative_paths(repo, relative_paths)

1077

1078 # Check if path already exists

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

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

1081

1082 # Normalize branch name

1083 if branch is not None:

1084 if isinstance(branch, str):

1085 branch = branch.encode()

1086 branch = local_branch_name(branch)

1087

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

1089 if branch and not force:

1090 for wt in list_worktrees(repo):

1091 if wt.branch == branch:

1092 raise ValueError(

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

1094 )

1095

1096 # Determine what to checkout

1097 if commit is not None:

1098 checkout_ref = commit

1099 detach = True

1100 elif branch is not None:

1101 # Check if branch exists

1102 try:

1103 checkout_ref = repo.refs[branch]

1104 except KeyError:

1105 if commit is None:

1106 # Create new branch from HEAD

1107 checkout_ref = repo.head()

1108 repo.refs[branch] = checkout_ref

1109 else:

1110 # Create new branch from specified commit

1111 checkout_ref = commit

1112 repo.refs[branch] = checkout_ref

1113 else:

1114 # Default to current HEAD

1115 checkout_ref = repo.head()

1116 detach = True

1117

1118 # Create the worktree directory

1119 os.makedirs(path, exist_ok=exist_ok)

1120

1121 # Initialize the worktree

1122 identifier = os.path.basename(path)

1123 wt_repo = RepoClass._init_new_working_directory(

1124 path, repo, identifier=identifier, relative_paths=use_relative

1125 )

1126

1127 # Set HEAD appropriately

1128 if detach:

1129 # Detached HEAD - write SHA directly to HEAD

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

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

1132 else:

1133 # Point to branch

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

1135 from dulwich.refs import HEADREF

1136

1137 wt_repo.refs.set_symbolic_ref(HEADREF, branch)

1138

1139 # Reset index to match HEAD

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

1141

1142 return wt_repo

1143

1144

1145def remove_worktree(

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

1147) -> None:

1148 """Remove a worktree.

1149

1150 Args:

1151 repo: The main repository

1152 path: Path to the worktree to remove

1153 force: Force removal even if there are local changes

1154

1155 Raises:

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

1157 """

1158 path = os.fspath(path)

1159 if isinstance(path, bytes):

1160 path = os.fsdecode(path)

1161

1162 # Don't allow removing the main worktree

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

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

1165

1166 # Find the worktree

1167 worktree_found = False

1168 worktree_id = None

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

1170

1171 if os.path.isdir(worktrees_dir):

1172 for entry in os.listdir(worktrees_dir):

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

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

1175

1176 try:

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

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

1179 wt_path = os.fsdecode(gitdir_contents)

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

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

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

1183

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

1185 worktree_found = True

1186 worktree_id = entry

1187 break

1188 except (FileNotFoundError, PermissionError):

1189 continue

1190

1191 if not worktree_found:

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

1193

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

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

1196

1197 # Check if locked

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

1199 if not force:

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

1201

1202 # Check for local changes if not forcing

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

1204 # TODO: Check for uncommitted changes in the worktree

1205 pass

1206

1207 # Remove the working directory

1208 if os.path.exists(path):

1209 shutil.rmtree(path)

1210

1211 # Remove the administrative files

1212 shutil.rmtree(worktree_control_dir)

1213

1214

1215def prune_worktrees(

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

1217) -> list[str]:

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

1219

1220 Args:

1221 repo: The main repository

1222 expire: Only prune worktrees older than this many seconds

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

1224

1225 Returns:

1226 List of pruned worktree identifiers

1227 """

1228 pruned: list[str] = []

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

1230

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

1232 return pruned

1233

1234 current_time = time.time()

1235

1236 for entry in os.listdir(worktrees_dir):

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

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

1239 continue

1240

1241 # Skip locked worktrees

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

1243 continue

1244

1245 should_prune = False

1246

1247 # Check if gitdir exists and points to valid location

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

1249 try:

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

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

1252 wt_path = os.fsdecode(gitdir_contents)

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

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

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

1256

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

1258 should_prune = True

1259 except (FileNotFoundError, PermissionError):

1260 should_prune = True

1261

1262 # Check expiry time if specified

1263 if should_prune and expire is not None:

1264 stat_info = os.stat(worktree_path)

1265 age = current_time - stat_info.st_mtime

1266 if age < expire:

1267 should_prune = False

1268

1269 if should_prune:

1270 pruned.append(entry)

1271 if not dry_run:

1272 shutil.rmtree(worktree_path)

1273

1274 return pruned

1275

1276

1277def lock_worktree(

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

1279) -> None:

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

1281

1282 Args:

1283 repo: The main repository

1284 path: Path to the worktree to lock

1285 reason: Optional reason for locking

1286 """

1287 worktree_id = _find_worktree_id(repo, path)

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

1289

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

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

1292 if reason:

1293 f.write(reason)

1294

1295

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

1297 """Unlock a worktree.

1298

1299 Args:

1300 repo: The main repository

1301 path: Path to the worktree to unlock

1302 """

1303 worktree_id = _find_worktree_id(repo, path)

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

1305

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

1307 if os.path.exists(lock_path):

1308 os.remove(lock_path)

1309

1310

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

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

1313

1314 Args:

1315 repo: The main repository

1316 path: Path to the worktree

1317

1318 Returns:

1319 The worktree identifier

1320

1321 Raises:

1322 ValueError: If the worktree is not found

1323 """

1324 path = os.fspath(path)

1325 if isinstance(path, bytes):

1326 path = os.fsdecode(path)

1327

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

1329

1330 if os.path.isdir(worktrees_dir):

1331 for entry in os.listdir(worktrees_dir):

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

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

1334

1335 try:

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

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

1338 wt_path = os.fsdecode(gitdir_contents)

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

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

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

1342

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

1344 return entry

1345 except (FileNotFoundError, PermissionError):

1346 continue

1347

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

1349

1350

1351def move_worktree(

1352 repo: Repo,

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

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

1355 relative_paths: bool | None = None,

1356) -> None:

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

1358

1359 Args:

1360 repo: The main repository

1361 old_path: Current path of the worktree

1362 new_path: New path for the worktree

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

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

1365

1366 Raises:

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

1368 """

1369 old_path = os.fspath(old_path)

1370 new_path = os.fspath(new_path)

1371 if isinstance(old_path, bytes):

1372 old_path = os.fsdecode(old_path)

1373 if isinstance(new_path, bytes):

1374 new_path = os.fsdecode(new_path)

1375

1376 # Don't allow moving the main worktree

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

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

1379

1380 # Check if new path already exists

1381 if os.path.exists(new_path):

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

1383

1384 # Find the worktree

1385 worktree_id = _find_worktree_id(repo, old_path)

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

1387

1388 # Read existing path to check format

1389 existing_path = None

1390 try:

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

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

1393 except (FileNotFoundError, PermissionError):

1394 pass

1395

1396 # Determine whether to use relative paths

1397 use_relative = _should_use_relative_paths(repo, relative_paths, existing_path)

1398

1399 # Move the actual worktree directory

1400 shutil.move(old_path, new_path)

1401

1402 # Update the gitdir file in the worktree

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

1404

1405 # Compute the path to write

1406 gitdir_path = _compute_gitdir_path(

1407 repo, gitdir_file_abs, worktree_control_dir, use_relative

1408 )

1409

1410 # Update the gitdir pointer in the control directory

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

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

1413

1414

1415def repair_worktree(

1416 repo: Repo,

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

1418 relative_paths: bool | None = None,

1419) -> list[str]:

1420 """Repair worktree administrative files.

1421

1422 This repairs the connection between worktrees and the main repository

1423 when they have been moved or become corrupted.

1424

1425 Args:

1426 repo: The main repository

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

1428 connections from the main repository to all linked worktrees.

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

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

1431

1432 Returns:

1433 List of repaired worktree paths

1434

1435 Raises:

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

1437 """

1438 repaired: list[str] = []

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

1440

1441 if paths:

1442 # Repair specific worktrees

1443 for path in paths:

1444 path_str = os.fspath(path)

1445 if isinstance(path_str, bytes):

1446 path_str = os.fsdecode(path_str)

1447 path_str = os.path.abspath(path_str)

1448

1449 # Check if this is a linked worktree

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

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

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

1453

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

1455 try:

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

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

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

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

1460 else:

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

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

1463 raise ValueError(

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

1465 ) from e

1466

1467 # Make the path absolute if it's relative

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

1469 worktree_control_path = os.path.abspath(

1470 os.path.join(path_str, worktree_control_path)

1471 )

1472

1473 # Update the gitdir file in the worktree control directory

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

1475 if os.path.exists(gitdir_pointer):

1476 # Read existing path to check format

1477 existing_path = None

1478 try:

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

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

1481 except (FileNotFoundError, PermissionError):

1482 pass

1483

1484 # Determine which format to use for this worktree

1485 use_relative = _should_use_relative_paths(

1486 repo, relative_paths, existing_path

1487 )

1488

1489 # Compute the path to write

1490 gitdir_path_to_write = _compute_gitdir_path(

1491 repo, gitdir_file, worktree_control_path, use_relative

1492 )

1493

1494 # Update to point to the current location

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

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

1497 repaired.append(path_str)

1498 else:

1499 # Repair from main repository to all linked worktrees

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

1501 return repaired

1502

1503 for entry in os.listdir(worktrees_dir):

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

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

1506 continue

1507

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

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

1510 try:

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

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

1513 old_gitdir_location = os.fsdecode(gitdir_contents)

1514 except (FileNotFoundError, PermissionError):

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

1516 continue

1517

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

1519 old_worktree_path = os.path.dirname(old_gitdir_location)

1520

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

1522 if os.path.exists(old_gitdir_location):

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

1524 try:

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

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

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

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

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

1530 current_pointer = os.path.abspath(

1531 os.path.join(old_worktree_path, current_pointer)

1532 )

1533

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

1535 expected_pointer = worktree_control_path

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

1537 expected_pointer

1538 ):

1539 # Determine which format to use

1540 use_relative = _should_use_relative_paths(

1541 repo, relative_paths, gitdir_contents

1542 )

1543

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

1545 pointer_to_write = _compute_gitdir_path(

1546 repo,

1547 worktree_control_path,

1548 old_worktree_path,

1549 use_relative,

1550 )

1551

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

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

1554 wf.write(

1555 b"gitdir: "

1556 + os.fsencode(pointer_to_write)

1557 + b"\n"

1558 )

1559 repaired.append(old_worktree_path)

1560 except (PermissionError, UnicodeDecodeError):

1561 continue

1562

1563 return repaired

1564

1565

1566@contextmanager

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

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

1569

1570 Args:

1571 repo: Dulwich repository object

1572 prefix: Prefix for the temporary directory name

1573

1574 Yields:

1575 Worktree object

1576 """

1577 temp_dir = None

1578 worktree = None

1579

1580 try:

1581 # Create temporary directory

1582 temp_dir = tempfile.mkdtemp(prefix=prefix)

1583

1584 # Add worktree

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

1586

1587 yield worktree

1588

1589 finally:

1590 # Clean up worktree registration

1591 if worktree:

1592 repo.worktrees.remove(worktree.path)

1593

1594 # Clean up temporary directory

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

1596 shutil.rmtree(temp_dir)