Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/patch.py: 10%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# patch.py -- For dealing with packed-style patches.
2# Copyright (C) 2009-2013 Jelmer Vernooij <jelmer@jelmer.uk>
3#
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.
9#
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"""Classes for dealing with git am-style patches.
24These patches are basically unified diffs with some extra metadata tacked
25on.
26"""
28__all__ = [
29 "DEFAULT_DIFF_ALGORITHM",
30 "FIRST_FEW_BYTES",
31 "DiffAlgorithmNotAvailable",
32 "FilePatch",
33 "MailinfoResult",
34 "PatchApplicationFailure",
35 "PatchHunk",
36 "apply_patch_hunks",
37 "apply_patches",
38 "commit_patch_id",
39 "gen_diff_header",
40 "get_summary",
41 "git_am_patch_split",
42 "git_base85_decode",
43 "is_binary",
44 "mailinfo",
45 "parse_patch_message",
46 "parse_unified_diff",
47 "patch_filename",
48 "patch_id",
49 "shortid",
50 "unified_diff",
51 "unified_diff_with_algorithm",
52 "write_blob_diff",
53 "write_commit_diff",
54 "write_commit_patch",
55 "write_object_diff",
56 "write_tree_diff",
57]
59import base64
60import email.message
61import email.parser
62import email.utils
63import os
64import re
65import stat
66import time
67import zlib
68from collections.abc import Generator, Sequence
69from dataclasses import dataclass
70from difflib import SequenceMatcher
71from typing import (
72 IO,
73 TYPE_CHECKING,
74 BinaryIO,
75 TextIO,
76)
78if TYPE_CHECKING:
79 from .config import Config
80 from .object_store import BaseObjectStore
81 from .repo import Repo
83from .objects import S_ISGITLINK, Blob, Commit, ObjectID, RawObjectID
85FIRST_FEW_BYTES = 8000
87DEFAULT_DIFF_ALGORITHM = "myers"
90class PatchApplicationFailure(Exception):
91 """Raised when a patch does not apply cleanly."""
94class DiffAlgorithmNotAvailable(Exception):
95 """Raised when a requested diff algorithm is not available."""
97 def __init__(self, algorithm: str, install_hint: str = "") -> None:
98 """Initialize exception.
100 Args:
101 algorithm: Name of the unavailable algorithm
102 install_hint: Optional installation hint
103 """
104 self.algorithm = algorithm
105 self.install_hint = install_hint
106 if install_hint:
107 super().__init__(
108 f"Diff algorithm '{algorithm}' requested but not available. {install_hint}"
109 )
110 else:
111 super().__init__(
112 f"Diff algorithm '{algorithm}' requested but not available."
113 )
116def write_commit_patch(
117 f: IO[bytes],
118 commit: "Commit",
119 contents: str | bytes,
120 progress: tuple[int, int],
121 version: str | None = None,
122 encoding: str | None = None,
123) -> None:
124 """Write a individual file patch.
126 Args:
127 f: File-like object to write to
128 commit: Commit object
129 contents: Contents of the patch
130 progress: tuple with current patch number and total.
131 version: Version string to include in patch header
132 encoding: Encoding to use for the patch
134 Returns:
135 tuple with filename and contents
136 """
137 encoding = encoding or getattr(f, "encoding", "ascii")
138 if encoding is None:
139 encoding = "ascii"
140 if isinstance(contents, str):
141 contents = contents.encode(encoding)
142 (num, total) = progress
143 f.write(
144 b"From "
145 + commit.id
146 + b" "
147 + time.ctime(commit.commit_time).encode(encoding)
148 + b"\n"
149 )
150 f.write(b"From: " + commit.author + b"\n")
151 f.write(
152 b"Date: " + time.strftime("%a, %d %b %Y %H:%M:%S %Z").encode(encoding) + b"\n"
153 )
154 f.write(
155 (f"Subject: [PATCH {num}/{total}] ").encode(encoding) + commit.message + b"\n"
156 )
157 f.write(b"\n")
158 f.write(b"---\n")
159 try:
160 import subprocess
162 p = subprocess.Popen(
163 ["diffstat"], stdout=subprocess.PIPE, stdin=subprocess.PIPE
164 )
165 except (ImportError, OSError):
166 pass # diffstat not available?
167 else:
168 (diffstat, _) = p.communicate(contents)
169 f.write(diffstat)
170 f.write(b"\n")
171 f.write(contents)
172 f.write(b"-- \n")
173 if version is None:
174 from dulwich import __version__ as dulwich_version
176 f.write(b"Dulwich %d.%d.%d\n" % dulwich_version)
177 else:
178 if encoding is None:
179 encoding = "ascii"
180 f.write(version.encode(encoding) + b"\n")
183def _sanitize_subject_for_filename(text: str, max_length: int = 52) -> str:
184 """Sanitize a string for safe use as part of a filename.
186 Matches git's ``format_sanitized_subject`` behavior:
188 - Only ``[A-Za-z0-9._]`` are kept; other characters become ``-``
189 (collapsed across runs).
190 - Consecutive ``.`` are collapsed to a single ``.``.
191 - The result is truncated to ``max_length`` characters.
192 - Trailing ``.`` and ``-`` are stripped.
194 Args:
195 text: Input string (typically a commit subject line).
196 max_length: Maximum length of the returned string.
198 Returns: Sanitized string safe to embed in a filename.
199 """
200 result: list[str] = []
201 # 2 = initial, 1 = saw a non-title char, 0 = saw a title char
202 space = 2
203 i = 0
204 text_len = len(text)
205 while i < text_len:
206 c = text[i]
207 if ("A" <= c <= "Z") or ("a" <= c <= "z") or ("0" <= c <= "9") or c in "._":
208 if space == 1:
209 result.append("-")
210 space = 0
211 result.append(c)
212 if c == ".":
213 while i + 1 < text_len and text[i + 1] == ".":
214 i += 1
215 else:
216 space |= 1
217 i += 1
218 if len(result) >= max_length:
219 break
221 return "".join(result)[:max_length].rstrip(".-")
224def get_summary(commit: "Commit") -> str:
225 """Determine the summary line for use in a filename.
227 Sanitizes the commit subject so it is safe to use as a filename
228 component, matching git's ``format_sanitized_subject`` behavior:
229 characters outside ``[A-Za-z0-9._]`` are replaced with ``-`` (with
230 runs collapsed) and consecutive ``.`` are collapsed. The result is
231 also length-limited to prevent overly long filenames.
233 Args:
234 commit: Commit
235 Returns: Sanitized summary string suitable for use as a filename
236 component.
237 """
238 decoded = commit.message.decode(errors="replace")
239 lines = decoded.splitlines()
240 if not lines:
241 return ""
242 return _sanitize_subject_for_filename(lines[0])
245# Unified Diff
246def _format_range_unified(start: int, stop: int) -> str:
247 """Convert range to the "ed" format."""
248 # Per the diff spec at http://www.unix.org/single_unix_specification/
249 beginning = start + 1 # lines start numbering with one
250 length = stop - start
251 if length == 1:
252 return f"{beginning}"
253 if not length:
254 beginning -= 1 # empty ranges begin at line just before the range
255 return f"{beginning},{length}"
258def unified_diff(
259 a: Sequence[bytes],
260 b: Sequence[bytes],
261 fromfile: bytes = b"",
262 tofile: bytes = b"",
263 fromfiledate: str = "",
264 tofiledate: str = "",
265 n: int = 3,
266 lineterm: str = "\n",
267 tree_encoding: str = "utf-8",
268 output_encoding: str = "utf-8",
269) -> Generator[bytes, None, None]:
270 """difflib.unified_diff that can detect "No newline at end of file" as original "git diff" does.
272 Based on the same function in Python2.7 difflib.py
273 """
274 started = False
275 for group in SequenceMatcher(a=a, b=b).get_grouped_opcodes(n):
276 if not started:
277 started = True
278 fromdate = f"\t{fromfiledate}" if fromfiledate else ""
279 todate = f"\t{tofiledate}" if tofiledate else ""
280 yield f"--- {fromfile.decode(tree_encoding)}{fromdate}{lineterm}".encode(
281 output_encoding
282 )
283 yield f"+++ {tofile.decode(tree_encoding)}{todate}{lineterm}".encode(
284 output_encoding
285 )
287 first, last = group[0], group[-1]
288 file1_range = _format_range_unified(first[1], last[2])
289 file2_range = _format_range_unified(first[3], last[4])
290 yield f"@@ -{file1_range} +{file2_range} @@{lineterm}".encode(output_encoding)
292 for tag, i1, i2, j1, j2 in group:
293 if tag == "equal":
294 for line in a[i1:i2]:
295 yield b" " + line
296 continue
297 if tag in ("replace", "delete"):
298 for line in a[i1:i2]:
299 if not line[-1:] == b"\n":
300 line += b"\n\\ No newline at end of file\n"
301 yield b"-" + line
302 if tag in ("replace", "insert"):
303 for line in b[j1:j2]:
304 if not line[-1:] == b"\n":
305 line += b"\n\\ No newline at end of file\n"
306 yield b"+" + line
309def _get_sequence_matcher(
310 algorithm: str, a: Sequence[bytes], b: Sequence[bytes]
311) -> SequenceMatcher[bytes]:
312 """Get appropriate sequence matcher for the given algorithm.
314 Args:
315 algorithm: Diff algorithm ("myers" or "patience")
316 a: First sequence
317 b: Second sequence
319 Returns:
320 Configured sequence matcher instance
322 Raises:
323 DiffAlgorithmNotAvailable: If patience requested but not available
324 """
325 if algorithm == "patience":
326 try:
327 from patiencediff import PatienceSequenceMatcher
329 return PatienceSequenceMatcher(None, a, b) # type: ignore[no-any-return,unused-ignore]
330 except ImportError:
331 raise DiffAlgorithmNotAvailable(
332 "patience", "Install with: pip install 'dulwich[patiencediff]'"
333 )
334 else:
335 return SequenceMatcher(a=a, b=b)
338def unified_diff_with_algorithm(
339 a: Sequence[bytes],
340 b: Sequence[bytes],
341 fromfile: bytes = b"",
342 tofile: bytes = b"",
343 fromfiledate: str = "",
344 tofiledate: str = "",
345 n: int = 3,
346 lineterm: str = "\n",
347 tree_encoding: str = "utf-8",
348 output_encoding: str = "utf-8",
349 algorithm: str | None = None,
350) -> Generator[bytes, None, None]:
351 """Generate unified diff with specified algorithm.
353 Args:
354 a: First sequence of lines
355 b: Second sequence of lines
356 fromfile: Name of first file
357 tofile: Name of second file
358 fromfiledate: Date of first file
359 tofiledate: Date of second file
360 n: Number of context lines
361 lineterm: Line terminator
362 tree_encoding: Encoding for tree paths
363 output_encoding: Encoding for output
364 algorithm: Diff algorithm to use ("myers" or "patience")
366 Returns:
367 Generator yielding diff lines
369 Raises:
370 DiffAlgorithmNotAvailable: If patience algorithm requested but patiencediff not available
371 """
372 if algorithm is None:
373 algorithm = DEFAULT_DIFF_ALGORITHM
375 matcher = _get_sequence_matcher(algorithm, a, b)
377 started = False
378 for group in matcher.get_grouped_opcodes(n):
379 if not started:
380 started = True
381 fromdate = f"\t{fromfiledate}" if fromfiledate else ""
382 todate = f"\t{tofiledate}" if tofiledate else ""
383 yield f"--- {fromfile.decode(tree_encoding)}{fromdate}{lineterm}".encode(
384 output_encoding
385 )
386 yield f"+++ {tofile.decode(tree_encoding)}{todate}{lineterm}".encode(
387 output_encoding
388 )
390 first, last = group[0], group[-1]
391 file1_range = _format_range_unified(first[1], last[2])
392 file2_range = _format_range_unified(first[3], last[4])
393 yield f"@@ -{file1_range} +{file2_range} @@{lineterm}".encode(output_encoding)
395 for tag, i1, i2, j1, j2 in group:
396 if tag == "equal":
397 for line in a[i1:i2]:
398 yield b" " + line
399 continue
400 if tag in ("replace", "delete"):
401 for line in a[i1:i2]:
402 if not line[-1:] == b"\n":
403 line += b"\n\\ No newline at end of file\n"
404 yield b"-" + line
405 if tag in ("replace", "insert"):
406 for line in b[j1:j2]:
407 if not line[-1:] == b"\n":
408 line += b"\n\\ No newline at end of file\n"
409 yield b"+" + line
412def is_binary(content: bytes) -> bool:
413 """See if the first few bytes contain any null characters.
415 Args:
416 content: Bytestring to check for binary content
417 """
418 return b"\0" in content[:FIRST_FEW_BYTES]
421def shortid(hexsha: bytes | None) -> bytes:
422 """Get short object ID.
424 Args:
425 hexsha: Full hex SHA or None
427 Returns:
428 7-character short ID
429 """
430 if hexsha is None:
431 return b"0" * 7
432 else:
433 return hexsha[:7]
436def patch_filename(p: bytes | None, root: bytes) -> bytes:
437 """Generate patch filename.
439 Args:
440 p: Path or None
441 root: Root directory
443 Returns:
444 Full patch filename
445 """
446 if p is None:
447 return b"/dev/null"
448 else:
449 return root + b"/" + p
452def write_object_diff(
453 f: IO[bytes],
454 store: "BaseObjectStore",
455 old_file: tuple[bytes | None, int | None, ObjectID | None],
456 new_file: tuple[bytes | None, int | None, ObjectID | None],
457 diff_binary: bool = False,
458 diff_algorithm: str | None = None,
459) -> None:
460 """Write the diff for an object.
462 Args:
463 f: File-like object to write to
464 store: Store to retrieve objects from, if necessary
465 old_file: (path, mode, hexsha) tuple
466 new_file: (path, mode, hexsha) tuple
467 diff_binary: Whether to diff files even if they
468 are considered binary files by is_binary().
469 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
471 Note: the tuple elements should be None for nonexistent files
472 """
473 (old_path, old_mode, old_id) = old_file
474 (new_path, new_mode, new_id) = new_file
475 patched_old_path = patch_filename(old_path, b"a")
476 patched_new_path = patch_filename(new_path, b"b")
478 def content(mode: int | None, hexsha: ObjectID | None) -> Blob:
479 """Get blob content for a file.
481 Args:
482 mode: File mode
483 hexsha: Object SHA
485 Returns:
486 Blob object
487 """
488 if hexsha is None:
489 return Blob.from_string(b"")
490 elif mode is not None and S_ISGITLINK(mode):
491 return Blob.from_string(b"Subproject commit " + hexsha + b"\n")
492 else:
493 obj = store[hexsha]
494 if isinstance(obj, Blob):
495 return obj
496 else:
497 # Fallback for non-blob objects
498 return Blob.from_string(obj.as_raw_string())
500 def lines(content: "Blob") -> list[bytes]:
501 """Split blob content into lines.
503 Args:
504 content: Blob content
506 Returns:
507 List of lines
508 """
509 if not content:
510 return []
511 else:
512 return content.splitlines()
514 f.writelines(
515 gen_diff_header((old_path, new_path), (old_mode, new_mode), (old_id, new_id))
516 )
517 old_content = content(old_mode, old_id)
518 new_content = content(new_mode, new_id)
519 if not diff_binary and (is_binary(old_content.data) or is_binary(new_content.data)):
520 binary_diff = (
521 b"Binary files "
522 + patched_old_path
523 + b" and "
524 + patched_new_path
525 + b" differ\n"
526 )
527 f.write(binary_diff)
528 else:
529 f.writelines(
530 unified_diff_with_algorithm(
531 lines(old_content),
532 lines(new_content),
533 patched_old_path,
534 patched_new_path,
535 algorithm=diff_algorithm,
536 )
537 )
540# TODO(jelmer): Support writing unicode, rather than bytes.
541def gen_diff_header(
542 paths: tuple[bytes | None, bytes | None],
543 modes: tuple[int | None, int | None],
544 shas: tuple[bytes | None, bytes | None],
545) -> Generator[bytes, None, None]:
546 """Write a blob diff header.
548 Args:
549 paths: Tuple with old and new path
550 modes: Tuple with old and new modes
551 shas: Tuple with old and new shas
552 """
553 (old_path, new_path) = paths
554 (old_mode, new_mode) = modes
555 (old_sha, new_sha) = shas
556 if old_path is None and new_path is not None:
557 old_path = new_path
558 if new_path is None and old_path is not None:
559 new_path = old_path
560 old_path = patch_filename(old_path, b"a")
561 new_path = patch_filename(new_path, b"b")
562 yield b"diff --git " + old_path + b" " + new_path + b"\n"
564 if old_mode != new_mode:
565 if new_mode is not None:
566 if old_mode is not None:
567 yield (f"old file mode {old_mode:o}\n").encode("ascii")
568 yield (f"new file mode {new_mode:o}\n").encode("ascii")
569 else:
570 yield (f"deleted file mode {old_mode:o}\n").encode("ascii")
571 yield b"index " + shortid(old_sha) + b".." + shortid(new_sha)
572 if new_mode is not None and old_mode is not None:
573 yield (f" {new_mode:o}").encode("ascii")
574 yield b"\n"
577# TODO(jelmer): Support writing unicode, rather than bytes.
578def write_blob_diff(
579 f: IO[bytes],
580 old_file: tuple[bytes | None, int | None, "Blob | None"],
581 new_file: tuple[bytes | None, int | None, "Blob | None"],
582 diff_algorithm: str | None = None,
583) -> None:
584 """Write blob diff.
586 Args:
587 f: File-like object to write to
588 old_file: (path, mode, hexsha) tuple (None if nonexisting)
589 new_file: (path, mode, hexsha) tuple (None if nonexisting)
590 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
592 Note: The use of write_object_diff is recommended over this function.
593 """
594 (old_path, old_mode, old_blob) = old_file
595 (new_path, new_mode, new_blob) = new_file
596 patched_old_path = patch_filename(old_path, b"a")
597 patched_new_path = patch_filename(new_path, b"b")
599 def lines(blob: "Blob | None") -> list[bytes]:
600 """Split blob content into lines.
602 Args:
603 blob: Blob object or None
605 Returns:
606 List of lines
607 """
608 if blob is not None:
609 return blob.splitlines()
610 else:
611 return []
613 f.writelines(
614 gen_diff_header(
615 (old_path, new_path),
616 (old_mode, new_mode),
617 (getattr(old_blob, "id", None), getattr(new_blob, "id", None)),
618 )
619 )
620 old_contents = lines(old_blob)
621 new_contents = lines(new_blob)
622 f.writelines(
623 unified_diff_with_algorithm(
624 old_contents,
625 new_contents,
626 patched_old_path,
627 patched_new_path,
628 algorithm=diff_algorithm,
629 )
630 )
633def write_tree_diff(
634 f: IO[bytes],
635 store: "BaseObjectStore",
636 old_tree: ObjectID | None,
637 new_tree: ObjectID | None,
638 diff_binary: bool = False,
639 diff_algorithm: str | None = None,
640) -> None:
641 """Write tree diff.
643 Args:
644 f: File-like object to write to.
645 store: Object store to read from
646 old_tree: Old tree id
647 new_tree: New tree id
648 diff_binary: Whether to diff files even if they
649 are considered binary files by is_binary().
650 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
651 """
652 changes = store.tree_changes(old_tree, new_tree)
653 for (oldpath, newpath), (oldmode, newmode), (oldsha, newsha) in changes:
654 write_object_diff(
655 f,
656 store,
657 (oldpath, oldmode, oldsha),
658 (newpath, newmode, newsha),
659 diff_binary=diff_binary,
660 diff_algorithm=diff_algorithm,
661 )
664def write_commit_diff(
665 f: IO[bytes],
666 store: "BaseObjectStore",
667 commit: "Commit",
668 diff_binary: bool = False,
669 diff_algorithm: str | None = None,
670) -> None:
671 """Write the diff a commit introduces against its first parent.
673 Args:
674 f: File-like object to write to.
675 store: Object store to read from.
676 commit: Commit whose changes to diff. Root commits (with no parents)
677 are diffed against the empty tree.
678 diff_binary: Whether to diff files even if they are considered binary
679 files by is_binary().
680 diff_algorithm: Algorithm to use for diffing ("myers" or "patience").
681 """
682 if commit.parents:
683 parent = store[commit.parents[0]]
684 assert isinstance(parent, Commit)
685 parent_tree: ObjectID | None = parent.tree
686 else:
687 parent_tree = None
688 write_tree_diff(
689 f,
690 store,
691 parent_tree,
692 commit.tree,
693 diff_binary=diff_binary,
694 diff_algorithm=diff_algorithm,
695 )
698def git_am_patch_split(
699 f: TextIO | BinaryIO, encoding: str | None = None
700) -> tuple["Commit", bytes, bytes | None]:
701 """Parse a git-am-style patch and split it up into bits.
703 Args:
704 f: File-like object to parse
705 encoding: Encoding to use when creating Git objects
706 Returns: Tuple with commit object, diff contents and git version
707 """
708 encoding = encoding or getattr(f, "encoding", "ascii")
709 encoding = encoding or "ascii"
710 contents = f.read()
711 if isinstance(contents, bytes):
712 bparser = email.parser.BytesParser()
713 msg = bparser.parsebytes(contents)
714 else:
715 uparser = email.parser.Parser()
716 msg = uparser.parsestr(contents)
717 return parse_patch_message(msg, encoding)
720def parse_patch_message(
721 msg: email.message.Message, encoding: str | None = None
722) -> tuple["Commit", bytes, bytes | None]:
723 """Extract a Commit object and patch from an e-mail message.
725 Args:
726 msg: An email message (email.message.Message)
727 encoding: Encoding to use to encode Git commits
728 Returns: Tuple with commit object, diff contents and git version
729 """
730 c = Commit()
731 if encoding is None:
732 encoding = "ascii"
733 c.author = msg["from"].encode(encoding)
734 c.committer = msg["from"].encode(encoding)
735 try:
736 patch_tag_start = msg["subject"].index("[PATCH")
737 except ValueError:
738 subject = msg["subject"]
739 else:
740 close = msg["subject"].index("] ", patch_tag_start)
741 subject = msg["subject"][close + 2 :]
742 c.message = (subject.replace("\n", "") + "\n").encode(encoding)
743 first = True
745 body = msg.get_payload(decode=True)
746 if isinstance(body, str):
747 body = body.encode(encoding)
748 if isinstance(body, bytes):
749 lines = body.splitlines(True)
750 else:
751 # Handle other types by converting to string first
752 lines = str(body).encode(encoding).splitlines(True)
753 line_iter = iter(lines)
755 for line in line_iter:
756 if line == b"---\n":
757 break
758 if first:
759 if line.startswith(b"From: "):
760 c.author = line[len(b"From: ") :].rstrip()
761 else:
762 c.message += b"\n" + line
763 first = False
764 else:
765 c.message += line
766 diff = b""
767 for line in line_iter:
768 if line == b"-- \n":
769 break
770 diff += line
771 try:
772 version = next(line_iter).rstrip(b"\n")
773 except StopIteration:
774 version = None
775 return c, diff, version
778def patch_id(diff_data: bytes) -> bytes:
779 """Compute patch ID for a diff.
781 The patch ID is computed by normalizing the diff and computing a SHA1 hash.
782 This follows git's patch-id algorithm which:
783 1. Removes whitespace from lines starting with + or -
784 2. Replaces line numbers in @@ headers with a canonical form
785 3. Computes SHA1 of the result
787 Args:
788 diff_data: Raw diff data as bytes
790 Returns:
791 SHA1 hash of normalized diff (40-byte hex string)
793 TODO: This implementation uses a simple line-by-line approach. For better
794 compatibility with git's patch-id, consider using proper patch parsing that:
795 - Handles edge cases in diff format (binary diffs, mode changes, etc.)
796 - Properly parses unified diff format according to the spec
797 - Matches git's exact normalization algorithm byte-for-byte
798 See git's patch-id.c for reference implementation.
799 """
800 import hashlib
801 import re
803 # Normalize the diff for patch-id computation
804 normalized_lines = []
806 for line in diff_data.split(b"\n"):
807 # Skip diff headers (diff --git, index, ---, +++)
808 if line.startswith(
809 (
810 b"diff --git ",
811 b"index ",
812 b"--- ",
813 b"+++ ",
814 b"new file mode ",
815 b"old file mode ",
816 b"deleted file mode ",
817 b"new mode ",
818 b"old mode ",
819 b"similarity index ",
820 b"dissimilarity index ",
821 b"rename from ",
822 b"rename to ",
823 b"copy from ",
824 b"copy to ",
825 )
826 ):
827 continue
829 # Normalize @@ headers to a canonical form
830 if line.startswith(b"@@"):
831 # Replace line numbers with canonical form
832 match = re.match(rb"^@@\s+-\d+(?:,\d+)?\s+\+\d+(?:,\d+)?\s+@@", line)
833 if match:
834 # Use canonical hunk header without line numbers
835 normalized_lines.append(b"@@")
836 continue
838 # For +/- lines, strip all whitespace
839 if line.startswith((b"+", b"-")):
840 # Keep the +/- prefix but remove all whitespace from the rest
841 if len(line) > 1:
842 # Remove all whitespace from the content
843 content = line[1:].replace(b" ", b"").replace(b"\t", b"")
844 normalized_lines.append(line[:1] + content)
845 else:
846 # Just +/- alone
847 normalized_lines.append(line[:1])
848 continue
850 # Keep context lines and other content as-is
851 if line.startswith(b" ") or line == b"":
852 normalized_lines.append(line)
854 # Join normalized lines and compute SHA1
855 normalized = b"\n".join(normalized_lines)
856 return hashlib.sha1(normalized).hexdigest().encode("ascii")
859def commit_patch_id(
860 store: "BaseObjectStore", commit_id: ObjectID | RawObjectID
861) -> bytes:
862 """Compute patch ID for a commit.
864 Args:
865 store: Object store to read objects from
866 commit_id: Commit ID (40-byte hex string)
868 Returns:
869 Patch ID (40-byte hex string)
870 """
871 from io import BytesIO
873 commit = store[commit_id]
874 assert isinstance(commit, Commit)
876 diff_output = BytesIO()
877 write_commit_diff(diff_output, store, commit)
879 return patch_id(diff_output.getvalue())
882@dataclass
883class MailinfoResult:
884 """Result of mailinfo parsing.
886 Attributes:
887 author_name: Author's name
888 author_email: Author's email address
889 author_date: Author's date (if present in the email)
890 subject: Processed subject line
891 message: Commit message body
892 patch: Patch content
893 message_id: Message-ID header (if -m/--message-id was used)
894 """
896 author_name: str
897 author_email: str
898 author_date: str | None
899 subject: str
900 message: str
901 patch: str
902 message_id: str | None = None
905def _munge_subject(subject: str, keep_subject: bool, keep_non_patch: bool) -> str:
906 """Munge email subject line for commit message.
908 Args:
909 subject: Original subject line
910 keep_subject: If True, keep subject intact (-k option)
911 keep_non_patch: If True, only strip [PATCH] (-b option)
913 Returns:
914 Processed subject line
915 """
916 if keep_subject:
917 return subject
919 result = subject
921 # First remove Re: prefixes (they can appear before brackets)
922 while True:
923 new_result = re.sub(r"^\s*(?:re|RE|Re):\s*", "", result, flags=re.IGNORECASE)
924 if new_result == result:
925 break
926 result = new_result
928 # Remove bracketed strings
929 if keep_non_patch:
930 # Only remove brackets containing "PATCH"
931 # Match each bracket individually anywhere in the string
932 while True:
933 # Remove PATCH bracket, but be careful with whitespace
934 new_result = re.sub(
935 r"\[[^\]]*?PATCH[^\]]*?\](\s+)?", r"\1", result, flags=re.IGNORECASE
936 )
937 if new_result == result:
938 break
939 result = new_result
940 else:
941 # Remove all bracketed strings
942 while True:
943 new_result = re.sub(r"^\s*\[.*?\]\s*", "", result)
944 if new_result == result:
945 break
946 result = new_result
948 # Remove leading/trailing whitespace
949 result = result.strip()
951 # Normalize multiple whitespace to single space
952 result = re.sub(r"\s+", " ", result)
954 return result
957def _find_scissors_line(lines: list[bytes]) -> int | None:
958 """Find the scissors line in message body.
960 Args:
961 lines: List of lines in the message body
963 Returns:
964 Index of scissors line, or None if not found
965 """
966 # The perforation alternatives are written so no two unbounded ``-+``
967 # quantifiers are separated by an optional/empty subpattern. The earlier
968 # form ``(?:>?\s*-+\s*)?(?:8<|>8)?\s*-+\s*`` let a single run of dashes be
969 # split between two ``-+`` groups, so a long non-matching line of dashes
970 # backtracked quadratically (ReDoS) when fed through ``mailinfo`` /
971 # ``git am --scissors`` on an untrusted mailbox. Here a marker is the
972 # mandatory pivot between the two dash runs, and the marker-less case uses a
973 # bounded ``(?:\s+-+)?`` second run, keeping the match linear.
974 scissors_pattern = re.compile(
975 rb"^>?\s*-+(?:\s+-+)?\s*$"
976 rb"|^>?\s*(?:-+\s*)?(?:8<|>8)\s*-+\s*$"
977 rb"|^(?:>?\s*-+\s*)(?:cut here|scissors)(?:\s*-+)?$",
978 re.IGNORECASE,
979 )
981 for i, line in enumerate(lines):
982 if scissors_pattern.match(line.strip()):
983 return i
985 return None
988def git_base85_decode(data: bytes) -> bytes:
989 """Decode Git's base85-encoded binary data.
991 Each line starts with a byte giving the decoded length of that line
992 (``A``-``Z`` for 1-26, ``a``-``z`` for 27-52), followed by groups of five
993 base85 characters that each encode four bytes. Git's alphabet is the
994 RFC 1924 one used by ``base64.b85decode``.
996 Args:
997 data: Base85-encoded data as bytes (may contain multiple lines)
999 Returns:
1000 Decoded binary data
1002 Raises:
1003 ValueError: If the data is invalid
1004 """
1005 result = bytearray()
1006 for line in data.splitlines():
1007 if not line:
1008 continue
1009 length_byte = line[0]
1010 if ord("A") <= length_byte <= ord("Z"):
1011 length = length_byte - ord("A") + 1
1012 elif ord("a") <= length_byte <= ord("z"):
1013 length = length_byte - ord("a") + 27
1014 else:
1015 raise ValueError(f"Invalid base85 line length byte: {line[:1]!r}")
1016 encoded = line[1:]
1017 if len(encoded) != (length + 3) // 4 * 5:
1018 raise ValueError(
1019 f"Base85 line has {len(encoded)} characters, "
1020 f"expected {(length + 3) // 4 * 5} for {length} bytes"
1021 )
1022 result.extend(base64.b85decode(encoded)[:length])
1023 return bytes(result)
1026@dataclass
1027class PatchHunk:
1028 """Represents a single hunk in a unified diff.
1030 Attributes:
1031 old_start: Starting line number in old file
1032 old_count: Number of lines in old file
1033 new_start: Starting line number in new file
1034 new_count: Number of lines in new file
1035 lines: List of diff lines (prefixed with ' ', '+', or '-')
1036 """
1038 old_start: int
1039 old_count: int
1040 new_start: int
1041 new_count: int
1042 lines: list[bytes]
1045@dataclass
1046class FilePatch:
1047 """Represents a patch for a single file.
1049 Attributes:
1050 old_path: Path to old file (None for new files)
1051 new_path: Path to new file (None for deleted files)
1052 old_mode: Mode of old file (None for new files)
1053 new_mode: Mode of new file (None for deleted files)
1054 hunks: List of PatchHunk objects
1055 binary: True if this is a binary patch
1056 rename_from: Original path for renames (None if not a rename)
1057 rename_to: New path for renames (None if not a rename)
1058 copy_from: Source path for copies (None if not a copy)
1059 copy_to: Destination path for copies (None if not a copy)
1060 binary_old: Reverse binary data for binary patches (base85 encoded)
1061 binary_new: Forward binary data for binary patches (base85 encoded)
1062 binary_old_delta: True if binary_old is a delta against the new
1063 content rather than a literal copy of the old content
1064 binary_new_delta: True if binary_new is a delta against the old
1065 content rather than a literal copy of the new content
1066 """
1068 old_path: bytes | None
1069 new_path: bytes | None
1070 old_mode: int | None
1071 new_mode: int | None
1072 hunks: list[PatchHunk]
1073 binary: bool = False
1074 rename_from: bytes | None = None
1075 rename_to: bytes | None = None
1076 copy_from: bytes | None = None
1077 copy_to: bytes | None = None
1078 binary_old: bytes | None = None
1079 binary_new: bytes | None = None
1080 binary_old_delta: bool = False
1081 binary_new_delta: bool = False
1084_C_STYLE_ESCAPES = {
1085 ord("a"): b"\a",
1086 ord("b"): b"\b",
1087 ord("t"): b"\t",
1088 ord("n"): b"\n",
1089 ord("v"): b"\v",
1090 ord("f"): b"\f",
1091 ord("r"): b"\r",
1092 ord('"'): b'"',
1093 ord("\\"): b"\\",
1094}
1097def _unquote_c_style(text: bytes) -> tuple[bytes, bytes]:
1098 """Unquote a C-style quoted name at the start of ``text``.
1100 Git quotes names containing special or non-ASCII characters this way.
1102 Returns:
1103 Tuple with the unquoted name and the remainder of ``text`` after the
1104 closing quote.
1106 Raises:
1107 ValueError: If the quoted name is malformed
1108 """
1109 if not text.startswith(b'"'):
1110 raise ValueError(f"Name is not quoted: {text!r}")
1111 result = bytearray()
1112 i = 1
1113 while i < len(text):
1114 c = text[i]
1115 if c == ord('"'):
1116 return bytes(result), text[i + 1 :]
1117 if c != ord("\\"):
1118 result.append(c)
1119 i += 1
1120 continue
1121 escaped = text[i + 1 : i + 2]
1122 if escaped and escaped[0] in _C_STYLE_ESCAPES:
1123 result += _C_STYLE_ESCAPES[escaped[0]]
1124 i += 2
1125 elif re.fullmatch(rb"[0-3][0-7][0-7]", text[i + 1 : i + 4]):
1126 result.append(int(text[i + 1 : i + 4], 8))
1127 i += 4
1128 else:
1129 raise ValueError(f"Invalid escape in quoted name: {text!r}")
1130 raise ValueError(f"Unterminated quoted name: {text!r}")
1133def _unquote_name(name: bytes) -> bytes:
1134 """Unquote a name from a patch header if git quoted it."""
1135 if not name.startswith(b'"'):
1136 return name
1137 unquoted, rest = _unquote_c_style(name)
1138 if rest:
1139 raise ValueError(f"Trailing data after quoted name: {name!r}")
1140 return unquoted
1143def _parse_file_line_path(path: bytes) -> bytes:
1144 """Parse the path from a ``---`` or ``+++`` line, dropping any timestamp."""
1145 if not path.startswith(b'"'):
1146 return path.split(b"\t")[0]
1147 unquoted, rest = _unquote_c_style(path)
1148 if rest and not rest.startswith(b"\t"):
1149 raise ValueError(f"Trailing data after quoted name: {path!r}")
1150 return unquoted
1153def _parse_git_diff_header_paths(line: bytes) -> tuple[bytes | None, bytes | None]:
1154 """Extract the old and new paths from a ``diff --git`` line.
1156 Unquoted names are only separable when they are the same apart from their
1157 prefixes (as for anything but a rename or copy), in which case the line is
1158 split in the middle. Returns ``(None, None)`` if the names differ.
1159 """
1160 names = line[len(b"diff --git ") :]
1161 if names.startswith(b'"'):
1162 old, rest = _unquote_c_style(names)
1163 if not rest.startswith(b" "):
1164 return None, None
1165 new = _unquote_name(rest[1:])
1166 else:
1167 half = len(names) // 2
1168 if len(names) % 2 != 1 or names[half : half + 1] != b" ":
1169 return None, None
1170 old, new = names[:half], names[half + 1 :]
1171 if old.partition(b"/")[2] != new.partition(b"/")[2]:
1172 return None, None
1173 return old, new
1176def parse_unified_diff(diff_text: bytes) -> list[FilePatch]:
1177 """Parse a unified diff into FilePatch objects.
1179 Args:
1180 diff_text: Unified diff content as bytes
1182 Returns:
1183 List of FilePatch objects
1184 """
1185 patches: list[FilePatch] = []
1186 lines = diff_text.split(b"\n")
1187 i = 0
1189 while i < len(lines):
1190 line = lines[i]
1192 # Look for diff header
1193 if line.startswith(b"diff --git "):
1194 # Parse file patch
1195 old_path = None
1196 new_path = None
1197 old_mode = None
1198 new_mode = None
1199 hunks: list[PatchHunk] = []
1200 binary = False
1201 rename_from = None
1202 rename_to = None
1203 copy_from = None
1204 copy_to = None
1205 binary_old = None
1206 binary_new = None
1207 binary_old_delta = False
1208 binary_new_delta = False
1209 header_old_path, header_new_path = _parse_git_diff_header_paths(line)
1211 # Parse extended headers
1212 i += 1
1213 while i < len(lines):
1214 line = lines[i]
1216 if line.startswith(b"old file mode "):
1217 old_mode = int(line.split()[-1], 8)
1218 i += 1
1219 elif line.startswith(b"new file mode "):
1220 new_mode = int(line.split()[-1], 8)
1221 header_old_path = None
1222 i += 1
1223 elif line.startswith(b"deleted file mode "):
1224 old_mode = int(line.split()[-1], 8)
1225 header_new_path = None
1226 i += 1
1227 elif line.startswith(b"new mode "):
1228 new_mode = int(line.split()[-1], 8)
1229 i += 1
1230 elif line.startswith(b"old mode "):
1231 old_mode = int(line.split()[-1], 8)
1232 i += 1
1233 elif line.startswith(b"rename from "):
1234 rename_from = _unquote_name(line[12:].strip())
1235 i += 1
1236 elif line.startswith(b"rename to "):
1237 rename_to = _unquote_name(line[10:].strip())
1238 i += 1
1239 elif line.startswith(b"copy from "):
1240 copy_from = _unquote_name(line[10:].strip())
1241 i += 1
1242 elif line.startswith(b"copy to "):
1243 copy_to = _unquote_name(line[8:].strip())
1244 i += 1
1245 elif line.startswith(b"similarity index "):
1246 # Just skip similarity index for now
1247 i += 1
1248 elif line.startswith(b"dissimilarity index "):
1249 # Just skip dissimilarity index for now
1250 i += 1
1251 elif line.startswith(b"index "):
1252 # "index <old>..<new> <mode>" gives the mode when the
1253 # patch doesn't change it.
1254 fields = line.split()
1255 if len(fields) == 3:
1256 mode = int(fields[2], 8)
1257 if old_mode is None:
1258 old_mode = mode
1259 if new_mode is None:
1260 new_mode = mode
1261 i += 1
1262 elif line.startswith(b"--- "):
1263 # Parse old file path
1264 path = _parse_file_line_path(line[4:])
1265 if path != b"/dev/null":
1266 old_path = path
1267 i += 1
1268 elif line.startswith(b"+++ "):
1269 # Parse new file path
1270 path = _parse_file_line_path(line[4:])
1271 if path != b"/dev/null":
1272 new_path = path
1273 i += 1
1274 break
1275 elif line.startswith(b"Binary files"):
1276 binary = True
1277 i += 1
1278 break
1279 elif line.startswith(b"GIT binary patch"):
1280 binary = True
1281 i += 1
1282 # The forward data comes first, followed by the reverse
1283 # data. Each is a "literal" or "delta" line, then base85
1284 # lines up to a blank line.
1285 blocks: list[tuple[bool, bytes]] = []
1286 while (
1287 i < len(lines)
1288 and len(blocks) < 2
1289 and lines[i].startswith((b"literal ", b"delta "))
1290 ):
1291 is_delta = lines[i].startswith(b"delta ")
1292 i += 1
1293 block_start = i
1294 while i < len(lines) and lines[i]:
1295 i += 1
1296 blocks.append(
1297 (
1298 is_delta,
1299 b"".join(data + b"\n" for data in lines[block_start:i]),
1300 )
1301 )
1302 i += 1
1303 if not blocks:
1304 raise ValueError("GIT binary patch without data")
1305 binary_new_delta, binary_new = blocks[0]
1306 if len(blocks) > 1:
1307 binary_old_delta, binary_old = blocks[1]
1308 break
1309 else:
1310 # Leave the line (e.g. the next "diff --git") to the
1311 # hunk parser below.
1312 break
1314 if old_path is None and new_path is None:
1315 # Binary patches and those without content changes (mode
1316 # changes, empty files) have no ---/+++ lines.
1317 old_path = header_old_path
1318 new_path = header_new_path
1320 # Parse hunks
1321 if not binary:
1322 while i < len(lines):
1323 line = lines[i]
1325 if line.startswith(b"@@ "):
1326 # Parse hunk header
1327 match = re.match(
1328 rb"@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", line
1329 )
1330 if match:
1331 old_start = int(match.group(1))
1332 old_count = int(match.group(2)) if match.group(2) else 1
1333 new_start = int(match.group(3))
1334 new_count = int(match.group(4)) if match.group(4) else 1
1336 # Parse hunk lines
1337 hunk_lines: list[bytes] = []
1338 i += 1
1339 while i < len(lines):
1340 line = lines[i]
1341 if line.startswith((b" ", b"+", b"-", b"\\")):
1342 hunk_lines.append(line)
1343 i += 1
1344 else:
1345 break
1347 hunks.append(
1348 PatchHunk(
1349 old_start=old_start,
1350 old_count=old_count,
1351 new_start=new_start,
1352 new_count=new_count,
1353 lines=hunk_lines,
1354 )
1355 )
1356 else:
1357 i += 1
1358 elif line.startswith(b"diff --git "):
1359 # Next file patch
1360 break
1361 else:
1362 i += 1
1363 if not line.strip():
1364 # Empty line, might be end of patch or separator
1365 break
1367 patches.append(
1368 FilePatch(
1369 old_path=old_path,
1370 new_path=new_path,
1371 old_mode=old_mode,
1372 new_mode=new_mode,
1373 hunks=hunks,
1374 binary=binary,
1375 rename_from=rename_from,
1376 rename_to=rename_to,
1377 copy_from=copy_from,
1378 copy_to=copy_to,
1379 binary_old=binary_old,
1380 binary_new=binary_new,
1381 binary_old_delta=binary_old_delta,
1382 binary_new_delta=binary_new_delta,
1383 )
1384 )
1385 else:
1386 i += 1
1388 return patches
1391def apply_patch_hunks(
1392 patch: FilePatch,
1393 original_lines: list[bytes],
1394) -> list[bytes] | None:
1395 """Apply patch hunks to file content.
1397 Args:
1398 patch: FilePatch object to apply
1399 original_lines: Original file content as list of lines
1401 Returns:
1402 Patched file content as list of lines, or None if patch cannot be applied
1403 """
1404 result = original_lines[:]
1405 offset = 0 # Track line offset as we apply hunks
1407 for hunk in patch.hunks:
1408 # Adjust hunk position by offset
1409 # old_start is 1-indexed; 0 means the hunk inserts at the beginning
1410 target_line = max(hunk.old_start - 1, 0) + offset
1412 # Extract old and new content from hunk
1413 old_content: list[bytes] = []
1414 new_content: list[bytes] = []
1416 previous = b""
1417 for line in hunk.lines:
1418 if line.startswith(b"\\"):
1419 # "\ No newline at end of file" applies to the line before it
1420 if previous in (b" ", b"-"):
1421 old_content[-1] = old_content[-1].removesuffix(b"\n")
1422 if previous in (b" ", b"+"):
1423 new_content[-1] = new_content[-1].removesuffix(b"\n")
1424 continue
1425 previous = line[:1]
1426 if line.startswith(b" "):
1427 # Context line - add newline if not present
1428 content = line[1:]
1429 if not content.endswith(b"\n"):
1430 content += b"\n"
1431 old_content.append(content)
1432 new_content.append(content)
1433 elif line.startswith(b"-"):
1434 # Deletion - add newline if not present
1435 content = line[1:]
1436 if not content.endswith(b"\n"):
1437 content += b"\n"
1438 old_content.append(content)
1439 elif line.startswith(b"+"):
1440 # Addition - add newline if not present
1441 content = line[1:]
1442 if not content.endswith(b"\n"):
1443 content += b"\n"
1444 new_content.append(content)
1446 # Verify context matches
1447 if target_line < 0 or target_line + len(old_content) > len(result):
1448 # TODO: Implement fuzzy matching
1449 return None
1451 for i, old_line in enumerate(old_content):
1452 if result[target_line + i] != old_line:
1453 # Context doesn't match
1454 # TODO: Implement fuzzy matching
1455 return None
1457 # Apply the patch
1458 result[target_line : target_line + len(old_content)] = new_content
1460 # Update offset for next hunk
1461 offset += len(new_content) - len(old_content)
1463 return result
1466def _ensure_within_repo(repo_path: bytes, fs_path: bytes, rel_path: bytes) -> None:
1467 """Reject patch target paths that resolve outside the working tree.
1469 Patch headers are untrusted (e.g. ``git am`` of a mailbox), so a name such
1470 as ``../../etc/cron.d/x`` or an absolute path must not be written through
1471 ``os.path.join``. Mirrors git's refusal of paths outside the work tree.
1472 """
1473 repo_root = os.path.realpath(repo_path)
1474 resolved = os.path.realpath(fs_path)
1475 if resolved != repo_root and not resolved.startswith(
1476 repo_root + os.fsencode(os.sep)
1477 ):
1478 raise ValueError(f"patch affects file outside repository: {rel_path!r}")
1481def _validate_patch_target(r: "Repo", repo_path: bytes, tree_path: bytes) -> bytes:
1482 """Validate a patch target path and return its filesystem path.
1484 ``_ensure_within_repo`` alone only refuses paths that escape the work tree;
1485 a target like ``.git/hooks/pre-commit`` stays inside it and would otherwise
1486 be written (and later executed as a hook). On top of that containment check,
1487 apply the same name and symlink checks git uses for checkout, so patch
1488 application cannot reach the control directory. Mirrors
1489 ``porcelain._checked_worktree_path``.
1491 Returns:
1492 The filesystem path under ``repo_path``, as bytes.
1493 """
1494 from .index import (
1495 InvalidPathError,
1496 get_path_element_validator,
1497 validate_path,
1498 verify_leading_dirs,
1499 )
1501 fs_path = os.path.join(repo_path, tree_path)
1502 _ensure_within_repo(repo_path, fs_path, tree_path)
1503 validator = get_path_element_validator(r.get_config_stack())
1504 if not validate_path(tree_path, validator):
1505 raise ValueError(f"refusing to write unsafe path: {tree_path!r}")
1506 try:
1507 verify_leading_dirs(tree_path, [], repo_path)
1508 except InvalidPathError:
1509 raise ValueError(f"refusing to write through symlink: {tree_path!r}")
1510 return fs_path
1513def _read_patch_target(fs_path: bytes) -> tuple[bytes, int] | None:
1514 """Read the content and mode of a patch target in the work tree.
1516 A symlink is read with ``readlink`` rather than followed, since its
1517 content in git is the link target.
1519 Returns:
1520 Tuple with content and mode, or None if the path does not exist
1521 """
1522 from .index import cleanup_mode
1524 try:
1525 st = os.lstat(fs_path)
1526 except FileNotFoundError:
1527 return None
1528 if stat.S_ISLNK(st.st_mode):
1529 return os.readlink(fs_path), stat.S_IFLNK
1530 with open(fs_path, "rb") as f:
1531 return f.read(), cleanup_mode(st.st_mode)
1534def _check_patch_target_type(
1535 tree_path: bytes, disk_mode: int, expected_mode: int, config: "Config"
1536) -> None:
1537 """Refuse to apply a patch for a symlink to a file or vice versa.
1539 ``verify_leading_dirs`` only checks the leading directories. A tracked
1540 symlink such as ``trap -> .git/hooks/pre-commit`` resolves inside the work
1541 tree, so writing a regular file patch to it would follow the link and land
1542 in the control directory. git refuses such patches with "wrong type".
1543 """
1544 if stat.S_ISLNK(disk_mode) == stat.S_ISLNK(expected_mode):
1545 return
1546 if (
1547 stat.S_ISLNK(expected_mode)
1548 and not stat.S_ISLNK(disk_mode)
1549 and not config.get_boolean(b"core", b"symlinks", True)
1550 ):
1551 # With core.symlinks=false links are checked out as plain files.
1552 return
1553 raise ValueError(f"wrong type for patch target: {tree_path!r}")
1556def _write_patch_target(
1557 fs_path: bytes, content: bytes, mode: int, config: "Config"
1558) -> os.stat_result:
1559 """Write a patch result to the work tree without following symlinks.
1561 Returns:
1562 The ``lstat`` result of the written path
1563 """
1564 from .index import build_file_from_blob, get_symlink_fn
1566 os.makedirs(os.path.dirname(fs_path), exist_ok=True)
1567 return build_file_from_blob(
1568 Blob.from_string(content), mode, fs_path, symlink_fn=get_symlink_fn(config)
1569 )
1572def _refuse_existing_target(fs_path: bytes, tree_path: bytes) -> None:
1573 """Refuse to create a file over an existing path, like git does."""
1574 if os.path.lexists(fs_path):
1575 raise ValueError(f"{tree_path!r} already exists in working directory")
1578def _load_patch_target(
1579 r: "Repo",
1580 fs_path: bytes,
1581 tree_path: bytes,
1582 expected_mode: int | None,
1583 config: "Config",
1584) -> tuple[bytes, int] | None:
1585 """Load the current content and mode of an existing patch target.
1587 Reads the work tree, falling back to the index if the path is missing
1588 there. The mode from the patch, if any, must match the type on disk.
1590 Returns:
1591 Tuple with content and mode, or None if the path can't be found
1592 """
1593 from .index import ConflictedIndexEntry, IndexEntry
1595 def index_entry() -> IndexEntry | None:
1596 try:
1597 entry = r.open_index(config=config)[tree_path]
1598 except (FileNotFoundError, KeyError):
1599 return None
1600 if isinstance(entry, ConflictedIndexEntry):
1601 return None
1602 return entry
1604 current = _read_patch_target(fs_path)
1605 if current is None:
1606 entry = index_entry()
1607 if entry is None:
1608 return None
1609 obj = r.object_store[entry.sha]
1610 if not isinstance(obj, Blob):
1611 return None
1612 return obj.data, entry.mode
1613 content, mode = current
1614 if expected_mode is not None:
1615 _check_patch_target_type(tree_path, mode, expected_mode, config)
1616 return content, expected_mode
1617 if not stat.S_ISLNK(mode) and not config.get_boolean(b"core", b"symlinks", True):
1618 # A link checked out as a plain file is only recognizable as such
1619 # from the index.
1620 entry = index_entry()
1621 if entry is not None and stat.S_ISLNK(entry.mode):
1622 return content, entry.mode
1623 return content, mode
1626def _apply_rename_or_copy(
1627 r: "Repo",
1628 src_path: bytes,
1629 dst_path: bytes,
1630 strip: int,
1631 patch: FilePatch,
1632 is_rename: bool,
1633 cached: bool,
1634 check: bool,
1635 config: "Config",
1636) -> tuple[list[bytes] | None, int, bool]:
1637 """Apply a rename or copy operation.
1639 Args:
1640 r: Repository object
1641 src_path: Source path
1642 dst_path: Destination path
1643 strip: Number of path components to strip
1644 patch: FilePatch object
1645 is_rename: True for rename, False for copy
1646 cached: Apply to index only, not working tree
1647 check: Check only, don't apply
1648 config: Repository configuration
1650 Returns:
1651 A tuple of (``original_lines``, ``old_mode``, ``should_continue``) where:
1652 - ``original_lines``: Content lines if hunks need to be applied, None otherwise
1653 - ``old_mode``: Mode of the source
1654 - ``should_continue``: True to skip to next patch, False to continue processing
1655 """
1656 from .index import IndexEntry, index_entry_from_stat
1658 # Strip path components
1659 src_stripped = src_path
1660 dst_stripped = dst_path
1661 if strip > 0:
1662 src_parts = src_path.split(b"/")
1663 if len(src_parts) > strip:
1664 src_stripped = b"/".join(src_parts[strip:])
1665 dst_parts = dst_path.split(b"/")
1666 if len(dst_parts) > strip:
1667 dst_stripped = b"/".join(dst_parts[strip:])
1669 repo_path_bytes = r.path.encode("utf-8") if isinstance(r.path, str) else r.path
1670 src_fs_path = _validate_patch_target(r, repo_path_bytes, src_stripped)
1671 dst_fs_path = _validate_patch_target(r, repo_path_bytes, dst_stripped)
1672 if not cached:
1673 _refuse_existing_target(dst_fs_path, dst_stripped)
1675 # Read content from source file
1676 op_name = "rename" if is_rename else "copy"
1677 current = _load_patch_target(r, src_fs_path, src_stripped, patch.old_mode, config)
1678 if current is None:
1679 raise ValueError(
1680 f"Cannot {op_name}: source {src_stripped.decode('utf-8', errors='replace')} not found"
1681 )
1682 content, old_mode = current
1684 # If the content changes too, return it for further processing
1685 if patch.hunks or patch.binary:
1686 return content.splitlines(keepends=True), old_mode, False
1688 # No hunks - pure rename/copy
1689 if check:
1690 return None, old_mode, True
1692 new_mode = patch.new_mode or old_mode
1693 index = r.open_index(config=config)
1694 blob = Blob.from_string(content)
1695 r.object_store.add_object(blob)
1697 if not cached:
1698 st = _write_patch_target(dst_fs_path, content, new_mode, config)
1699 entry = index_entry_from_stat(st, blob.id, mode=new_mode)
1700 else:
1701 entry = IndexEntry(
1702 ctime=(0, 0),
1703 mtime=(0, 0),
1704 dev=0,
1705 ino=0,
1706 mode=new_mode,
1707 uid=0,
1708 gid=0,
1709 size=len(content),
1710 sha=blob.id,
1711 flags=0,
1712 )
1714 index[dst_stripped] = entry
1716 # For renames, remove the old file
1717 if is_rename:
1718 if not cached and os.path.lexists(src_fs_path):
1719 os.remove(src_fs_path)
1720 if src_stripped in index:
1721 del index[src_stripped]
1723 index.write()
1724 return None, old_mode, True
1727def _apply_binary_patch(patch: FilePatch, original: bytes, reverse: bool) -> bytes:
1728 """Compute the new content of a file from a ``GIT binary patch``.
1730 Args:
1731 patch: Binary FilePatch
1732 original: Current content of the file
1733 reverse: Apply the reverse data instead of the forward data
1735 Returns:
1736 The patched content
1738 Raises:
1739 ValueError: If the patch data is invalid or doesn't apply
1740 """
1741 from .errors import ApplyDeltaError
1742 from .pack import apply_delta
1744 if reverse:
1745 data, is_delta = patch.binary_old, patch.binary_old_delta
1746 else:
1747 data, is_delta = patch.binary_new, patch.binary_new_delta
1748 if data is None:
1749 # "Binary files differ" message without actual patch data
1750 raise NotImplementedError(
1751 "Binary patch detected but no patch data provided (use git diff --binary)"
1752 )
1753 try:
1754 decoded = zlib.decompress(git_base85_decode(data))
1755 except (ValueError, zlib.error) as e:
1756 raise ValueError(f"Failed to decode binary patch: {e}")
1757 if not is_delta:
1758 return decoded
1759 try:
1760 return b"".join(apply_delta(original, decoded))
1761 except ApplyDeltaError as e:
1762 raise ValueError(f"Binary delta does not apply: {e}")
1765def apply_patches(
1766 r: "Repo",
1767 patches: list[FilePatch],
1768 cached: bool = False,
1769 reverse: bool = False,
1770 check: bool = False,
1771 strip: int = 1,
1772 three_way: bool = False,
1773 *,
1774 config: "Config | None" = None,
1775) -> None:
1776 """Apply a list of file patches to a repository.
1778 Args:
1779 r: Repository object
1780 patches: List of FilePatch objects to apply
1781 cached: Apply patch to index only, not working tree
1782 reverse: Apply patch in reverse
1783 check: Only check if patch can be applied, don't apply
1784 strip: Number of leading path components to strip (default: 1)
1785 three_way: Fall back to 3-way merge if patch does not apply cleanly
1786 config: Repository configuration. If None, falls back to
1787 ``r.get_config_stack()``.
1789 Raises:
1790 ValueError: If patch cannot be applied
1791 """
1792 from .index import (
1793 IndexEntry,
1794 index_entry_from_stat,
1795 )
1797 if config is None:
1798 config = r.get_config_stack()
1800 for patch in patches:
1801 # Determine the file path
1802 # For renames/copies without hunks, old_path/new_path may be None
1803 # Use local variables to avoid mutating the patch object
1804 old_path = patch.old_path
1805 new_path = patch.new_path
1807 if new_path is None and old_path is None:
1808 if patch.rename_to is not None:
1809 # Use rename_to for the target path
1810 new_path = patch.rename_to
1811 old_path = patch.rename_from
1812 elif patch.copy_to is not None:
1813 # Use copy_to for the target path
1814 new_path = patch.copy_to
1815 old_path = patch.copy_from
1816 else:
1817 raise ValueError("Patch has no file path")
1819 # Choose path based on operation
1820 file_path: bytes
1821 if new_path is None:
1822 # Deletion
1823 if old_path is None:
1824 raise ValueError("Patch has no file path")
1825 file_path = old_path
1826 elif old_path is None:
1827 # Addition
1828 file_path = new_path
1829 else:
1830 # Modification (use new path)
1831 file_path = new_path
1833 # Strip path components
1834 if strip > 0:
1835 parts = file_path.split(b"/")
1836 if len(parts) > strip:
1837 file_path = b"/".join(parts[strip:])
1839 # Convert to filesystem path
1840 tree_path = file_path
1841 repo_path_bytes = r.path.encode("utf-8") if isinstance(r.path, str) else r.path
1842 fs_path = _validate_patch_target(r, repo_path_bytes, file_path)
1844 # Handle renames and copies
1845 original_lines: list[bytes] | None = None
1846 old_mode: int | None = None
1847 if patch.rename_from is not None and patch.rename_to is not None:
1848 original_lines, old_mode, should_continue = _apply_rename_or_copy(
1849 r,
1850 patch.rename_from,
1851 patch.rename_to,
1852 strip,
1853 patch,
1854 is_rename=True,
1855 cached=cached,
1856 check=check,
1857 config=config,
1858 )
1859 if should_continue:
1860 continue
1861 elif patch.copy_from is not None and patch.copy_to is not None:
1862 original_lines, old_mode, should_continue = _apply_rename_or_copy(
1863 r,
1864 patch.copy_from,
1865 patch.copy_to,
1866 strip,
1867 patch,
1868 is_rename=False,
1869 cached=cached,
1870 check=check,
1871 config=config,
1872 )
1873 if should_continue:
1874 continue
1876 # Read original file content (unless already loaded from rename/copy)
1877 if original_lines is None:
1878 if old_path is None:
1879 # New file
1880 if not cached:
1881 _refuse_existing_target(fs_path, tree_path)
1882 original_lines = []
1883 else:
1884 current = _load_patch_target(
1885 r, fs_path, tree_path, patch.old_mode, config
1886 )
1887 if current is None:
1888 original_lines = []
1889 else:
1890 content, old_mode = current
1891 original_lines = content.splitlines(keepends=True)
1892 new_mode = patch.new_mode or old_mode or patch.old_mode or 0o100644
1894 # Reverse patch if requested
1895 if reverse:
1896 # Swap old and new in hunks
1897 for hunk in patch.hunks:
1898 hunk.old_start, hunk.new_start = hunk.new_start, hunk.old_start
1899 hunk.old_count, hunk.new_count = hunk.new_count, hunk.old_count
1900 # Swap +/- prefixes
1901 reversed_lines = []
1902 for line in hunk.lines:
1903 if line.startswith(b"+"):
1904 reversed_lines.append(b"-" + line[1:])
1905 elif line.startswith(b"-"):
1906 reversed_lines.append(b"+" + line[1:])
1907 else:
1908 reversed_lines.append(line)
1909 hunk.lines = reversed_lines
1911 # Apply the patch
1912 assert original_lines is not None
1913 result: list[bytes] | None
1914 # Deletions don't need the binary data, so treat those like text.
1915 if patch.binary and new_path is not None:
1916 result = _apply_binary_patch(
1917 patch, b"".join(original_lines), reverse
1918 ).splitlines(keepends=True)
1919 else:
1920 result = apply_patch_hunks(patch, original_lines)
1922 if result is None and three_way:
1923 # Try 3-way merge fallback
1924 from .merge import merge_blobs
1926 # Reconstruct base version from the patch
1927 # Base is what you get by taking only the old lines from hunks
1928 base_lines = []
1929 theirs_lines = []
1931 for hunk in patch.hunks:
1932 for line in hunk.lines:
1933 if line.startswith(b"\\"):
1934 # Skip "\ No newline at end of file" markers
1935 continue
1936 elif line.startswith(b" "):
1937 # Context line - in both base and theirs
1938 content = line[1:]
1939 if not content.endswith(b"\n"):
1940 content += b"\n"
1941 base_lines.append(content)
1942 theirs_lines.append(content)
1943 elif line.startswith(b"-"):
1944 # Deletion - only in base
1945 content = line[1:]
1946 if not content.endswith(b"\n"):
1947 content += b"\n"
1948 base_lines.append(content)
1949 elif line.startswith(b"+"):
1950 # Addition - only in theirs
1951 content = line[1:]
1952 if not content.endswith(b"\n"):
1953 content += b"\n"
1954 theirs_lines.append(content)
1956 # Create blobs for merging
1957 base_content = b"".join(base_lines)
1958 ours_content = b"".join(original_lines)
1959 theirs_content = b"".join(theirs_lines)
1961 base_blob = Blob.from_string(base_content) if base_content else None
1962 ours_blob = Blob.from_string(ours_content) if ours_content else None
1963 theirs_blob = Blob.from_string(theirs_content)
1965 # Perform 3-way merge
1966 merged_content, _had_conflicts = merge_blobs(
1967 base_blob, ours_blob, theirs_blob, path=tree_path
1968 )
1970 result = merged_content.splitlines(keepends=True)
1972 # Note: if _had_conflicts is True, the result contains conflict markers
1973 # Git would exit with error code, but we continue processing
1974 elif result is None:
1975 raise PatchApplicationFailure(
1976 f"Patch does not apply to {file_path.decode('utf-8', errors='replace')}"
1977 )
1979 if check:
1980 # Just checking, don't actually apply
1981 continue
1983 # Write result
1984 result_content = b"".join(result)
1986 if new_path is None:
1987 # File deletion
1988 if not cached and os.path.lexists(fs_path):
1989 os.remove(fs_path)
1990 # Remove from index
1991 index = r.open_index(config=config)
1992 if tree_path in index:
1993 del index[tree_path]
1994 index.write()
1995 else:
1996 # File addition or modification
1997 index = r.open_index(config=config)
1998 blob = Blob.from_string(result_content)
1999 r.object_store.add_object(blob)
2001 if not cached:
2002 st = _write_patch_target(fs_path, result_content, new_mode, config)
2003 entry = index_entry_from_stat(st, blob.id, mode=new_mode)
2004 else:
2005 # Create a minimal index entry for cached-only changes
2006 entry = IndexEntry(
2007 ctime=(0, 0),
2008 mtime=(0, 0),
2009 dev=0,
2010 ino=0,
2011 mode=new_mode,
2012 uid=0,
2013 gid=0,
2014 size=len(result_content),
2015 sha=blob.id,
2016 flags=0,
2017 )
2019 index[tree_path] = entry
2021 # Handle cleanup for renames with hunks
2022 if patch.rename_from is not None and patch.rename_to is not None:
2023 # Remove old file after successful rename
2024 old_rename_path = patch.rename_from
2025 if strip > 0:
2026 old_parts = old_rename_path.split(b"/")
2027 if len(old_parts) > strip:
2028 old_rename_path = b"/".join(old_parts[strip:])
2030 old_fs_path = os.path.join(
2031 r.path.encode("utf-8") if isinstance(r.path, str) else r.path,
2032 old_rename_path,
2033 )
2035 if not cached and os.path.lexists(old_fs_path):
2036 os.remove(old_fs_path)
2037 if old_rename_path in index:
2038 del index[old_rename_path]
2040 index.write()
2043def mailinfo(
2044 msg: email.message.Message | BinaryIO | TextIO,
2045 keep_subject: bool = False,
2046 keep_non_patch: bool = False,
2047 encoding: str | None = None,
2048 scissors: bool = False,
2049 message_id: bool = False,
2050) -> MailinfoResult:
2051 """Extract patch information from an email message.
2053 This function parses an email message and extracts commit metadata
2054 (author, email, subject) and separates the commit message from the
2055 patch content, similar to git mailinfo.
2057 Args:
2058 msg: Email message (email.message.Message object) or file handle to read from
2059 keep_subject: If True, keep subject intact without munging (-k)
2060 keep_non_patch: If True, only strip [PATCH] from brackets (-b)
2061 encoding: Character encoding to use (default: detect from message)
2062 scissors: If True, remove everything before scissors line
2063 message_id: If True, include Message-ID in commit message (-m)
2065 Returns:
2066 MailinfoResult with parsed information
2068 Raises:
2069 ValueError: If message is malformed or missing required fields
2070 """
2071 # Parse message if given a file handle
2072 parsed_msg: email.message.Message
2073 if not isinstance(msg, email.message.Message):
2074 if hasattr(msg, "read"):
2075 content = msg.read()
2076 if isinstance(content, bytes):
2077 bparser = email.parser.BytesParser()
2078 parsed_msg = bparser.parsebytes(content)
2079 else:
2080 sparser = email.parser.Parser()
2081 parsed_msg = sparser.parsestr(content)
2082 else:
2083 raise ValueError("msg must be an email.message.Message or file-like object")
2084 else:
2085 parsed_msg = msg
2087 # Detect encoding from message if not specified
2088 if encoding is None:
2089 encoding = parsed_msg.get_content_charset() or "utf-8"
2091 # Extract author information
2092 from_header = parsed_msg.get("From", "")
2093 if not from_header:
2094 raise ValueError("Email message missing 'From' header")
2096 # Parse "Name <email>" format
2097 author_name, author_email = email.utils.parseaddr(from_header)
2098 if not author_email:
2099 raise ValueError(
2100 f"Could not parse email address from 'From' header: {from_header}"
2101 )
2103 # Extract date
2104 date_header = parsed_msg.get("Date")
2105 author_date = date_header if date_header else None
2107 # Extract and process subject
2108 subject = parsed_msg.get("Subject", "")
2109 if not subject:
2110 subject = "(no subject)"
2112 # Convert Header object to string if needed
2113 subject = str(subject)
2115 # Remove newlines from subject
2116 subject = subject.replace("\n", " ").replace("\r", " ")
2117 subject = _munge_subject(subject, keep_subject, keep_non_patch)
2119 # Extract Message-ID if requested
2120 msg_id = None
2121 if message_id:
2122 msg_id = parsed_msg.get("Message-ID")
2124 # Get message body
2125 body = parsed_msg.get_payload(decode=True)
2126 if body is None:
2127 body = b""
2128 elif isinstance(body, str):
2129 body = body.encode(encoding)
2130 elif not isinstance(body, bytes):
2131 # Handle multipart or other types
2132 body = str(body).encode(encoding)
2134 # Split into lines
2135 lines = body.splitlines(keepends=True)
2137 # Handle scissors
2138 scissors_idx = None
2139 if scissors:
2140 scissors_idx = _find_scissors_line(lines)
2141 if scissors_idx is not None:
2142 # Remove everything up to and including scissors line
2143 lines = lines[scissors_idx + 1 :]
2145 # Separate commit message from patch
2146 # Look for the "---" separator that indicates start of diffstat/patch
2147 message_lines: list[bytes] = []
2148 patch_lines: list[bytes] = []
2149 in_patch = False
2151 for line in lines:
2152 if not in_patch and line == b"---\n":
2153 in_patch = True
2154 patch_lines.append(line)
2155 elif in_patch:
2156 # Stop at signature marker "-- "
2157 if line == b"-- \n":
2158 break
2159 patch_lines.append(line)
2160 else:
2161 message_lines.append(line)
2163 # Build commit message
2164 commit_message = b"".join(message_lines).decode(encoding, errors="replace")
2166 # Clean up commit message
2167 commit_message = commit_message.strip()
2169 # Append Message-ID if requested
2170 if message_id and msg_id:
2171 if commit_message:
2172 commit_message += "\n\n"
2173 commit_message += f"Message-ID: {msg_id}"
2175 # Build patch content
2176 patch_content = b"".join(patch_lines).decode(encoding, errors="replace")
2178 return MailinfoResult(
2179 author_name=author_name,
2180 author_email=author_email,
2181 author_date=author_date,
2182 subject=subject,
2183 message=commit_message,
2184 patch=patch_content,
2185 message_id=msg_id,
2186 )