Coverage for src/pullapprove/config.py: 93%
349 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-24 16:00 -0500
1from __future__ import annotations
3import posixpath
4import re
5import tomllib
6import warnings
8with warnings.catch_warnings():
9 warnings.simplefilter("ignore", DeprecationWarning)
10 import sre_parse
11from collections.abc import Generator
12from enum import StrEnum
13from pathlib import Path
14from typing import Any, Self
16from pydantic import (
17 BaseModel,
18 ConfigDict,
19 Field,
20 RootModel,
21 field_validator,
22 model_validator,
23)
24from wcmatch import glob
26from .checklists import Checklist
28CONFIG_FILENAME_PREFIX = "CODEREVIEW"
30_REPEAT_OPS = {sre_parse.MAX_REPEAT, sre_parse.MIN_REPEAT}
33def _has_nested_quantifiers(data: Any) -> bool:
34 """Detect patterns like (a+)+ that cause catastrophic backtracking."""
35 for op, av in data:
36 if op in _REPEAT_OPS:
37 if _contains_quantifier(av[2]):
38 return True
39 elif op == sre_parse.SUBPATTERN:
40 if _has_nested_quantifiers(av[-1]):
41 return True
42 elif op == sre_parse.BRANCH:
43 if any(_has_nested_quantifiers(branch) for branch in av[1]):
44 return True
45 return False
48def _contains_quantifier(data: Any) -> bool:
49 for op, av in data:
50 if op in _REPEAT_OPS:
51 return True
52 elif op == sre_parse.SUBPATTERN:
53 if _contains_quantifier(av[-1]):
54 return True
55 elif op == sre_parse.BRANCH:
56 if any(_contains_quantifier(branch) for branch in av[1]):
57 return True
58 return False
61CONFIG_FILENAME = "CODEREVIEW.toml"
64def _matches_branches(branches: list[str], base_branch: str, head_branch: str) -> bool:
65 """Check if the given base/head branches match any of the branch patterns."""
66 if not branches:
67 return True
69 for pattern in branches:
70 splitter = "..." if "..." in pattern else ".."
71 parts = pattern.split(splitter, 1)
73 base_pattern = parts[0]
74 head_pattern = parts[1] if len(parts) > 1 else None
76 base_match = glob.globmatch(base_branch, base_pattern) if base_pattern else True
77 head_match = glob.globmatch(head_branch, head_pattern) if head_pattern else True
79 if base_match and head_match:
80 return True
82 return False
85def _expand_aliases(
86 values: list[str],
87 aliases: dict[str, list[str]],
88 _seen: set[str] | None = None,
89 _path: list[str] | None = None,
90) -> list[str]:
91 """Replace alias references in a list with their mapped values recursively."""
92 if _seen is None:
93 _seen = set()
94 if _path is None:
95 _path = []
97 expanded: list[str] = []
98 for value in values:
99 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
100 if value.startswith("!$"):
101 prefix = "!"
102 alias_ref = value[2:]
103 elif value.startswith("$"):
104 prefix = ""
105 alias_ref = value[1:]
106 else:
107 expanded.append(value)
108 continue
110 if alias_ref in _seen:
111 # Cycle detected, raise an error with the cycle path
112 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
113 raise ValueError(
114 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
115 )
116 if alias_ref in aliases:
117 _seen.add(alias_ref)
118 _path.append(alias_ref)
119 # Recursively expand the alias values
120 nested_expanded = _expand_aliases(aliases[alias_ref], aliases, _seen, _path)
121 if prefix:
122 expanded.extend(prefix + v for v in nested_expanded)
123 else:
124 expanded.extend(nested_expanded)
125 _path.pop()
126 _seen.remove(alias_ref)
127 else:
128 # Unknown alias — surface it loudly instead of silently dropping the
129 # reference (a typo'd alias would otherwise vanish reviewers/paths).
130 raise ValueError(f"Unknown alias: {prefix}${alias_ref}")
132 # Remove duplicates while preserving order
133 return list(dict.fromkeys(expanded))
136def _resolve_extends_path(extending_path: str, extends_ref: str) -> str:
137 """Resolve an `extends` reference to a canonical repo-relative config key.
139 - `/x` is repo-root-relative.
140 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending
141 file's directory.
143 Raises if the reference escapes above the repo root.
144 """
145 if extends_ref.startswith("/"):
146 resolved = posixpath.normpath(extends_ref.lstrip("/"))
147 else:
148 base_dir = posixpath.dirname(extending_path)
149 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref))
151 if resolved == ".." or resolved.startswith("../"):
152 raise ValueError(
153 f"Invalid extends path: '{extends_ref}' points above the repo root"
154 )
156 return resolved
159def _anchor_path(base_dir: str, pattern: str) -> str:
160 """Anchor a scope path glob at `base_dir` (the owning config's directory).
162 Scope paths are written relative to the config they live in. A leading `/`
163 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is
164 preserved. With an empty `base_dir` (root config) the pattern is unchanged.
165 """
166 negate = pattern.startswith("!")
167 if negate:
168 pattern = pattern[1:]
170 if pattern.startswith("/"):
171 anchored = pattern.lstrip("/")
172 elif base_dir:
173 anchored = f"{base_dir}/{pattern}"
174 else:
175 anchored = pattern
177 return f"!{anchored}" if negate else anchored
180class ReviewedForChoices(StrEnum):
181 EMPTY = ""
182 REQUIRED = "required"
183 IGNORED = "ignored"
186class OwnershipChoices(StrEnum):
187 EMPTY = ""
188 APPEND = "append"
189 GLOBAL = "global"
192class ScopeModel(BaseModel):
193 model_config = ConfigDict(extra="forbid")
195 # Required fields
196 name: str = Field(min_length=1)
197 paths: list[str] = Field(min_length=1)
199 # Optional fields
201 # Expanded version of lines could be dict
202 # with fnmatch, regex, exclude patterns, etc?
203 code: list[str] = []
205 # This only filtering field that can't be used with raw diff/files...
206 # If we get into that, the others are:
207 # - labels
208 # - ref (have branches at the root level...)
209 # - statuses
210 # - dates
211 # - body
212 # - title
213 # - other scopes
214 # (this is how I ended up with expressions...
215 # I'm not trying to build a general purpose workflow tool,
216 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
217 authors: list[str] = []
218 branches: list[str] = []
220 # (defaults should be the "empty" values)
221 description: str = ""
222 reviewers: list[str] = []
223 alternates: list[str] = []
224 cc: list[str] = []
226 # Review scoring
227 require: int = 0
228 reviewed_for: ReviewedForChoices = ReviewedForChoices.EMPTY
229 author_value: int = 0
231 # How scopes are combined
232 ownership: OwnershipChoices = OwnershipChoices.EMPTY
234 # Actionable items
235 request: int = 0
236 labels: list[str] = []
237 instructions: str = ""
239 # Approval checklist
240 checklist: Checklist | None = None
242 @field_validator("name", mode="after")
243 @classmethod
244 def validate_name(cls, name: str) -> str:
245 if "," in name:
246 raise ValueError("Scope name cannot contain commas")
247 return name
249 @field_validator("code", mode="after")
250 @classmethod
251 def validate_code_patterns(cls, code: list[str]) -> list[str]:
252 for pattern in code:
253 try:
254 parsed = sre_parse.parse(pattern)
255 except re.error as e:
256 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
257 if _has_nested_quantifiers(parsed):
258 raise ValueError(
259 f"Regex pattern '{pattern}' contains nested quantifiers, "
260 "which can cause catastrophic backtracking."
261 )
262 return code
264 @model_validator(mode="after")
265 def validate_reviewers_for_require(self) -> ScopeModel:
266 all_reviewers = self.reviewers + self.alternates
268 # Skip if wildcard - anyone can review
269 if "*" in all_reviewers:
270 return self
272 # Skip if aliases not yet expanded (will validate again after compilation)
273 if any(r.startswith("$") for r in all_reviewers):
274 return self
276 if len(all_reviewers) < self.require:
277 raise ValueError(
278 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
279 )
280 return self
282 @model_validator(mode="after")
283 def validate_checklist_reviewed_for(self) -> ScopeModel:
284 if self.checklist and self.reviewed_for == ReviewedForChoices.REQUIRED:
285 raise ValueError(
286 "checklist and reviewed_for='required' cannot be used together. "
287 "The checklist already requires explicit scope acknowledgment."
288 )
289 return self
291 def printed_name(self) -> str:
292 match self.ownership:
293 case OwnershipChoices.APPEND:
294 return "+" + self.name
295 case OwnershipChoices.GLOBAL:
296 return "*" + self.name
298 return self.name
300 def __eq__(self, other: Any) -> bool:
301 return self.name == other.name
303 def matches_path(self, path: Path) -> bool:
304 # TODO paths shouldn't start with /
305 return glob.globmatch(
306 path,
307 self.paths,
308 flags=glob.GLOBSTAR
309 | glob.BRACE
310 | glob.NEGATE
311 | glob.IGNORECASE
312 | glob.DOTGLOB,
313 )
315 def matches_code(self, code: str) -> Generator[dict[str, int]]:
316 patterns = getattr(self, "_code_regex_patterns", [])
317 if not patterns:
318 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
319 self._code_regex_patterns = patterns
321 for pattern in patterns:
322 for match in pattern.finditer(code):
323 start_index = match.start()
324 end_index = match.end()
326 start_line = code.count("\n", 0, start_index) + 1
327 start_col = start_index - code.rfind("\n", 0, start_index)
329 end_line = code.count("\n", 0, end_index) + 1
330 end_col = end_index - code.rfind("\n", 0, end_index)
332 yield {
333 "start_line": start_line,
334 "start_col": start_col,
335 "end_line": end_line,
336 "end_col": end_col,
337 }
339 def matches_author(self, author_username: str) -> bool:
340 if not self.authors:
341 # No authors specified, so assume it matches
342 return True
344 author_username_lower = author_username.lower()
346 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
347 authors = [a.lower() for a in self.authors if not a.startswith("!")]
349 if author_username_lower in negated_authors:
350 # If the author is in the negated list, return False
351 return False
353 if not authors:
354 # Negation-only: everyone not negated matches
355 return True
357 if author_username_lower in authors:
358 # If the author is in the authors list, return True
359 return True
361 return False
363 def matches_branches(self, base_branch: str, head_branch: str) -> bool:
364 return _matches_branches(self.branches, base_branch, head_branch)
366 def enabled_for_pullrequest(
367 self, author_username: str, base_branch: str, head_branch: str
368 ) -> bool:
369 # Paths/code are matched during diff parsing,
370 # but we also consider authors and branches in the context of a pull request.
371 return self.matches_author(author_username) and self.matches_branches(
372 base_branch, head_branch
373 )
376class LargeScaleChangeModel(BaseModel):
377 model_config = ConfigDict(extra="forbid")
379 # Note, an LSC only applies to diffs, not raw files,
380 # because we have to know what *changed*.
382 # Pretty similar to a scope, but more manual.
383 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
384 require: int = 1
385 reviewers: list[str] = [] # Field(min_length=1)
386 # min_paths: int = 300
387 # min_lines: int = 3000
388 labels: list[str] = []
389 # really need author value too...?
392class ConfigModel(BaseModel):
393 model_config = ConfigDict(extra="forbid")
395 # Nothing is technically required
396 extends: list[str] = []
397 template: bool = False
398 aliases: dict[str, list[str]] = {}
399 large_scale_change: LargeScaleChangeModel | None = None
400 scopes: list[ScopeModel] = []
402 @field_validator("scopes", mode="after")
403 @classmethod
404 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
405 seen: set[str] = set()
406 for scope in scopes:
407 if scope.name.lower() in seen:
408 raise ValueError(f"Duplicate scope name: {scope.name}")
409 seen.add(scope.name.lower())
411 return scopes
413 @field_validator("extends", mode="before")
414 @classmethod
415 def validate_extends(cls, extends: list[str]) -> list[str]:
416 for i, path in enumerate(extends):
417 basename = Path(path).name
418 if not basename.startswith(CONFIG_FILENAME_PREFIX):
419 raise ValueError(
420 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
421 )
422 return extends
424 def compiled_config(
425 self, config_path: Path, other_configs: ConfigModels
426 ) -> ConfigModel:
427 """
428 Resolve `extends` and replace aliases, returning the effective config.
430 Two phases: flatten the whole extends chain into one merged (raw,
431 unexpanded) config, then expand aliases once. Expanding after the full
432 merge is what makes transitive inheritance and cross-chain alias
433 scoping work — an alias defined anywhere in the chain resolves anywhere.
435 Pure function of the raw `self` and `other_configs` (it never reads its
436 own anchored output), so it is safe to call uncached.
437 """
439 compiled_data = self._merged_data(config_path, other_configs)
441 # Expand aliases for any aliasable list fields
442 for scope in compiled_data["scopes"]:
443 for field in [
444 "paths",
445 "code",
446 "authors",
447 "branches",
448 "reviewers",
449 "alternates",
450 "cc",
451 "labels",
452 ]:
453 if field in scope:
454 scope[field] = _expand_aliases(
455 scope[field], compiled_data["aliases"]
456 )
458 if large_scale_change := compiled_data.get("large_scale_change"):
459 for field in ["reviewers", "labels"]:
460 large_scale_change[field] = _expand_aliases(
461 large_scale_change[field],
462 compiled_data["aliases"],
463 )
465 # Anchor each scope's paths at the directory tagged during flattening
466 # (after alias expansion, so any `$path-alias` is resolved first). The
467 # transient tag is popped so it never reaches the model.
468 for scope in compiled_data["scopes"]:
469 anchor_dir = scope.pop("_anchor_dir", "")
470 scope["paths"] = [_anchor_path(anchor_dir, p) for p in scope["paths"]]
472 # The compiled config is the self-contained effective config: extends
473 # are already merged in and aliases already expanded, so drop both. This
474 # keeps stored results lean and makes the compiled form standalone (it
475 # can never dangle on a missing extends target or re-expand differently).
476 compiled_data["extends"] = []
477 compiled_data["aliases"] = {}
479 return ConfigModel.from_data(
480 data=compiled_data,
481 path=config_path,
482 )
484 def _merged_data(
485 self,
486 config_path: Path,
487 other_configs: ConfigModels,
488 _in_progress: list[str] | None = None,
489 _seen: set[str] | None = None,
490 ) -> dict[str, Any]:
491 """
492 Flatten the `extends` chain into one merged, *unexpanded* config dict.
494 Parents are merged before this config (so a child can specialize), with
495 aliases unioned child-wins and the large-scale-change config taken from
496 the child if set else the first parent that defines one.
498 `_in_progress` is the current ancestor path, used to detect circular
499 extends. `_seen` is every config already merged into this flatten, used
500 to merge a shared ancestor only once (diamond dedup).
501 """
502 if _in_progress is None:
503 _in_progress = []
504 if _seen is None:
505 _seen = set()
507 config_path_str = str(config_path)
508 if config_path_str in _in_progress:
509 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
510 config_path_str
511 ]
512 raise ValueError(
513 f"Circular reference detected in extends: {' -> '.join(cycle)}"
514 )
515 _in_progress.append(config_path_str)
517 inherited_scopes: list[dict[str, Any]] = []
518 inherited_aliases: dict[str, list[str]] = {}
519 inherited_lsc: dict[str, Any] | None = None
521 for extend_path in self.extends:
522 resolved_path = _resolve_extends_path(config_path_str, extend_path)
523 if resolved_path not in other_configs:
524 raise ValueError(
525 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
526 )
527 if resolved_path in _seen:
528 # Already merged via another branch (diamond) — skip the dup.
529 continue
531 parent_data = other_configs[resolved_path]._merged_data(
532 Path(resolved_path), other_configs, _in_progress, _seen
533 )
534 inherited_scopes = inherited_scopes + parent_data["scopes"]
535 inherited_aliases = inherited_aliases | parent_data["aliases"]
536 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
538 merged = self.model_dump()
539 merged["scopes"] = inherited_scopes + merged["scopes"]
540 merged["aliases"] = inherited_aliases | merged["aliases"]
541 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
543 # Tag each scope with the directory its paths should anchor at. A scope's
544 # paths are relative to the config that owns it, so the first
545 # non-template config to consume a scope claims it: a non-template's own
546 # scopes (and any it inherits from a template) anchor at its directory,
547 # while a template defers to its consumer. `setdefault` means an
548 # already-tagged scope (from a non-template ancestor) keeps its anchor.
549 if not self.template:
550 base_dir = posixpath.dirname(config_path_str)
551 for scope in merged["scopes"]:
552 scope.setdefault("_anchor_dir", base_dir)
554 _seen.add(config_path_str)
555 _in_progress.pop()
557 return merged
559 @classmethod
560 def from_filesystem(cls, path: Path | str) -> ConfigModel:
561 with open(path, "rb") as f:
562 return cls.from_data(tomllib.load(f), path)
564 @classmethod
565 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
566 return cls.from_data(tomllib.loads(content), path)
568 @classmethod
569 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
570 return cls(**data)
573class _ConfigModelsBase(RootModel):
574 """Shared storage and accessors for a set of configs keyed by repo path."""
576 root: dict[str, ConfigModel]
578 @classmethod
579 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
580 """Build from a dict of already-constructed configs keyed by path."""
581 configs = cls(root={})
582 for path, config_model in models.items():
583 configs.root[str(Path(path))] = config_model
584 return configs
586 def get_config_models(self) -> dict[str, ConfigModel]:
587 return dict(self.root.items())
589 def __bool__(self) -> bool:
590 return bool(self.root)
592 def __getitem__(self, key: str) -> ConfigModel:
593 return self.root[key]
595 def __contains__(self, key: str) -> bool:
596 return key in self.root
598 def __len__(self) -> int:
599 return len(self.root)
602class ConfigModels(_ConfigModelsBase):
603 """Configs exactly as loaded from the repo — extends unresolved, aliases
604 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
606 @classmethod
607 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
608 """Load configs from a dict of parsed config data keyed by path."""
609 configs = cls(root={})
611 for path, config_data in data.items():
612 config = ConfigModel.from_data(config_data, Path(path))
613 configs.add_config(config, Path(path))
615 return configs
617 def add_config(self, config: ConfigModel, path: Path) -> None:
618 self.root[str(path)] = config
620 def compiled(self) -> CompiledConfigModels:
621 """Resolve the whole set into its effective, PR-independent form.
623 Each non-template config is compiled once — extends merged, aliases
624 expanded, paths anchored. Templates are NOT compiled standalone: a
625 template scope may reference an alias the consuming config provides, and
626 its paths anchor at the consumer. They are carried through untouched
627 (folded into each consumer during that consumer's compile, and kept in
628 the set for display).
630 The result is an immutable `CompiledConfigModels` — there is no way to
631 compile it again, so the non-idempotent path anchoring can never
632 double-apply.
633 """
634 effective: dict[str, ConfigModel] = {}
635 for path, config in self.root.items():
636 if config.template:
637 effective[path] = config
638 else:
639 effective[path] = config.compiled_config(Path(path), self)
641 return CompiledConfigModels.from_config_models(effective)
644class CompiledConfigModels(_ConfigModelsBase):
645 """The effective configs used for matching: every non-template config is
646 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
648 def closest_config(self, file_path: Path) -> ConfigModel:
649 """Return the closest non-template config governing this file."""
650 for parent in file_path.parents:
651 parent_config_path = str(parent / CONFIG_FILENAME)
653 if parent_config_path in self.root:
654 config = self.root[parent_config_path]
656 if config.template:
657 # Skip templates
658 continue
660 return config
662 raise ValueError(f"No config found for {file_path}")
664 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
665 """The primary (repo-root) config's large-scale-change section, if any.
667 The primary was compiled by `compiled()`, so its reviewers/labels are
668 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
669 true` repo root is a misconfiguration (templates are meant to be
670 extended, not be the primary); it is passed through uncompiled, so its
671 LSC would read with aliases unexpanded.
672 """
673 if CONFIG_FILENAME in self.root:
674 if lsc := self.root[CONFIG_FILENAME].large_scale_change:
675 return lsc
677 return LargeScaleChangeModel()
679 def filter_for_pullrequest(
680 self,
681 base_branch: str,
682 head_branch: str,
683 author_username: str,
684 ) -> CompiledConfigModels:
685 """
686 Overlay PR-dependent scope gating: drop scopes that branch/author rules
687 disable for this pull request.
689 This is the only PR-dependent step. The configs are already compiled, so
690 each config's scopes are self-contained and dropping one is a plain list
691 filter — no re-inheritance. Templates are passed through (they are never
692 matched directly; their scopes already live in each consumer).
693 """
694 effective: dict[str, ConfigModel] = {}
695 for config_path, config in self.root.items():
696 if config.template:
697 # Templates are never matched directly; pass them through.
698 effective[config_path] = config
699 continue
701 kept_scopes = [
702 scope
703 for scope in config.scopes
704 if scope.enabled_for_pullrequest(
705 author_username, base_branch, head_branch
706 )
707 ]
708 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
710 return CompiledConfigModels.from_config_models(effective)