Coverage for src/pullapprove/config.py: 89%
308 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-24 10:59 -0500
1from __future__ import annotations
3import re
4import tomllib
5import warnings
7with warnings.catch_warnings():
8 warnings.simplefilter("ignore", DeprecationWarning)
9 import sre_parse
10from collections.abc import Generator
11from enum import StrEnum
12from pathlib import Path
13from typing import Any
15from pydantic import (
16 BaseModel,
17 ConfigDict,
18 Field,
19 RootModel,
20 field_validator,
21 model_validator,
22)
23from wcmatch import glob
25from .checklists import Checklist
27CONFIG_FILENAME_PREFIX = "CODEREVIEW"
29_REPEAT_OPS = {sre_parse.MAX_REPEAT, sre_parse.MIN_REPEAT}
32def _has_nested_quantifiers(data: Any) -> bool:
33 """Detect patterns like (a+)+ that cause catastrophic backtracking."""
34 for op, av in data:
35 if op in _REPEAT_OPS:
36 if _contains_quantifier(av[2]):
37 return True
38 elif op == sre_parse.SUBPATTERN:
39 if _has_nested_quantifiers(av[-1]):
40 return True
41 elif op == sre_parse.BRANCH:
42 if any(_has_nested_quantifiers(branch) for branch in av[1]):
43 return True
44 return False
47def _contains_quantifier(data: Any) -> bool:
48 for op, av in data:
49 if op in _REPEAT_OPS:
50 return True
51 elif op == sre_parse.SUBPATTERN:
52 if _contains_quantifier(av[-1]):
53 return True
54 elif op == sre_parse.BRANCH:
55 if any(_contains_quantifier(branch) for branch in av[1]):
56 return True
57 return False
60CONFIG_FILENAME = "CODEREVIEW.toml"
63def _matches_branches(branches: list[str], base_branch: str, head_branch: str) -> bool:
64 """Check if the given base/head branches match any of the branch patterns."""
65 if not branches:
66 return True
68 for pattern in branches:
69 splitter = "..." if "..." in pattern else ".."
70 parts = pattern.split(splitter, 1)
72 base_pattern = parts[0]
73 head_pattern = parts[1] if len(parts) > 1 else None
75 base_match = glob.globmatch(base_branch, base_pattern) if base_pattern else True
76 head_match = glob.globmatch(head_branch, head_pattern) if head_pattern else True
78 if base_match and head_match:
79 return True
81 return False
84def _expand_aliases(
85 values: list[str],
86 aliases: dict[str, list[str]],
87 _seen: set[str] | None = None,
88 _path: list[str] | None = None,
89) -> list[str]:
90 """Replace alias references in a list with their mapped values recursively."""
91 if _seen is None:
92 _seen = set()
93 if _path is None:
94 _path = []
96 expanded: list[str] = []
97 for value in values:
98 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
99 if value.startswith("!$"):
100 prefix = "!"
101 alias_ref = value[2:]
102 elif value.startswith("$"):
103 prefix = ""
104 alias_ref = value[1:]
105 else:
106 expanded.append(value)
107 continue
109 if alias_ref in _seen:
110 # Cycle detected, raise an error with the cycle path
111 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
112 raise ValueError(
113 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
114 )
115 if alias_ref in aliases:
116 _seen.add(alias_ref)
117 _path.append(alias_ref)
118 # Recursively expand the alias values
119 nested_expanded = _expand_aliases(aliases[alias_ref], aliases, _seen, _path)
120 if prefix:
121 expanded.extend(prefix + v for v in nested_expanded)
122 else:
123 expanded.extend(nested_expanded)
124 _path.pop()
125 _seen.remove(alias_ref)
127 # Remove duplicates while preserving order
128 return list(dict.fromkeys(expanded))
131class ReviewedForChoices(StrEnum):
132 EMPTY = ""
133 REQUIRED = "required"
134 IGNORED = "ignored"
137class OwnershipChoices(StrEnum):
138 EMPTY = ""
139 APPEND = "append"
140 GLOBAL = "global"
143class ScopeModel(BaseModel):
144 model_config = ConfigDict(extra="forbid")
146 # Required fields
147 name: str = Field(min_length=1)
148 paths: list[str] = Field(min_length=1)
150 # Optional fields
152 # Expanded version of lines could be dict
153 # with fnmatch, regex, exclude patterns, etc?
154 code: list[str] = []
156 # This only filtering field that can't be used with raw diff/files...
157 # If we get into that, the others are:
158 # - labels
159 # - ref (have branches at the root level...)
160 # - statuses
161 # - dates
162 # - body
163 # - title
164 # - other scopes
165 # (this is how I ended up with expressions...
166 # I'm not trying to build a general purpose workflow tool,
167 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
168 authors: list[str] = []
169 branches: list[str] = []
171 # (defaults should be the "empty" values)
172 description: str = ""
173 reviewers: list[str] = []
174 alternates: list[str] = []
175 cc: list[str] = []
177 # Review scoring
178 require: int = 0
179 reviewed_for: ReviewedForChoices = ReviewedForChoices.EMPTY
180 author_value: int = 0
182 # How scopes are combined
183 ownership: OwnershipChoices = OwnershipChoices.EMPTY
185 # Actionable items
186 request: int = 0
187 labels: list[str] = []
188 instructions: str = ""
190 # Approval checklist
191 checklist: Checklist | None = None
193 @field_validator("name", mode="after")
194 @classmethod
195 def validate_name(cls, name: str) -> str:
196 if "," in name:
197 raise ValueError("Scope name cannot contain commas")
198 return name
200 @field_validator("code", mode="after")
201 @classmethod
202 def validate_code_patterns(cls, code: list[str]) -> list[str]:
203 for pattern in code:
204 try:
205 parsed = sre_parse.parse(pattern)
206 except re.error as e:
207 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
208 if _has_nested_quantifiers(parsed):
209 raise ValueError(
210 f"Regex pattern '{pattern}' contains nested quantifiers, "
211 "which can cause catastrophic backtracking."
212 )
213 return code
215 @model_validator(mode="after")
216 def validate_reviewers_for_require(self) -> ScopeModel:
217 all_reviewers = self.reviewers + self.alternates
219 # Skip if wildcard - anyone can review
220 if "*" in all_reviewers:
221 return self
223 # Skip if aliases not yet expanded (will validate again after compilation)
224 if any(r.startswith("$") for r in all_reviewers):
225 return self
227 if len(all_reviewers) < self.require:
228 raise ValueError(
229 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
230 )
231 return self
233 @model_validator(mode="after")
234 def validate_checklist_reviewed_for(self) -> ScopeModel:
235 if self.checklist and self.reviewed_for == ReviewedForChoices.REQUIRED:
236 raise ValueError(
237 "checklist and reviewed_for='required' cannot be used together. "
238 "The checklist already requires explicit scope acknowledgment."
239 )
240 return self
242 def printed_name(self) -> str:
243 match self.ownership:
244 case OwnershipChoices.APPEND:
245 return "+" + self.name
246 case OwnershipChoices.GLOBAL:
247 return "*" + self.name
249 return self.name
251 def __eq__(self, other: Any) -> bool:
252 return self.name == other.name
254 def matches_path(self, path: Path) -> bool:
255 # TODO paths shouldn't start with /
256 return glob.globmatch(
257 path,
258 self.paths,
259 flags=glob.GLOBSTAR
260 | glob.BRACE
261 | glob.NEGATE
262 | glob.IGNORECASE
263 | glob.DOTGLOB,
264 )
266 def matches_code(self, code: str) -> Generator[dict[str, int]]:
267 patterns = getattr(self, "_code_regex_patterns", [])
268 if not patterns:
269 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
270 self._code_regex_patterns = patterns
272 for pattern in patterns:
273 for match in pattern.finditer(code):
274 start_index = match.start()
275 end_index = match.end()
277 start_line = code.count("\n", 0, start_index) + 1
278 start_col = start_index - code.rfind("\n", 0, start_index)
280 end_line = code.count("\n", 0, end_index) + 1
281 end_col = end_index - code.rfind("\n", 0, end_index)
283 yield {
284 "start_line": start_line,
285 "start_col": start_col,
286 "end_line": end_line,
287 "end_col": end_col,
288 }
290 def matches_author(self, author_username: str) -> bool:
291 if not self.authors:
292 # No authors specified, so assume it matches
293 return True
295 author_username_lower = author_username.lower()
297 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
298 authors = [a.lower() for a in self.authors if not a.startswith("!")]
300 if author_username_lower in negated_authors:
301 # If the author is in the negated list, return False
302 return False
304 if not authors:
305 # Negation-only: everyone not negated matches
306 return True
308 if author_username_lower in authors:
309 # If the author is in the authors list, return True
310 return True
312 return False
314 def matches_branches(self, base_branch: str, head_branch: str) -> bool:
315 return _matches_branches(self.branches, base_branch, head_branch)
317 def enabled_for_pullrequest(
318 self, author_username: str, base_branch: str, head_branch: str
319 ) -> bool:
320 # Paths/code are matched during diff parsing,
321 # but we also consider authors and branches in the context of a pull request.
322 return self.matches_author(author_username) and self.matches_branches(
323 base_branch, head_branch
324 )
327class LargeScaleChangeModel(BaseModel):
328 model_config = ConfigDict(extra="forbid")
330 # Note, an LSC only applies to diffs, not raw files,
331 # because we have to know what *changed*.
333 # Pretty similar to a scope, but more manual.
334 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
335 require: int = 1
336 reviewers: list[str] = [] # Field(min_length=1)
337 # min_paths: int = 300
338 # min_lines: int = 3000
339 labels: list[str] = []
340 # really need author value too...?
343class ConfigModel(BaseModel):
344 model_config = ConfigDict(extra="forbid")
346 # Nothing is technically required
347 extends: list[str] = []
348 template: bool = False
349 aliases: dict[str, list[str]] = {}
350 large_scale_change: LargeScaleChangeModel | None = None
351 scopes: list[ScopeModel] = []
353 @field_validator("scopes", mode="after")
354 @classmethod
355 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
356 seen: set[str] = set()
357 for scope in scopes:
358 if scope.name.lower() in seen:
359 raise ValueError(f"Duplicate scope name: {scope.name}")
360 seen.add(scope.name.lower())
362 return scopes
364 @field_validator("extends", mode="before")
365 @classmethod
366 def validate_extends(cls, extends: list[str]) -> list[str]:
367 for i, path in enumerate(extends):
368 basename = Path(path).name
369 if not basename.startswith(CONFIG_FILENAME_PREFIX):
370 raise ValueError(
371 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
372 )
373 return extends
375 def compiled_config(
376 self, config_path: Path, other_configs: ConfigModels
377 ) -> ConfigModel:
378 """
379 Merge extends, replace aliases.
380 """
382 if getattr(self, "_compiled_config", None) is not None:
383 return self._compiled_config
385 # Create a copy of the data from what we have currently
386 compiled_data = self.model_dump()
388 for extend_path in self.extends:
389 if extend_path not in other_configs:
390 raise ValueError(f"Config {extend_path} not found")
392 extended_config = other_configs[extend_path]
393 extended_config.compiled_config(Path(extend_path), other_configs)
395 extended_config_dumped = extended_config.model_dump(
396 include={"aliases", "scopes", "large_scale_change"}
397 )
399 compiled_data["scopes"] = (
400 extended_config_dumped["scopes"] + compiled_data["scopes"]
401 )
402 compiled_data["large_scale_change"] = (
403 compiled_data["large_scale_change"]
404 or extended_config_dumped["large_scale_change"]
405 )
406 compiled_data["aliases"] = (
407 extended_config_dumped["aliases"] | compiled_data["aliases"]
408 )
410 # Expand aliases for any aliasable list fields
411 for scope in compiled_data["scopes"]:
412 for field in [
413 "paths",
414 "code",
415 "authors",
416 "branches",
417 "reviewers",
418 "alternates",
419 "cc",
420 "labels",
421 ]:
422 if field in scope:
423 scope[field] = _expand_aliases(
424 scope[field], compiled_data["aliases"]
425 )
427 if large_scale_change := compiled_data.get("large_scale_change"):
428 for field in ["reviewers", "labels"]:
429 large_scale_change[field] = _expand_aliases(
430 large_scale_change[field],
431 compiled_data["aliases"],
432 )
434 # Create a new config from the merged data
435 self._compiled_config = ConfigModel.from_data(
436 data=compiled_data,
437 path=config_path,
438 )
440 return self._compiled_config
442 @classmethod
443 def from_filesystem(cls, path: Path | str) -> ConfigModel:
444 with open(path, "rb") as f:
445 return cls.from_data(tomllib.load(f), path)
447 @classmethod
448 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
449 return cls.from_data(tomllib.loads(content), path)
451 @classmethod
452 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
453 # config = cls(path)
455 # config.data = data
456 # config = ConfigModel(**config.data)
458 return cls(**data)
461class ConfigModels(RootModel):
462 root: dict[str, ConfigModel]
464 # def __init__(self, configs: dict[str, CodeReviewConfig] = None):
465 # if configs is None:
466 # configs = {}
467 # self = configs
469 # def __repr__(self):
470 # return f"CodeReviewConfigs({self})"
472 @classmethod
473 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
474 """Load configs from a dict of data"""
475 configs = cls(root={})
477 for path, config_data in data.items():
478 config = ConfigModel.from_data(config_data, Path(path))
479 configs.add_config(config, Path(path))
481 return configs
483 @classmethod
484 def from_config_models(cls, models: dict[str, ConfigModel]) -> ConfigModels:
485 """Load configs from a dict of models"""
486 configs = cls(root={})
488 for path, config_model in models.items():
489 # config = ConfigModel.from_model(config_model, Path(path))
490 configs.add_config(config_model, Path(path))
492 return configs
494 def get_config_models(self) -> dict[str, ConfigModel]:
495 return dict(self.root.items())
497 def add_config(self, config: ConfigModel, path: Path) -> None:
498 self.root[str(path)] = config
500 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
501 """Get the root config, which is the first one found in the list"""
502 primary_config = CONFIG_FILENAME
504 if primary_config in self.root:
505 # Compile first so the large-scale-change section gets the same
506 # extends-merging and alias expansion as everything else. Without
507 # this, reviewers like ["$backend"] are matched literally.
508 compiled = self.root[primary_config].compiled_config(
509 Path(primary_config), self
510 )
511 if lsc := compiled.large_scale_change:
512 return lsc
514 return LargeScaleChangeModel()
516 def __bool__(self) -> bool:
517 return bool(self.root)
519 def __getitem__(self, key: str) -> ConfigModel:
520 return self.root[key]
522 def __contains__(self, key: str) -> bool:
523 return key in self.root
525 def __len__(self) -> int:
526 return len(self.root)
528 def compile_closest_config(self, file_path: Path) -> ConfigModel:
529 """Find the closest config file to this file"""
530 for parent in file_path.parents:
531 parent_config_path = str(parent / CONFIG_FILENAME)
533 if parent_config_path in self.root:
534 config = self.root[parent_config_path]
536 if config.template:
537 # Skip templates
538 continue
540 compiled = config.compiled_config(Path(parent_config_path), self)
542 return compiled
544 raise ValueError(f"No config found for {file_path}")
546 def iter_compiled_configs(self) -> Generator[tuple[str, ConfigModel]]:
547 for config_path, config in self.root.items():
548 if config.template:
549 # Skip templates
550 continue
552 yield config_path, config.compiled_config(Path(config_path), self)
554 def num_scopes(self) -> int:
555 """
556 Count the total number of scopes across all configs.
557 """
558 return sum(len(config.scopes) for config in self.root.values())
560 def num_reviewers(self) -> int:
561 """
562 Count the total number of reviewers across all configs.
563 """
564 return sum(
565 len(scope.reviewers)
566 for config in self.root.values()
567 for scope in config.scopes
568 )
570 def filter_for_pullrequest(
571 self,
572 base_branch: str,
573 head_branch: str,
574 author_username: str,
575 ) -> ConfigModels:
576 """
577 Look at all configs (including templates) and filter out
578 scopes based on branches, authors, etc.
579 """
580 filtered_configs = {}
582 for config_path, config in self.root.items():
583 compiled_config = config.compiled_config(Path(config_path), self)
585 # Collect the names of scopes to remove
586 scopes_to_remove = set()
587 for scope in compiled_config.scopes:
588 if not scope.enabled_for_pullrequest(
589 author_username, base_branch, head_branch
590 ):
591 scopes_to_remove.add(scope.name)
593 # Create a filtered copy of the config data
594 filtered_config_data = config.model_dump()
595 filtered_config_data["scopes"] = [
596 scope
597 for scope in filtered_config_data["scopes"]
598 if scope["name"] not in scopes_to_remove
599 ]
601 # Rebuild using the original/modified raw data
602 filtered_configs[config_path] = filtered_config_data
604 return ConfigModels.from_configs_data(filtered_configs)