Coverage for src/pullapprove/diff.py: 95%
256 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-28 14:54 -0500
1from __future__ import annotations
3import re
4from collections.abc import Callable, Generator, Iterable, Iterator
5from functools import cached_property
6from typing import TYPE_CHECKING
8if TYPE_CHECKING:
9 from .trust import Trust
12class DiffFile:
13 def __init__(self, *, old_path: str, new_path: str):
14 self.old_path = old_path
15 self.new_path = new_path
16 self.hunks: list[DiffHunk] = []
17 # The raw `diff --git` line plus every metadata line before the first
18 # hunk (index/mode/rename/`---`/`+++`), verbatim. The client keys a
19 # hunkless file's review entry on a hash of exactly these lines
20 # (enumerateHunks in review-state.ts), so the file-level trust hash must
21 # be built from the same bytes.
22 self.header_lines: list[str] = []
24 def __repr__(self) -> str:
25 return f"<DiffFile old_path={self.old_path} new_path={self.new_path}>"
27 def is_move(self) -> bool:
28 return self.old_path != self.new_path
31class DiffHunk:
32 def __init__(
33 self,
34 *,
35 old_line: int,
36 old_length: int | None,
37 new_line: int,
38 new_length: int | None,
39 ):
40 self.old_line = old_line
41 self.old_length = old_length
42 self.new_line = new_line
43 self.new_length = new_length
44 self.lines: list[DiffCode] = []
45 self.trust: Trust | None = None
47 @property
48 def is_new_file(self) -> bool:
49 """True when this hunk adds a brand-new file (no old content).
51 Both checks matter: a mid-file pure insertion in a zero-context diff
52 (`@@ -5,0 +6,2 @@`) also has old_length 0, but only a genuinely new
53 file anchors at old line 0 (`@@ -0,0 +1,N @@`)."""
54 return self.old_length == 0 and self.old_line == 0
56 # Cached: a hunk's lines are appended once during parsing and never mutated
57 # after, but the trust rules read these accessors ~12x per hunk as they walk
58 # the rule chain — recomputing the list each time is pure waste.
59 @cached_property
60 def changed_lines(self) -> list[DiffCode]:
61 return [line for line in self.lines if not line.is_context()]
63 @cached_property
64 def added_lines(self) -> list[DiffCode]:
65 return [line for line in self.lines if line.is_addition()]
67 @cached_property
68 def removed_lines(self) -> list[DiffCode]:
69 return [line for line in self.lines if line.is_deletion()]
72class DiffCode:
73 def __init__(
74 self,
75 *,
76 old_line_number: int | None,
77 new_line_number: int | None,
78 content: str,
79 change_type: str,
80 ):
81 self.old_line_number = old_line_number
82 self.new_line_number = new_line_number
83 self.content = content
84 self.change_type = change_type
86 def is_addition(self) -> bool:
87 return self.change_type == "+"
89 def is_deletion(self) -> bool:
90 return self.change_type == "-"
92 def is_context(self) -> bool:
93 return self.change_type == ""
95 @property
96 def line_number(self) -> int:
97 """For backwards compatibility - returns the appropriate line number."""
98 if self.is_deletion():
99 return self.old_line_number or 0
100 return self.new_line_number or 0
102 def __str__(self) -> str:
103 return f"{self.line_number}: {self.change_type or ' '}{self.content}"
105 def __repr__(self) -> str:
106 return f"<DiffCode change_type={self.change_type} old_line={self.old_line_number} new_line={self.new_line_number} content={self.content}>"
108 def raw(self) -> str:
109 return f"{self.change_type or ' '}{self.content}"
112# git C-quote single-char escapes (a quoted diff path may use any of these).
113_GIT_QUOTE_ESCAPES = {
114 '"': '"',
115 "\\": "\\",
116 "t": "\t",
117 "n": "\n",
118 "r": "\r",
119 "f": "\f",
120 "b": "\b",
121 "a": "\a",
122 "v": "\v",
123}
126def _unquote_git_path(inner: str) -> str:
127 """Decode a git C-quoted path (the text between the surrounding quotes),
128 resolving backslash escapes and octal byte escapes to a UTF-8 string."""
129 buf = bytearray()
130 i, n = 0, len(inner)
131 while i < n:
132 ch = inner[i]
133 if ch == "\\" and i + 1 < n:
134 nxt = inner[i + 1]
135 if nxt in _GIT_QUOTE_ESCAPES:
136 buf.extend(_GIT_QUOTE_ESCAPES[nxt].encode())
137 i += 2
138 continue
139 if "0" <= nxt <= "7": # octal byte, e.g. \303 for a UTF-8 lead byte
140 # Consume only consecutive octal digits (max 3) — git emits exactly
141 # 3, but arbitrary stdin might not, and `int("7y", 8)` would raise.
142 end = i + 2
143 while end < n and end < i + 4 and "0" <= inner[end] <= "7":
144 end += 1
145 buf.append(int(inner[i + 1 : end], 8) & 0xFF)
146 i = end
147 continue
148 buf.extend(ch.encode())
149 i += 1
150 return buf.decode("utf-8", "replace")
153def _read_quoted_git_path(rest: str, i: int) -> tuple[str, int] | None:
154 """Read one `"C-quoted"` `a/…` path starting at `rest[i]` (which must be the
155 opening quote). Returns (decoded path, index just past the closing quote), or
156 None for an unterminated quote."""
157 j = i + 1
158 while j < len(rest):
159 if rest[j] == "\\":
160 j += 2 # an escaped char can't close the quote
161 continue
162 if rest[j] == '"':
163 return _unquote_git_path(rest[i + 1 : j]), j + 1
164 j += 1
165 return None
168def _split_equal_git_names(rest: str) -> tuple[str, str] | None:
169 """Split an unquoted `a/PATH b/PATH` header when both sides name the SAME
170 path (a non-rename), returning (path, path).
172 Git doesn't quote spaces, so `diff --git a/Section 1/report.py b/Section
173 1/report.py` is ambiguous to a greedy ` \\w/` split (which lands on the last
174 separator and mis-paths the file). But a non-rename repeats one identical
175 path on both sides, so the split is exactly the center: `<p>/NAME <p>/NAME`
176 with equal-length halves. Resolve it structurally, the way git's own header
177 parser assumes equal names first. Returns None when the halves differ (a
178 rename — handled by the caller's fallback)."""
179 # rest == prefix(2) + NAME + " " + prefix(2) + NAME -> len 5 + 2*len(NAME)
180 if len(rest) < 5 or (len(rest) - 5) % 2 != 0:
181 return None
182 name_len = (len(rest) - 5) // 2
183 left, middle, right = (
184 rest[2 : 2 + name_len],
185 rest[2 + name_len : 5 + name_len],
186 rest[5 + name_len :],
187 )
188 # Both prefixes are one word-char + "/", the separator is a space, and the
189 # two names are identical.
190 if left == right and re.match(r"^\w/$", rest[:2]) and re.match(r"^ \w/$", middle):
191 return left, right
192 return None
195def parse_diff_file_line(line: str) -> DiffFile | None:
196 # Cheap guard: skip the parsing on the vast majority of lines that can't match.
197 if not line.startswith("diff --git "):
198 return None
199 # Streamed lines keep their trailing newline; a name must never absorb it.
200 rest = line[len("diff --git ") :].strip()
201 if '"' not in rest:
202 # A non-rename repeats the same path on both sides; resolve that
203 # exactly (so a space-containing path isn't mis-split). Fall back to the
204 # greedy split for renames — different paths whose spaces make this line
205 # genuinely ambiguous (git also emits `--- a/`/`rename from` for those).
206 if names := _split_equal_git_names(rest):
207 return DiffFile(old_path=names[0], new_path=names[1])
208 match = re.match(r"^\w/(.*) \w/(.*)$", rest)
209 if match:
210 return DiffFile(
211 old_path=match.group(1),
212 new_path=match.group(2),
213 )
214 return None
215 # A name with special/non-ASCII bytes is C-quoted (prefix inside the quotes),
216 # e.g. `diff --git "a/caf\303\251.py" b/plain.txt` — git quotes each side
217 # independently, and a BARE side may still contain spaces. Read both tokens so
218 # the header is recognized — else, mid-filter, a hidden hunk before it would
219 # swallow this file's whole diff — and quoted paths decoded.
220 if rest.startswith('"'):
221 first = _read_quoted_git_path(rest, 0)
222 if first is None or first[1] >= len(rest) or rest[first[1]] != " ":
223 return None
224 old_path, second_start = first[0], first[1] + 1
225 second_raw = rest[second_start:]
226 else:
227 # Bare first + quoted second: the second token starts at the ` "` split,
228 # so a bare first name keeps any spaces it contains.
229 split = rest.find(' "')
230 if split == -1:
231 return None
232 old_path, second_raw = rest[:split], rest[split + 1 :]
233 if second_raw.startswith('"'):
234 second = _read_quoted_git_path(second_raw, 0)
235 if second is None:
236 return None
237 new_path = second[0]
238 else:
239 new_path = second_raw # bare to end of line — spaces stay in the name
240 return DiffFile(
241 old_path=re.sub(r"^\w/", "", old_path),
242 new_path=re.sub(r"^\w/", "", new_path),
243 )
246def parse_diff_hunk_line(line: str) -> DiffHunk | None:
247 # Cheap guard: skip the regex on the vast majority of lines that can't match.
248 if not line.startswith("@@ "):
249 return None
250 match = re.match(r"^@@ -(\d+),?(\d+)? \+(\d+),?(\d+)? @@", line)
251 if match:
252 old_line, old_length, new_line, new_length = match.groups()
253 return DiffHunk(
254 old_line=int(old_line),
255 old_length=int(old_length) if old_length else None,
256 new_line=int(new_line),
257 new_length=int(new_length) if new_length else None,
258 )
259 return None
262def _lookahead(lines: Iterable[str]) -> Generator[tuple[str, bool]]:
263 """Yield each line paired with whether another line follows it."""
264 it = iter(lines)
265 try:
266 prev = next(it)
267 except StopIteration:
268 return
269 for line in it:
270 yield prev, True
271 prev = line
272 yield prev, False
275def _iterate_diff_parts(
276 diff: Iterable[str] | str,
277) -> Generator[DiffFile | DiffHunk | DiffCode]:
278 """Single source of truth for diff parsing.
280 Yields files, hunk headers, and code lines in order. `iterate_diff_parts`
281 filters out the hunk headers to preserve its long-standing
282 `DiffFile | DiffCode` contract; `parse_diff` keeps them to group lines
283 under their hunks.
284 """
285 current_file: DiffFile | None = None
286 current_hunk: DiffHunk | None = None
288 # Track where we are in the hunk as we go.
289 minus_line = plus_line = 0
291 diff_iterator = diff.splitlines() if isinstance(diff, str) else diff
293 # A bare `""` body line is ambiguous: it's either a genuine blank context
294 # line (some hosts strip its leading space) or the trailing artifact that
295 # `some_diff_text.split("\n")` appends when the text ends with a newline
296 # (e.g. the CLI's `diff --hide`, which splits that way on purpose so its
297 # two parsing passes stay aligned — see cli.py). The two are only told
298 # apart by position: the split artifact is always the LAST line of the
299 # whole input, with nothing after it, while a real blank context line is
300 # always followed by more content (more body, the next hunk, the next
301 # file, or at minimum its own file's closing newline). So a `""` only
302 # counts as context when something follows it — `_lookahead` reports that
303 # via `has_next`. Header count mismatches (real diffs, and this test
304 # suite's fixtures, aren't always internally consistent) rule out using
305 # the hunk's declared old/new length for this instead.
306 for raw, has_next in _lookahead(diff_iterator):
307 if new_file := parse_diff_file_line(raw):
308 current_file, current_hunk = new_file, None
309 new_file.header_lines.append(raw)
310 yield new_file
311 elif current_file:
312 if new_hunk := parse_diff_hunk_line(raw):
313 current_hunk = new_hunk
314 minus_line, plus_line = new_hunk.old_line, new_hunk.new_line
315 yield new_hunk
317 # Git can pack the first context line onto the hunk header,
318 # after the closing `@@` (e.g. `@@ -6,7 +6,7 @@ def foo():`).
319 # Emit it as a context line so it isn't dropped.
320 trailing = (
321 raw.split("@@", 2)[-1].lstrip() if raw.count("@@") > 1 else ""
322 )
323 if trailing:
324 yield DiffCode(
325 old_line_number=minus_line,
326 new_line_number=plus_line,
327 content=trailing,
328 change_type="",
329 )
330 minus_line += 1
331 plus_line += 1
332 elif current_hunk:
333 if raw.startswith("+"):
334 yield DiffCode(
335 old_line_number=None,
336 new_line_number=plus_line,
337 content=raw[1:],
338 change_type="+",
339 )
340 plus_line += 1
341 elif raw.startswith("-"):
342 yield DiffCode(
343 old_line_number=minus_line,
344 new_line_number=None,
345 content=raw[1:],
346 change_type="-",
347 )
348 minus_line += 1
349 elif (raw.startswith(" ") or raw == "") and (raw != "" or has_next):
350 # Context exists on both sides (some hosts strip a blank
351 # context line's leading space, hence the `""` case —
352 # matches the JS parsers in file-contents.ts/review-state.ts).
353 yield DiffCode(
354 old_line_number=minus_line,
355 new_line_number=plus_line,
356 content="" if raw == "" else raw[1:],
357 change_type="",
358 )
359 minus_line += 1
360 plus_line += 1
361 elif raw != "" or has_next:
362 # A metadata line between the `diff --git` line and the first
363 # hunk (index/mode/rename/`---`/`+++`). Captured verbatim —
364 # the client does the same (splitDiffByFile's headerLines), so
365 # the two sides hash identical bytes for a hunkless file. The
366 # final `""` a split("\n") appends for a newline-terminated
367 # diff is an artifact, not metadata — same `has_next` test the
368 # context-line branch above uses. (The served diff is never
369 # newline-terminated, so the client never sees that artifact
370 # and parity holds.)
371 current_file.header_lines.append(raw)
374def iterate_diff_parts(
375 diff: Iterator[str] | str,
376) -> Generator[DiffFile | DiffCode]:
377 """Stream a diff as files and code lines (hunk headers omitted)."""
378 for part in _iterate_diff_parts(diff):
379 if not isinstance(part, DiffHunk):
380 yield part
383def parse_diff(diff: Iterator[str] | str) -> list[DiffFile]:
384 """Parse a diff into files, each grouping its hunks and their lines."""
385 files: list[DiffFile] = []
386 current_file: DiffFile | None = None
387 current_hunk: DiffHunk | None = None
389 for part in _iterate_diff_parts(diff):
390 if isinstance(part, DiffFile):
391 current_file, current_hunk = part, None
392 files.append(part)
393 elif isinstance(part, DiffHunk):
394 current_hunk = part
395 if current_file is not None:
396 current_file.hunks.append(part)
397 elif current_hunk is not None:
398 current_hunk.lines.append(part)
400 return files
403def iter_file_hunks(diff: Iterable[str] | str) -> Iterator[tuple[DiffFile, DiffHunk]]:
404 """Stream each completed `(file, hunk)` pair as the diff parses, holding only
405 the current hunk's lines in memory. (`parse_diff` keeps the whole tree; this
406 keeps one hunk — which is all a per-hunk consumer needs.) A new file or hunk,
407 and end-of-input, each flush the hunk that just finished; a hunk only exists
408 after its file, so both are set whenever a pair is yielded. Files without
409 hunks (binary, pure rename, mode-only) are never yielded.
411 This mirrors `parse_diff`'s grouping of the same parts stream; keep the two
412 in sync if that dispatch changes."""
413 file: DiffFile | None = None
414 hunk: DiffHunk | None = None
416 for part in _iterate_diff_parts(diff):
417 if isinstance(part, DiffFile):
418 if file is not None and hunk is not None:
419 yield file, hunk
420 file, hunk = part, None
421 elif isinstance(part, DiffHunk):
422 if file is not None and hunk is not None:
423 yield file, hunk
424 hunk = part
425 elif hunk is not None:
426 hunk.lines.append(part)
428 if file is not None and hunk is not None:
429 yield file, hunk
432def iter_hunkless_files(diff: Iterable[str] | str) -> Iterator[DiffFile]:
433 """Stream each file that finishes with no hunks — a binary, pure-rename, or
434 mode-only change. The complement of `iter_file_hunks`: that yields every
435 (file, hunk) pair and never a hunkless file; this yields only the hunkless
436 files. A file is only known to be hunkless once the next file starts (or the
437 input ends), so each yield happens there — by which point the file's
438 `header_lines` are complete."""
439 file: DiffFile | None = None
440 has_hunks = False
442 for part in _iterate_diff_parts(diff):
443 if isinstance(part, DiffFile):
444 if file is not None and not has_hunks:
445 yield file
446 file, has_hunks = part, False
447 elif isinstance(part, DiffHunk):
448 has_hunks = True
450 if file is not None and not has_hunks:
451 yield file
454def filter_diff_text(
455 diff_text: str,
456 keep_hunk: Callable[[str, int, int], bool],
457) -> str:
458 """Re-emit a unified diff, keeping only the hunks `keep_hunk` accepts.
460 `keep_hunk(new_path, old_line, new_line)` is called once per hunk. Hunk
461 bodies are copied verbatim — the diff is split on "\\n" only (not
462 str.splitlines, which also breaks on "\\r", "\\f", and Unicode separators),
463 and bytes are never rebuilt from the parsed model — so CRLF endings and any
464 control characters in the content survive, and the result stays a valid
465 unified diff.
467 A file whose hunks are all dropped is removed, header and all. A file with
468 no hunks at all — a binary, pure-rename, or mode-only change — has nothing
469 to classify and is never "trusted", so it is kept unchanged. Any preamble
470 before the first file (e.g. a `git show` commit message) passes through.
471 """
472 out: list[str] = []
473 header: list[str] = [] # file-level lines buffered until a hunk is kept
474 new_path: str | None = None
475 in_hunk = False # have we passed this file's first @@ yet?
476 keeping = False # is the current hunk being kept?
478 def flush_unhunked_file() -> None:
479 # A file with header lines but no @@ hunks (binary, pure rename, or
480 # mode-only change) has nothing to classify, so emit it as-is rather
481 # than dropping it. A file whose hunks were dropped has in_hunk set and
482 # is intentionally skipped.
483 if header and not in_hunk:
484 out.extend(header)
486 for line in diff_text.split("\n"):
487 if file := parse_diff_file_line(line):
488 flush_unhunked_file()
489 new_path = file.new_path
490 header = [line]
491 in_hunk = False
492 keeping = False
493 elif new_path is not None and (hunk := parse_diff_hunk_line(line)):
494 in_hunk = True
495 keeping = keep_hunk(new_path, hunk.old_line, hunk.new_line)
496 if keeping:
497 out.extend(header)
498 header = []
499 out.append(line)
500 elif not in_hunk:
501 # File metadata before the first hunk, or preamble before any file.
502 (header if new_path is not None else out).append(line)
503 elif keeping:
504 out.append(line) # a body line of a kept hunk
505 flush_unhunked_file()
507 result = "\n".join(out)
508 # The input's final newline arrives as a trailing "" split element attached
509 # to the last hunk's body; if that hunk was dropped, restore the newline so
510 # the output stays a well-terminated diff.
511 if diff_text.endswith("\n") and result and not result.endswith("\n"):
512 result += "\n"
513 return result