Coverage for src/pullapprove/config.py: 93%
321 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-24 17:28 -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 _expand_aliases(
65 values: list[str],
66 aliases: dict[str, list[str]],
67 _seen: set[str] | None = None,
68 _path: list[str] | None = None,
69) -> list[str]:
70 """Replace alias references in a list with their mapped values recursively."""
71 if _seen is None:
72 _seen = set()
73 if _path is None:
74 _path = []
76 expanded: list[str] = []
77 for value in values:
78 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
79 if value.startswith("!$"):
80 prefix = "!"
81 alias_ref = value[2:]
82 elif value.startswith("$"):
83 prefix = ""
84 alias_ref = value[1:]
85 else:
86 expanded.append(value)
87 continue
89 if alias_ref in _seen:
90 # Cycle detected, raise an error with the cycle path
91 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
92 raise ValueError(
93 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
94 )
95 if alias_ref in aliases:
96 _seen.add(alias_ref)
97 _path.append(alias_ref)
98 # Recursively expand the alias values
99 nested_expanded = _expand_aliases(aliases[alias_ref], aliases, _seen, _path)
100 if prefix:
101 expanded.extend(prefix + v for v in nested_expanded)
102 else:
103 expanded.extend(nested_expanded)
104 _path.pop()
105 _seen.remove(alias_ref)
106 else:
107 # Unknown alias — surface it loudly instead of silently dropping the
108 # reference (a typo'd alias would otherwise vanish reviewers/paths).
109 raise ValueError(f"Unknown alias: {prefix}${alias_ref}")
111 # Remove duplicates while preserving order
112 return list(dict.fromkeys(expanded))
115def _resolve_extends_path(extending_path: str, extends_ref: str) -> str:
116 """Resolve an `extends` reference to a canonical repo-relative config key.
118 - `/x` is repo-root-relative.
119 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending
120 file's directory.
122 Raises if the reference escapes above the repo root.
123 """
124 if extends_ref.startswith("/"):
125 resolved = posixpath.normpath(extends_ref.lstrip("/"))
126 else:
127 base_dir = posixpath.dirname(extending_path)
128 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref))
130 if resolved == ".." or resolved.startswith("../"):
131 raise ValueError(
132 f"Invalid extends path: '{extends_ref}' points above the repo root"
133 )
135 return resolved
138def _anchor_path(base_dir: str, pattern: str) -> str:
139 """Anchor a scope path glob at `base_dir` (the owning config's directory).
141 Scope paths are written relative to the config they live in. A leading `/`
142 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is
143 preserved. With an empty `base_dir` (root config) the pattern is unchanged.
144 """
145 negate = pattern.startswith("!")
146 if negate:
147 pattern = pattern[1:]
149 if pattern.startswith("/"):
150 anchored = pattern.lstrip("/")
151 elif base_dir:
152 anchored = f"{base_dir}/{pattern}"
153 else:
154 anchored = pattern
156 return f"!{anchored}" if negate else anchored
159class OwnershipChoices(StrEnum):
160 EMPTY = ""
161 APPEND = "append"
162 GLOBAL = "global"
165class ScopeModel(BaseModel):
166 model_config = ConfigDict(extra="forbid")
168 # Required fields
169 name: str = Field(min_length=1)
170 paths: list[str] = Field(min_length=1)
172 # Optional fields
174 # Expanded version of lines could be dict
175 # with fnmatch, regex, exclude patterns, etc?
176 code: list[str] = []
178 # This only filtering field that can't be used with raw diff/files...
179 # If we get into that, the others are:
180 # - labels
181 # - ref
182 # - statuses
183 # - dates
184 # - body
185 # - title
186 # - other scopes
187 # (this is how I ended up with expressions...
188 # I'm not trying to build a general purpose workflow tool,
189 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
190 authors: list[str] = []
192 # (defaults should be the "empty" values)
193 description: str = ""
194 reviewers: list[str] = []
195 alternates: list[str] = []
196 cc: list[str] = []
198 # Review scoring
199 require: int = 0
200 author_value: int = 0
202 # How scopes are combined
203 ownership: OwnershipChoices = OwnershipChoices.EMPTY
205 # Actionable items
206 request: int = 0
207 labels: list[str] = []
208 instructions: str = ""
210 # Approval checklist
211 checklist: Checklist | None = None
213 @field_validator("name", mode="after")
214 @classmethod
215 def validate_name(cls, name: str) -> str:
216 if "," in name:
217 raise ValueError("Scope name cannot contain commas")
218 return name
220 @field_validator("code", mode="after")
221 @classmethod
222 def validate_code_patterns(cls, code: list[str]) -> list[str]:
223 for pattern in code:
224 try:
225 parsed = sre_parse.parse(pattern)
226 except re.error as e:
227 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
228 if _has_nested_quantifiers(parsed):
229 raise ValueError(
230 f"Regex pattern '{pattern}' contains nested quantifiers, "
231 "which can cause catastrophic backtracking."
232 )
233 return code
235 @model_validator(mode="after")
236 def validate_reviewers_for_require(self) -> ScopeModel:
237 all_reviewers = self.reviewers + self.alternates
239 # Skip if wildcard - anyone can review
240 if "*" in all_reviewers:
241 return self
243 # Skip if aliases not yet expanded (will validate again after compilation)
244 if any(r.startswith("$") for r in all_reviewers):
245 return self
247 if len(all_reviewers) < self.require:
248 raise ValueError(
249 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
250 )
251 return self
253 def printed_name(self) -> str:
254 match self.ownership:
255 case OwnershipChoices.APPEND:
256 return "+" + self.name
257 case OwnershipChoices.GLOBAL:
258 return "*" + self.name
260 return self.name
262 def __eq__(self, other: Any) -> bool:
263 return self.name == other.name
265 def matches_path(self, path: Path) -> bool:
266 # TODO paths shouldn't start with /
267 return glob.globmatch(
268 path,
269 self.paths,
270 flags=glob.GLOBSTAR
271 | glob.BRACE
272 | glob.NEGATE
273 | glob.IGNORECASE
274 | glob.DOTGLOB,
275 )
277 def matches_code(self, code: str) -> Generator[dict[str, int]]:
278 patterns = getattr(self, "_code_regex_patterns", [])
279 if not patterns:
280 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
281 self._code_regex_patterns = patterns
283 for pattern in patterns:
284 for match in pattern.finditer(code):
285 start_index = match.start()
286 end_index = match.end()
288 start_line = code.count("\n", 0, start_index) + 1
289 start_col = start_index - code.rfind("\n", 0, start_index)
291 end_line = code.count("\n", 0, end_index) + 1
292 end_col = end_index - code.rfind("\n", 0, end_index)
294 yield {
295 "start_line": start_line,
296 "start_col": start_col,
297 "end_line": end_line,
298 "end_col": end_col,
299 }
301 def matches_author(self, author_username: str) -> bool:
302 if not self.authors:
303 # No authors specified, so assume it matches
304 return True
306 author_username_lower = author_username.lower()
308 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
309 authors = [a.lower() for a in self.authors if not a.startswith("!")]
311 if author_username_lower in negated_authors:
312 # If the author is in the negated list, return False
313 return False
315 if not authors:
316 # Negation-only: everyone not negated matches
317 return True
319 if author_username_lower in authors:
320 # If the author is in the authors list, return True
321 return True
323 return False
326class LargeScaleChangeModel(BaseModel):
327 model_config = ConfigDict(extra="forbid")
329 # Note, an LSC only applies to diffs, not raw files,
330 # because we have to know what *changed*.
332 # Pretty similar to a scope, but more manual.
333 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
334 require: int = 1
335 reviewers: list[str] = [] # Field(min_length=1)
336 # min_paths: int = 300
337 # min_lines: int = 3000
338 labels: list[str] = []
339 # really need author value too...?
342class ConfigModel(BaseModel):
343 model_config = ConfigDict(extra="forbid")
345 # Nothing is technically required
346 extends: list[str] = []
347 template: bool = False
348 aliases: dict[str, list[str]] = {}
349 large_scale_change: LargeScaleChangeModel | None = None
350 scopes: list[ScopeModel] = []
352 @field_validator("scopes", mode="after")
353 @classmethod
354 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
355 seen: set[str] = set()
356 for scope in scopes:
357 if scope.name.lower() in seen:
358 raise ValueError(f"Duplicate scope name: {scope.name}")
359 seen.add(scope.name.lower())
361 return scopes
363 @field_validator("extends", mode="before")
364 @classmethod
365 def validate_extends(cls, extends: list[str]) -> list[str]:
366 for i, path in enumerate(extends):
367 basename = Path(path).name
368 if not basename.startswith(CONFIG_FILENAME_PREFIX):
369 raise ValueError(
370 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
371 )
372 return extends
374 def compiled_config(
375 self, config_path: Path, other_configs: ConfigModels
376 ) -> ConfigModel:
377 """
378 Resolve `extends` and replace aliases, returning the effective config.
380 Two phases: flatten the whole extends chain into one merged (raw,
381 unexpanded) config, then expand aliases once. Expanding after the full
382 merge is what makes transitive inheritance and cross-chain alias
383 scoping work — an alias defined anywhere in the chain resolves anywhere.
385 Pure function of the raw `self` and `other_configs` (it never reads its
386 own anchored output), so it is safe to call uncached.
387 """
389 compiled_data = self._merged_data(config_path, other_configs)
391 # Expand aliases for any aliasable list fields
392 for scope in compiled_data["scopes"]:
393 for field in [
394 "paths",
395 "code",
396 "authors",
397 "reviewers",
398 "alternates",
399 "cc",
400 "labels",
401 ]:
402 if field in scope:
403 scope[field] = _expand_aliases(
404 scope[field], compiled_data["aliases"]
405 )
407 if large_scale_change := compiled_data.get("large_scale_change"):
408 for field in ["reviewers", "labels"]:
409 large_scale_change[field] = _expand_aliases(
410 large_scale_change[field],
411 compiled_data["aliases"],
412 )
414 # Anchor each scope's paths at the directory tagged during flattening
415 # (after alias expansion, so any `$path-alias` is resolved first). The
416 # transient tag is popped so it never reaches the model.
417 for scope in compiled_data["scopes"]:
418 anchor_dir = scope.pop("_anchor_dir", "")
419 scope["paths"] = [_anchor_path(anchor_dir, p) for p in scope["paths"]]
421 # The compiled config is the self-contained effective config: extends
422 # are already merged in and aliases already expanded, so drop both. This
423 # keeps stored results lean and makes the compiled form standalone (it
424 # can never dangle on a missing extends target or re-expand differently).
425 compiled_data["extends"] = []
426 compiled_data["aliases"] = {}
428 return ConfigModel.from_data(
429 data=compiled_data,
430 path=config_path,
431 )
433 def _merged_data(
434 self,
435 config_path: Path,
436 other_configs: ConfigModels,
437 _in_progress: list[str] | None = None,
438 _seen: set[str] | None = None,
439 ) -> dict[str, Any]:
440 """
441 Flatten the `extends` chain into one merged, *unexpanded* config dict.
443 Parents are merged before this config (so a child can specialize), with
444 aliases unioned child-wins and the large-scale-change config taken from
445 the child if set else the first parent that defines one.
447 `_in_progress` is the current ancestor path, used to detect circular
448 extends. `_seen` is every config already merged into this flatten, used
449 to merge a shared ancestor only once (diamond dedup).
450 """
451 if _in_progress is None:
452 _in_progress = []
453 if _seen is None:
454 _seen = set()
456 config_path_str = str(config_path)
457 if config_path_str in _in_progress:
458 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
459 config_path_str
460 ]
461 raise ValueError(
462 f"Circular reference detected in extends: {' -> '.join(cycle)}"
463 )
464 _in_progress.append(config_path_str)
466 inherited_scopes: list[dict[str, Any]] = []
467 inherited_aliases: dict[str, list[str]] = {}
468 inherited_lsc: dict[str, Any] | None = None
470 for extend_path in self.extends:
471 resolved_path = _resolve_extends_path(config_path_str, extend_path)
472 if resolved_path not in other_configs:
473 raise ValueError(
474 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
475 )
476 if resolved_path in _seen:
477 # Already merged via another branch (diamond) — skip the dup.
478 continue
480 parent_data = other_configs[resolved_path]._merged_data(
481 Path(resolved_path), other_configs, _in_progress, _seen
482 )
483 inherited_scopes = inherited_scopes + parent_data["scopes"]
484 inherited_aliases = inherited_aliases | parent_data["aliases"]
485 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
487 merged = self.model_dump()
488 merged["scopes"] = inherited_scopes + merged["scopes"]
489 merged["aliases"] = inherited_aliases | merged["aliases"]
490 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
492 # Tag each scope with the directory its paths should anchor at. A scope's
493 # paths are relative to the config that owns it, so the first
494 # non-template config to consume a scope claims it: a non-template's own
495 # scopes (and any it inherits from a template) anchor at its directory,
496 # while a template defers to its consumer. `setdefault` means an
497 # already-tagged scope (from a non-template ancestor) keeps its anchor.
498 if not self.template:
499 base_dir = posixpath.dirname(config_path_str)
500 for scope in merged["scopes"]:
501 scope.setdefault("_anchor_dir", base_dir)
503 _seen.add(config_path_str)
504 _in_progress.pop()
506 return merged
508 @classmethod
509 def from_filesystem(cls, path: Path | str) -> ConfigModel:
510 with open(path, "rb") as f:
511 return cls.from_data(tomllib.load(f), path)
513 @classmethod
514 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
515 return cls.from_data(tomllib.loads(content), path)
517 @classmethod
518 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
519 return cls(**data)
522class _ConfigModelsBase(RootModel):
523 """Shared storage and accessors for a set of configs keyed by repo path."""
525 root: dict[str, ConfigModel]
527 @classmethod
528 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
529 """Build from a dict of already-constructed configs keyed by path."""
530 configs = cls(root={})
531 for path, config_model in models.items():
532 configs.root[str(Path(path))] = config_model
533 return configs
535 def get_config_models(self) -> dict[str, ConfigModel]:
536 return dict(self.root.items())
538 def __bool__(self) -> bool:
539 return bool(self.root)
541 def __getitem__(self, key: str) -> ConfigModel:
542 return self.root[key]
544 def __contains__(self, key: str) -> bool:
545 return key in self.root
547 def __len__(self) -> int:
548 return len(self.root)
551class ConfigModels(_ConfigModelsBase):
552 """Configs exactly as loaded from the repo — extends unresolved, aliases
553 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
555 @classmethod
556 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
557 """Load configs from a dict of parsed config data keyed by path."""
558 configs = cls(root={})
560 for path, config_data in data.items():
561 config = ConfigModel.from_data(config_data, Path(path))
562 configs.add_config(config, Path(path))
564 return configs
566 def add_config(self, config: ConfigModel, path: Path) -> None:
567 self.root[str(path)] = config
569 def compiled(self) -> CompiledConfigModels:
570 """Resolve the whole set into its effective, PR-independent form.
572 Each non-template config is compiled once — extends merged, aliases
573 expanded, paths anchored. Templates are NOT compiled standalone: a
574 template scope may reference an alias the consuming config provides, and
575 its paths anchor at the consumer. They are carried through untouched
576 (folded into each consumer during that consumer's compile, and kept in
577 the set for display).
579 The result is an immutable `CompiledConfigModels` — there is no way to
580 compile it again, so the non-idempotent path anchoring can never
581 double-apply.
582 """
583 effective: dict[str, ConfigModel] = {}
584 for path, config in self.root.items():
585 if config.template:
586 effective[path] = config
587 else:
588 effective[path] = config.compiled_config(Path(path), self)
590 return CompiledConfigModels.from_config_models(effective)
593class CompiledConfigModels(_ConfigModelsBase):
594 """The effective configs used for matching: every non-template config is
595 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
597 def closest_config(self, file_path: Path) -> ConfigModel:
598 """Return the closest non-template config governing this file."""
599 for parent in file_path.parents:
600 parent_config_path = str(parent / CONFIG_FILENAME)
602 if parent_config_path in self.root:
603 config = self.root[parent_config_path]
605 if config.template:
606 # Skip templates
607 continue
609 return config
611 raise ValueError(f"No config found for {file_path}")
613 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
614 """The primary (repo-root) config's large-scale-change section, if any.
616 The primary was compiled by `compiled()`, so its reviewers/labels are
617 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
618 true` repo root is a misconfiguration (templates are meant to be
619 extended, not be the primary); it is passed through uncompiled, so its
620 LSC would read with aliases unexpanded.
621 """
622 if CONFIG_FILENAME in self.root:
623 if lsc := self.root[CONFIG_FILENAME].large_scale_change:
624 return lsc
626 return LargeScaleChangeModel()
628 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels:
629 """
630 Overlay PR-dependent scope gating: drop scopes that author rules disable
631 for this pull request.
633 This is the only PR-dependent step. The configs are already compiled, so
634 each config's scopes are self-contained and dropping one is a plain list
635 filter — no re-inheritance. Templates are passed through (they are never
636 matched directly; their scopes already live in each consumer).
637 """
638 effective: dict[str, ConfigModel] = {}
639 for config_path, config in self.root.items():
640 if config.template:
641 # Templates are never matched directly; pass them through.
642 effective[config_path] = config
643 continue
645 kept_scopes = [
646 scope
647 for scope in config.scopes
648 if scope.matches_author(author_username)
649 ]
650 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
652 return CompiledConfigModels.from_config_models(effective)