Coverage for src/pullapprove/matches.py: 85%
174 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-09-17 14: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 additions = 0
182 deletions = 0
184 def iterate() -> Generator[ScopePathMatch | ScopeCodeMatch]:
185 nonlocal additions, deletions
186 for diff_obj, matches in iterate_diff(configs, diff):
187 # Track additions/deletions during existing iteration
188 if isinstance(diff_obj, DiffCode):
189 if diff_obj.is_addition():
190 additions += 1
191 elif diff_obj.is_deletion():
192 deletions += 1
194 if isinstance(diff_obj, DiffFile) and diff_obj.new_path in configs:
195 config_paths_modified.add(diff_obj.new_path)
196 if isinstance(diff_obj, DiffFile) and diff_obj.old_path in configs:
197 config_paths_modified.add(diff_obj.old_path)
199 yield from matches
201 try:
202 return DiffResults(
203 matches=ChangeMatches.from_config_matches(configs, iterate()),
204 config_paths_modified=list(config_paths_modified),
205 additions=additions,
206 deletions=deletions,
207 )
208 except LargeScaleChangeException:
209 # Get the large scale change config from CODEREVIEW.toml
210 lsc = configs.get_default_large_scale_change()
212 return DiffResults(
213 matches=ChangeMatches.from_large_scale_change(
214 configs=configs,
215 large_scale_change=lsc,
216 ),
217 config_paths_modified=list(config_paths_modified),
218 additions=additions,
219 deletions=deletions,
220 )
223class DiffResults(BaseModel):
224 """Results from analyzing a diff against configs."""
226 model_config = ConfigDict(extra="forbid")
228 matches: ChangeMatches
229 config_paths_modified: list[str] = Field(default_factory=list)
230 additions: int = 0
231 deletions: int = 0
234class ChangeMatches(BaseModel):
235 """
236 The matches for a given diff or set of files.
238 This knows nothing about a pull request (branches, commits, etc.)
239 """
241 model_config = ConfigDict(extra="forbid")
243 # Instead we could do
244 # - scopes
245 # - config
246 # - paths
247 # - code
248 # could add points, reviewers, etc to this
249 # but then we're mixing concerns... looking at raw files will just have empty values?
251 # Three modes are:
252 # - raw files
253 # - raw diff
254 # - pull request (has reviews)
256 configs: dict[str, ConfigModel] = {}
258 # The matching LSC, if there is one.
259 large_scale_change: LargeScaleChangeModel | None = None
261 # All scopes found in the results
262 scopes: dict[str, ScopeModel] = {}
264 # All evaluated paths
265 paths: dict[str, ScopePathMatch] = {}
267 # All code matches
268 code: dict[str, ScopeCodeMatch] = {}
270 def as_dict(self) -> dict[str, Any]:
271 return self.model_dump()
273 def __bool__(self) -> bool:
274 return bool(self.scopes)
276 @classmethod
277 def from_config_matches(
278 cls,
279 configs: CompiledConfigModels,
280 matches: Iterator[ScopePathMatch | ScopeCodeMatch],
281 ) -> ChangeMatches:
282 scopes: dict[str, ScopeModel] = {}
283 paths: dict[str, ScopePathMatch] = {}
284 code: dict[str, ScopeCodeMatch] = {}
286 # Where each scope sits in the configs: which config it came from,
287 # then its position in that config's scope list. Keyed by object
288 # identity, not by name. Two configs may declare the same scope name,
289 # and a name key would give one of them the other's position. The
290 # matched scopes are the same ScopeModel instances these configs hold,
291 # so identity is exact.
292 #
293 # The `scopes` dict below is keyed by name and so still collapses
294 # duplicate names across configs. That is pre-existing; ordering by
295 # identity keeps whichever scope survives that collapse in its own
296 # declared position rather than making it worse.
297 declaration_order = {
298 id(scope): (config_index, scope_index)
299 for config_index, config in enumerate(configs.root.values())
300 for scope_index, scope in enumerate(config.scopes)
301 }
303 for match in matches:
304 # Store seen scopes as we go from all matches
305 for scope in match._scopes:
306 scopes[scope.name] = scope
308 if isinstance(match, ScopePathMatch):
309 if not match._scopes:
310 # Right now we don't care about storing anything that doesn't have scopes.
311 # This prevents an unnecessarily huge dump on big repos or PRs.
312 continue
314 paths[match.path] = match
316 elif isinstance(match, ScopeCodeMatch):
317 code_location_id = match.location_id
319 # Store it in the code results
320 code[code_location_id] = match
322 else:
323 raise TypeError(f"Unknown match type: {match}")
325 # Sort into declaration order. Without this, the order is whichever
326 # scope the first changed file matched, which is a property of the
327 # diff rather than the configs. Consumers that walk scopes in sequence
328 # need declaration order: review requesting counts what the scopes
329 # before this one already asked for, so the order decides who gets
330 # asked. A scope missing from the map sorts last and keeps its
331 # position; the match came from these configs, so this cannot
332 # normally happen.
333 ordered_scopes = {
334 name: scopes[name]
335 for name in sorted(
336 scopes,
337 key=lambda name: declaration_order.get(
338 id(scopes[name]), (len(configs.root), 0)
339 ),
340 )
341 }
343 return cls(
344 large_scale_change=None,
345 scopes=ordered_scopes,
346 paths=paths,
347 code=code,
348 # These are the compiled, effective configs (extends merged, aliases
349 # expanded, paths anchored, PR-disabled scopes dropped) — i.e. what
350 # actually applied to this diff.
351 configs=configs.get_config_models(),
352 )
354 @classmethod
355 def from_large_scale_change(
356 cls,
357 configs: CompiledConfigModels,
358 large_scale_change: LargeScaleChangeModel,
359 ) -> ChangeMatches:
360 return cls(
361 configs=configs.get_config_models(),
362 large_scale_change=large_scale_change,
363 scopes={},
364 paths={},
365 code={},
366 )
369class ScopePathMatch(BaseModel):
370 model_config = ConfigDict(extra="forbid")
372 path: str = Field(min_length=1)
373 scopes: list[str] # Field(min_length=1)
374 # code: list[str] = []
376 # Store this internally during processing (full reference of scope models)
377 _scopes: list[ScopeModel] = []
379 def add_scope(self, scope: ScopeModel) -> None:
380 if not scope.ownership:
381 # Remove any other scopes that don't have special ownership rules
382 # (i.e. we only want one primary scope in the end)
383 self._scopes = [s for s in self._scopes if s.ownership]
385 self._scopes.append(scope)
387 self.scopes = [s.name for s in self._scopes]
390class ScopeCodeMatch(BaseModel):
391 model_config = ConfigDict(extra="forbid")
393 # In a diff match, we could see both sides of the diff, i.e. repeated lines if the before and after both match...
394 path: str = Field(min_length=1)
395 start_line: int
396 end_line: int
397 start_column: int
398 end_column: int
399 scopes: list[str] # Field(min_length=1)
400 location_id: str
402 # Store this internally during processing (full reference of scope models)
403 _scopes: list[ScopeModel] = []
405 def printed_location(self) -> str:
406 if self.start_line == self.end_line:
407 return f"Ln {self.start_line}, Col {self.start_column}-{self.end_column}"
408 else:
409 return f"Ln {self.start_line}-{self.end_line}"
411 def add_scope(self, scope: ScopeModel) -> None:
412 if not scope.ownership:
413 # Remove any other scopes that don't have special ownership rules
414 # (i.e. we only want one primary scope in the end)
415 self._scopes = [s for s in self._scopes if s.ownership]
417 self._scopes.append(scope)
419 self.scopes = [s.name for s in self._scopes]
421 @model_validator(mode="after")
422 def compute_location_id(self) -> ScopeCodeMatch:
423 # only compute if the caller didn't provide one
424 if not self.location_id:
425 loc = {
426 "path": self.path,
427 "start_line": self.start_line,
428 "end_line": self.end_line,
429 "start_column": self.start_column,
430 "end_column": self.end_column,
431 }
432 raw = json.dumps(loc, sort_keys=True, separators=(",", ":")).encode()
433 self.location_id = hashlib.md5(raw).hexdigest()
434 return self
437# how to store what was reviewed? ideally we could be fine-grained, at some point
438# so we need to know who, which scopes, which paths, which codes (location hash) then we can cross reference everything?