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