Coverage for src/pullapprove/matches.py: 85%
178 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-10-21 11:55 -0500
1from __future__ import annotations
3import hashlib
4import json
5from collections.abc import Generator, Iterator
6from pathlib import Path
7from typing import Any
9from pydantic import BaseModel, ConfigDict, Field, model_validator
11from .config import (
12 CompiledConfigModels,
13 ConfigModel,
14 LargeScaleChangeModel,
15 ScopeModel,
16)
17from .diff import DiffCode, DiffFile, iterate_diff_parts
18from .exceptions import LargeScaleChangeException
21def match_path(
22 *, path: Path, config: ConfigModel
23) -> tuple[ScopePathMatch, list[ScopeModel]]:
24 path_match = ScopePathMatch(path=str(path), scopes=[])
26 scopes_matching_paths = [
27 scope for scope in config.scopes if scope.matches_path(path)
28 ]
29 code_scopes = [scope for scope in scopes_matching_paths if scope.code]
30 path_scopes = [scope for scope in scopes_matching_paths if not scope.code]
32 # Set the scopes on the path itself
33 for scope in path_scopes:
34 path_match.add_scope(scope)
36 return path_match, code_scopes
39def match_code(
40 *, path: str, code: str, scopes: list[ScopeModel], line_offset: int = 0
41) -> Generator[ScopeCodeMatch]:
42 code_matches: dict[str, ScopeCodeMatch] = {}
44 for scope in scopes:
45 for match in scope.matches_code(code):
46 code_match = ScopeCodeMatch(
47 path=path,
48 start_line=line_offset + match["start_line"],
49 end_line=line_offset + match["end_line"],
50 start_column=match["start_col"],
51 end_column=match["end_col"],
52 scopes=[scope.name],
53 location_id="",
54 )
55 code_match._scopes = [scope]
57 if code_match.location_id in code_matches:
58 # Just add the scopes to it
59 code_matches[code_match.location_id].add_scope(scope)
60 else:
61 code_matches[code_match.location_id] = code_match
63 yield from code_matches.values()
66def match_files(configs: CompiledConfigModels, files: Iterator[str]) -> ChangeMatches:
67 def _iterate() -> Generator[ScopePathMatch | ScopeCodeMatch]:
68 for f in files:
69 file_path = Path(f)
71 config = configs.closest_config(file_path)
73 path_match, code_scopes = match_path(
74 path=file_path,
75 config=config,
76 )
78 # Yield the paths first
79 yield path_match
81 # Then go line by line to find scopes that match lines
82 if code_scopes:
83 try:
84 code = file_path.read_text()
85 yield from match_code(
86 path=str(file_path),
87 code=code,
88 scopes=code_scopes,
89 )
90 except UnicodeDecodeError:
91 # Skip binary files that can't be decoded as text
92 pass
94 return ChangeMatches.from_config_matches(configs, _iterate())
97def iterate_diff(
98 configs: CompiledConfigModels, diff: Iterator[str] | str
99) -> Generator[tuple[DiffFile | DiffCode, list[ScopePathMatch | ScopeCodeMatch]]]:
100 # We can still iterate a diff without configs, just by yield the diff objs
101 if not configs:
102 for diff_obj in iterate_diff_parts(diff):
103 yield diff_obj, []
105 return
107 # Keep track of these as we go and jump between file header
108 # and raw code during iteration
109 check_code_scopes: list[ScopeModel] = []
110 current_code_path = None
112 current_code_diffs = []
114 # TODO get root config here, check diff size as we go and raise exception?
115 # or we need to keep track per LSC? should be a compiled value...
117 def yield_code_diffs() -> Generator[
118 tuple[DiffCode, list[ScopePathMatch | ScopeCodeMatch]]
119 ]:
120 # We're passing the entire diff chunk to see if there's a match inside,
121 # but if there is, it probably won't match EVERY line in the chunk
122 assert current_code_path is not None, "current_code_path must be set"
123 current_code_chunk = "\n".join([code.raw() for code in current_code_diffs])
124 current_code_line_number = current_code_diffs[0].line_number - 1
126 code_matches = match_code(
127 path=current_code_path,
128 code=current_code_chunk,
129 scopes=check_code_scopes,
130 line_offset=current_code_line_number,
131 )
132 code_matches = list(code_matches)
134 for diff_line_index, diff_code in enumerate(current_code_diffs):
135 subcode_matches: list[ScopePathMatch | ScopeCodeMatch] = [
136 code_match
137 for code_match in code_matches
138 if code_match.start_line
139 <= (current_code_line_number + diff_line_index + 1)
140 <= code_match.end_line
141 ]
142 yield diff_code, subcode_matches
144 for diff_obj in iterate_diff_parts(diff):
145 if isinstance(diff_obj, DiffFile):
146 # Yield a code chunk if we finished one
147 if current_code_diffs:
148 yield from yield_code_diffs()
150 current_code_path = None
151 current_code_diffs = []
153 diff_file = diff_obj
154 file_path = Path(diff_file.new_path)
155 config = configs.closest_config(file_path)
157 path_match, code_scopes = match_path(
158 path=file_path,
159 config=config,
160 )
162 current_code_path = str(file_path)
163 check_code_scopes = code_scopes
165 yield diff_obj, [path_match]
166 elif isinstance(diff_obj, DiffCode):
167 if check_code_scopes:
168 # It will be yielded later
169 current_code_diffs.append(diff_obj)
170 else:
171 # Skip all code lines if we don't care about code
172 yield diff_obj, []
174 # Yield the last code chunk we saw
175 if current_code_diffs:
176 yield from yield_code_diffs()
179def match_diff(configs: CompiledConfigModels, diff: Iterator[str] | str) -> DiffResults:
180 config_paths_modified: set[str] = set()
181 changed_paths: set[str] = set()
182 additions = 0
183 deletions = 0
185 def iterate() -> Generator[ScopePathMatch | ScopeCodeMatch]:
186 nonlocal additions, deletions
187 for diff_obj, matches in iterate_diff(configs, diff):
188 # Track additions/deletions during existing iteration
189 if isinstance(diff_obj, DiffCode):
190 if diff_obj.is_addition():
191 additions += 1
192 elif diff_obj.is_deletion():
193 deletions += 1
195 if isinstance(diff_obj, DiffFile):
196 changed_paths.add(diff_obj.new_path)
197 if diff_obj.new_path in configs:
198 config_paths_modified.add(diff_obj.new_path)
199 if diff_obj.old_path in configs:
200 config_paths_modified.add(diff_obj.old_path)
202 yield from matches
204 try:
205 return DiffResults(
206 matches=ChangeMatches.from_config_matches(configs, iterate()),
207 config_paths_modified=list(config_paths_modified),
208 changed_paths=sorted(changed_paths),
209 additions=additions,
210 deletions=deletions,
211 )
212 except LargeScaleChangeException:
213 # Get the large scale change config from CODEREVIEW.toml
214 lsc = configs.get_default_large_scale_change()
216 return DiffResults(
217 matches=ChangeMatches.from_large_scale_change(
218 configs=configs,
219 large_scale_change=lsc,
220 ),
221 config_paths_modified=list(config_paths_modified),
222 additions=additions,
223 deletions=deletions,
224 )
227class DiffResults(BaseModel):
228 """Results from analyzing a diff against configs."""
230 model_config = ConfigDict(extra="forbid")
232 matches: ChangeMatches
233 config_paths_modified: list[str] = Field(default_factory=list)
234 # Every file the diff touched. What a result depends on, config-wise, is
235 # the compiled config governing each of these; empty for a large-scale
236 # change, whose verdict comes from the root config alone and whose
237 # iteration stopped at the threshold.
238 changed_paths: list[str] = Field(default_factory=list)
239 additions: int = 0
240 deletions: int = 0
243class ChangeMatches(BaseModel):
244 """
245 The matches for a given diff or set of files.
247 This knows nothing about a pull request (branches, commits, etc.)
248 """
250 model_config = ConfigDict(extra="forbid")
252 # Instead we could do
253 # - scopes
254 # - config
255 # - paths
256 # - code
257 # could add points, reviewers, etc to this
258 # but then we're mixing concerns... looking at raw files will just have empty values?
260 # Three modes are:
261 # - raw files
262 # - raw diff
263 # - pull request (has reviews)
265 configs: dict[str, ConfigModel] = {}
267 # The matching LSC, if there is one.
268 large_scale_change: LargeScaleChangeModel | None = None
270 # All scopes found in the results
271 scopes: dict[str, ScopeModel] = {}
273 # All evaluated paths
274 paths: dict[str, ScopePathMatch] = {}
276 # All code matches
277 code: dict[str, ScopeCodeMatch] = {}
279 def as_dict(self) -> dict[str, Any]:
280 return self.model_dump()
282 def __bool__(self) -> bool:
283 return bool(self.scopes)
285 @classmethod
286 def from_config_matches(
287 cls,
288 configs: CompiledConfigModels,
289 matches: Iterator[ScopePathMatch | ScopeCodeMatch],
290 ) -> ChangeMatches:
291 scopes: dict[str, ScopeModel] = {}
292 paths: dict[str, ScopePathMatch] = {}
293 code: dict[str, ScopeCodeMatch] = {}
295 # Where each scope sits in the configs: which config it came from,
296 # then its position in that config's scope list. Keyed by object
297 # identity, not by name. Two configs may declare the same scope name,
298 # and a name key would give one of them the other's position. The
299 # matched scopes are the same ScopeModel instances these configs hold,
300 # so identity is exact.
301 #
302 # The `scopes` dict below is keyed by name and so still collapses
303 # duplicate names across configs. That is pre-existing; ordering by
304 # identity keeps whichever scope survives that collapse in its own
305 # declared position rather than making it worse.
306 declaration_order = {
307 id(scope): (config_index, scope_index)
308 for config_index, config in enumerate(configs.root.values())
309 for scope_index, scope in enumerate(config.scopes)
310 }
312 for match in matches:
313 # Store seen scopes as we go from all matches
314 for scope in match._scopes:
315 scopes[scope.name] = scope
317 if isinstance(match, ScopePathMatch):
318 if not match._scopes:
319 # Right now we don't care about storing anything that doesn't have scopes.
320 # This prevents an unnecessarily huge dump on big repos or PRs.
321 continue
323 paths[match.path] = match
325 elif isinstance(match, ScopeCodeMatch):
326 code_location_id = match.location_id
328 # Store it in the code results
329 code[code_location_id] = match
331 else:
332 raise TypeError(f"Unknown match type: {match}")
334 # Sort into declaration order. Without this, the order is whichever
335 # scope the first changed file matched, which is a property of the
336 # diff rather than the configs. Consumers that walk scopes in sequence
337 # need declaration order: review requesting counts what the scopes
338 # before this one already asked for, so the order decides who gets
339 # asked. A scope missing from the map sorts last and keeps its
340 # position; the match came from these configs, so this cannot
341 # normally happen.
342 ordered_scopes = {
343 name: scopes[name]
344 for name in sorted(
345 scopes,
346 key=lambda name: declaration_order.get(
347 id(scopes[name]), (len(configs.root), 0)
348 ),
349 )
350 }
352 return cls(
353 large_scale_change=None,
354 scopes=ordered_scopes,
355 paths=paths,
356 code=code,
357 # These are the compiled, effective configs (extends merged, aliases
358 # expanded, paths anchored, PR-disabled scopes dropped) — i.e. what
359 # actually applied to this diff.
360 configs=configs.get_config_models(),
361 )
363 @classmethod
364 def from_large_scale_change(
365 cls,
366 configs: CompiledConfigModels,
367 large_scale_change: LargeScaleChangeModel,
368 ) -> ChangeMatches:
369 return cls(
370 configs=configs.get_config_models(),
371 large_scale_change=large_scale_change,
372 scopes={},
373 paths={},
374 code={},
375 )
378class ScopePathMatch(BaseModel):
379 model_config = ConfigDict(extra="forbid")
381 path: str = Field(min_length=1)
382 scopes: list[str] # Field(min_length=1)
383 # code: list[str] = []
385 # Store this internally during processing (full reference of scope models)
386 _scopes: list[ScopeModel] = []
388 def add_scope(self, scope: ScopeModel) -> None:
389 if not scope.ownership:
390 # Remove any other scopes that don't have special ownership rules
391 # (i.e. we only want one primary scope in the end)
392 self._scopes = [s for s in self._scopes if s.ownership]
394 self._scopes.append(scope)
396 self.scopes = [s.name for s in self._scopes]
399class ScopeCodeMatch(BaseModel):
400 model_config = ConfigDict(extra="forbid")
402 # In a diff match, we could see both sides of the diff, i.e. repeated lines if the before and after both match...
403 path: str = Field(min_length=1)
404 start_line: int
405 end_line: int
406 start_column: int
407 end_column: int
408 scopes: list[str] # Field(min_length=1)
409 location_id: str
411 # Store this internally during processing (full reference of scope models)
412 _scopes: list[ScopeModel] = []
414 def printed_location(self) -> str:
415 if self.start_line == self.end_line:
416 return f"Ln {self.start_line}, Col {self.start_column}-{self.end_column}"
417 else:
418 return f"Ln {self.start_line}-{self.end_line}"
420 def add_scope(self, scope: ScopeModel) -> None:
421 if not scope.ownership:
422 # Remove any other scopes that don't have special ownership rules
423 # (i.e. we only want one primary scope in the end)
424 self._scopes = [s for s in self._scopes if s.ownership]
426 self._scopes.append(scope)
428 self.scopes = [s.name for s in self._scopes]
430 @model_validator(mode="after")
431 def compute_location_id(self) -> ScopeCodeMatch:
432 # only compute if the caller didn't provide one
433 if not self.location_id:
434 loc = {
435 "path": self.path,
436 "start_line": self.start_line,
437 "end_line": self.end_line,
438 "start_column": self.start_column,
439 "end_column": self.end_column,
440 }
441 raw = json.dumps(loc, sort_keys=True, separators=(",", ":")).encode()
442 self.location_id = hashlib.md5(raw).hexdigest()
443 return self
446# how to store what was reviewed? ideally we could be fine-grained, at some point
447# so we need to know who, which scopes, which paths, which codes (location hash) then we can cross reference everything?