Coverage for src/pullapprove/config.py: 95%
436 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-15 09:22 -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"
64_TEAM_REF_SEGMENT = r"[a-zA-Z0-9][a-zA-Z0-9\-_.]*"
65_TEAM_REF_RE = re.compile(rf"^{_TEAM_REF_SEGMENT}(/{_TEAM_REF_SEGMENT})+$")
67# Fields that hold reviewer identities (plain usernames, `$aliases`, and
68# `@team` refs) rather than paths/code/labels. Team refs only expand here.
69USER_LIST_FIELDS = ("authors", "reviewers", "alternates", "cc")
71# Roster fields where "!" performs compile-time subtraction (remove from the
72# resolved list) rather than surviving as a match-time predicate. `authors`
73# is deliberately excluded — its "!" entries are consumed by `matches_author`.
74ROSTER_FIELDS = ("reviewers", "alternates", "cc")
77def _split_team_ref(value: str) -> tuple[str, str] | None:
78 """Split a value into `(prefix, ref)` if it has team-reference shape
79 (`@org/team` or `!@org/team`), `prefix` being `"!"` or `""`. Returns
80 `None` for anything else, so callers fall back to their own handling of
81 plain values.
82 """
83 if value.startswith("!@"):
84 return "!", value[2:]
85 if value.startswith("@"):
86 return "", value[1:]
87 return None
90def is_unexpanded_ref(value: str) -> bool:
91 """True if the value is a `$alias` or `@team` reference (optionally
92 negated with a leading `!`) that has not been expanded — as opposed to a
93 plain username. Only an offline compile (`teams=None`) leaves such values
94 in user-list fields.
95 """
96 return value.removeprefix("!").startswith(("$", "@"))
99def _validate_team_refs(values: list[str]) -> list[str]:
100 """Team references (`@org/team`, `!@org/team`) need at least two
101 slash-separated segments. This only checks shape — membership is resolved
102 later, at compile time, against the caller-provided `teams` mapping.
104 Any other value containing "@" is rejected too — that shape is reserved
105 for team references (and, in future, email-style identifiers), so it
106 can't be confused with a plain username.
107 """
108 for value in values:
109 split = _split_team_ref(value)
110 if split is None:
111 if "@" in value:
112 raise ValueError(
113 f"Invalid value '{value}': email addresses are not "
114 "supported here — use the platform username"
115 )
116 continue
118 _prefix, ref = split
119 if not _TEAM_REF_RE.match(ref):
120 raise ValueError(
121 f"Invalid team reference '{value}': team references need the "
122 "'org/team' form"
123 )
124 return values
127def _expand_team_ref(
128 ref: str, teams: dict[str, list[str]] | None, prefix: str
129) -> list[str]:
130 """Expand a single team reference (without its `@`/`!@`) to member usernames.
132 `teams=None` is the offline/CLI mode: the library never calls out to
133 GitHub/GitLab itself, so with no mapping provided the reference passes
134 through unexpanded rather than erroring. With a `teams` mapping (even an
135 empty one), an unresolvable ref is a loud config error, matching how
136 unknown `$aliases` are handled.
137 """
138 if teams is None:
139 return [f"{prefix}@{ref}"]
141 members = teams.get(ref.lower())
142 if members is None:
143 raise ValueError(f"Unknown team: {prefix}@{ref}")
145 return [f"{prefix}{member}" for member in members]
148def _expand_aliases(
149 values: list[str],
150 aliases: dict[str, list[str]],
151 teams: dict[str, list[str]] | None = None,
152 expand_teams: bool = False,
153 _seen: set[str] | None = None,
154 _path: list[str] | None = None,
155) -> list[str]:
156 """Replace alias references in a list with their mapped values recursively.
158 Team references (`@org/team`) are only expanded when `expand_teams` is
159 True — user-list fields (reviewers, alternates, authors, cc). Elsewhere
160 (paths, code, labels) a leading `@` is a literal string, e.g. npm-style
161 scoped path patterns like `@vendor/pkg/**`. Teams are always leaf nodes —
162 their values are plain usernames, never other refs — so expanding one
163 never recurses further.
164 """
165 if _seen is None:
166 _seen = set()
167 if _path is None:
168 _path = []
170 expanded: list[str] = []
171 for value in values:
172 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
173 if value.startswith("!$"):
174 prefix = "!"
175 alias_ref = value[2:]
176 elif value.startswith("$"):
177 prefix = ""
178 alias_ref = value[1:]
179 elif expand_teams and (team_ref := _split_team_ref(value)) is not None:
180 team_prefix, ref = team_ref
181 expanded.extend(_expand_team_ref(ref=ref, teams=teams, prefix=team_prefix))
182 continue
183 else:
184 expanded.append(value)
185 continue
187 if alias_ref in _seen:
188 # Cycle detected, raise an error with the cycle path
189 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
190 raise ValueError(
191 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
192 )
193 if alias_ref in aliases:
194 _seen.add(alias_ref)
195 _path.append(alias_ref)
196 # Recursively expand the alias values
197 nested_expanded = _expand_aliases(
198 aliases[alias_ref],
199 aliases=aliases,
200 teams=teams,
201 expand_teams=expand_teams,
202 _seen=_seen,
203 _path=_path,
204 )
205 if prefix:
206 expanded.extend(prefix + v for v in nested_expanded)
207 else:
208 expanded.extend(nested_expanded)
209 _path.pop()
210 _seen.remove(alias_ref)
211 else:
212 # Unknown alias — surface it loudly instead of silently dropping the
213 # reference (a typo'd alias would otherwise vanish reviewers/paths).
214 raise ValueError(f"Unknown alias: {prefix}${alias_ref}")
216 # Remove duplicates while preserving order
217 return list(dict.fromkeys(expanded))
220def _apply_negations(values: list[str]) -> list[str]:
221 """Compile-time subtraction for roster fields (`ROSTER_FIELDS`): a "!name"
222 entry removes "name" from the resolved list instead of surviving as a
223 match-time rule (contrast `authors`, handled by `matches_author`).
225 A list still holding an unexpanded `@team` ref (offline compile, i.e.
226 teams=None) is returned untouched — it isn't fully resolved yet, so
227 subtraction can't run, and the partially-resolved config must round-trip
228 unchanged. `$aliases` must already be expanded by the caller.
230 The wildcard+negation error fires BEFORE that early return: it's a check
231 on the written form (`"*"` alongside any `"!"` entry, team ref or not),
232 so an offline compile must reject it the same way the server will — a
233 `pullapprove check` that passes locally can't then error in production.
234 """
235 if "*" in values and any(v.startswith("!") for v in values):
236 raise ValueError(
237 'Negation cannot be combined with "*" in reviewers/alternates/cc '
238 "(wildcard exclusion is not supported)"
239 )
241 if any(_split_team_ref(v) is not None for v in values):
242 return values
244 negations = {v[1:].lower() for v in values if v.startswith("!")}
245 return [v for v in values if not v.startswith("!") and v.lower() not in negations]
248def _validate_review_counts(label: str, data: dict[str, Any]) -> None:
249 """Compile-time rejection of negative require/request/author_value.
250 Shared by scopes and large_scale_change (which has no
251 request/author_value — the `.get` defaults pass trivially there).
253 These are compile-time checks (not field validators) because negative
254 values were previously accepted, so compiled configs stored inside old
255 processing results must keep parsing (where they keep their old
256 behavior: a negative require always passed, a negative request
257 requested nobody).
258 """
259 for field in ("require", "request", "author_value"):
260 if data.get(field, 0) < 0:
261 raise ValueError(f"{label}: {field} cannot be negative")
264def _resolve_extends_path(extending_path: str, extends_ref: str) -> str:
265 """Resolve an `extends` reference to a canonical repo-relative config key.
267 - `/x` is repo-root-relative.
268 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending
269 file's directory.
271 Raises if the reference escapes above the repo root.
272 """
273 if extends_ref.startswith("/"):
274 resolved = posixpath.normpath(extends_ref.lstrip("/"))
275 else:
276 base_dir = posixpath.dirname(extending_path)
277 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref))
279 if resolved == ".." or resolved.startswith("../"):
280 raise ValueError(
281 f"Invalid extends path: '{extends_ref}' points above the repo root"
282 )
284 return resolved
287def _anchor_path(base_dir: str, pattern: str) -> str:
288 """Anchor a scope path glob at `base_dir` (the owning config's directory).
290 Scope paths are written relative to the config they live in. A leading `/`
291 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is
292 preserved. With an empty `base_dir` (root config) the pattern is unchanged.
293 """
294 negate = pattern.startswith("!")
295 if negate:
296 pattern = pattern[1:]
298 if pattern.startswith("/"):
299 anchored = pattern.lstrip("/")
300 elif base_dir:
301 anchored = f"{base_dir}/{pattern}"
302 else:
303 anchored = pattern
305 return f"!{anchored}" if negate else anchored
308class OwnershipChoices(StrEnum):
309 EMPTY = ""
310 APPEND = "append"
311 GLOBAL = "global"
314class ScopeModel(BaseModel):
315 model_config = ConfigDict(extra="forbid")
317 # Required fields
318 name: str = Field(min_length=1)
319 paths: list[str] = Field(min_length=1)
321 # Optional fields
323 # Expanded version of lines could be dict
324 # with fnmatch, regex, exclude patterns, etc?
325 code: list[str] = []
327 # This only filtering field that can't be used with raw diff/files...
328 # If we get into that, the others are:
329 # - labels
330 # - ref
331 # - statuses
332 # - dates
333 # - body
334 # - title
335 # - other scopes
336 # (this is how I ended up with expressions...
337 # I'm not trying to build a general purpose workflow tool,
338 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
339 authors: list[str] = []
341 # (defaults should be the "empty" values)
342 description: str = ""
343 reviewers: list[str] = []
344 alternates: list[str] = []
345 cc: list[str] = []
347 # Review scoring
348 # Negative values are rejected at compile time (_validate_review_counts)
349 require: int = 0
350 author_value: int = 0
352 # How scopes are combined
353 ownership: OwnershipChoices = OwnershipChoices.EMPTY
355 # Actionable items
356 request: int = 0
357 labels: list[str] = []
358 instructions: str = ""
360 # Approval checklist
361 checklist: Checklist | None = None
363 @field_validator("name", mode="after")
364 @classmethod
365 def validate_name(cls, name: str) -> str:
366 if "," in name:
367 raise ValueError("Scope name cannot contain commas")
368 return name
370 @field_validator(*USER_LIST_FIELDS, mode="after")
371 @classmethod
372 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
373 return _validate_team_refs(values)
375 @field_validator("code", mode="after")
376 @classmethod
377 def validate_code_patterns(cls, code: list[str]) -> list[str]:
378 for pattern in code:
379 try:
380 parsed = sre_parse.parse(pattern)
381 except re.error as e:
382 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
383 if _has_nested_quantifiers(parsed):
384 raise ValueError(
385 f"Regex pattern '{pattern}' contains nested quantifiers, "
386 "which can cause catastrophic backtracking."
387 )
388 return code
390 @model_validator(mode="after")
391 def validate_reviewers_for_require(self) -> ScopeModel:
392 all_reviewers = self.reviewers + self.alternates
394 # Skip if wildcard - anyone can review
395 if "*" in all_reviewers:
396 return self
398 # Skip if aliases or team refs (possibly negated with "!") are not yet
399 # expanded. Re-validation after compilation only re-checks refs that
400 # actually resolve — an offline compile (teams=None) leaves refs
401 # unexpanded, so this same skip fires again there too.
402 if any(is_unexpanded_ref(r) for r in all_reviewers):
403 return self
405 if len(all_reviewers) < self.require:
406 raise ValueError(
407 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
408 )
409 return self
411 def author_points(self, author_username: str) -> int:
412 """Author points only count if the author is explicitly listed as a
413 reviewer (a wildcard is not converted to usernames)."""
414 if author_username.lower() in {r.lower() for r in self.reviewers}:
415 return self.author_value
416 return 0
418 def unsolvable_reason(self, author_username: str) -> str | None:
419 """
420 Explain why this scope can never pass for a PR authored by this user,
421 or None if it can.
423 Messages must stay under ~130 chars: they flow into the git-host
424 commit status description, which the adapters slice to 140.
425 """
426 if "*" in self.reviewers:
427 # Anyone can review, so any require is satisfiable
428 return None
430 # Count eligible reviewers (excluding author who can't self-approve)
431 eligible_reviewers = {r.lower() for r in self.reviewers + self.alternates} - {
432 author_username.lower()
433 }
434 max_possible_points = len(eligible_reviewers) + self.author_points(
435 author_username
436 )
438 if self.require > 0 and max_possible_points < self.require:
439 if not eligible_reviewers:
440 return (
441 "PR author is the only reviewer/alternate and cannot self-approve"
442 )
443 return f"require={self.require} but only {max_possible_points} possible approvals (excluding author)"
445 return None
447 def printed_name(self) -> str:
448 match self.ownership:
449 case OwnershipChoices.APPEND:
450 return "+" + self.name
451 case OwnershipChoices.GLOBAL:
452 return "*" + self.name
454 return self.name
456 def __eq__(self, other: Any) -> bool:
457 return self.name == other.name
459 def matches_path(self, path: Path) -> bool:
460 # TODO paths shouldn't start with /
461 return glob.globmatch(
462 path,
463 self.paths,
464 flags=glob.GLOBSTAR
465 | glob.BRACE
466 | glob.NEGATE
467 | glob.IGNORECASE
468 | glob.DOTGLOB,
469 )
471 def matches_code(self, code: str) -> Generator[dict[str, int]]:
472 patterns = getattr(self, "_code_regex_patterns", [])
473 if not patterns:
474 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
475 self._code_regex_patterns = patterns
477 for pattern in patterns:
478 for match in pattern.finditer(code):
479 start_index = match.start()
480 end_index = match.end()
482 start_line = code.count("\n", 0, start_index) + 1
483 start_col = start_index - code.rfind("\n", 0, start_index)
485 end_line = code.count("\n", 0, end_index) + 1
486 end_col = end_index - code.rfind("\n", 0, end_index)
488 yield {
489 "start_line": start_line,
490 "start_col": start_col,
491 "end_line": end_line,
492 "end_col": end_col,
493 }
495 def matches_author(self, author_username: str) -> bool:
496 if not self.authors:
497 # No authors specified, so assume it matches
498 return True
500 author_username_lower = author_username.lower()
502 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
503 authors = [a.lower() for a in self.authors if not a.startswith("!")]
505 if author_username_lower in negated_authors:
506 # If the author is in the negated list, return False
507 return False
509 if not authors:
510 # Negation-only: everyone not negated matches
511 return True
513 if author_username_lower in authors:
514 # If the author is in the authors list, return True
515 return True
517 return False
520class LargeScaleChangeModel(BaseModel):
521 model_config = ConfigDict(extra="forbid")
523 # Note, an LSC only applies to diffs, not raw files,
524 # because we have to know what *changed*.
526 # Pretty similar to a scope, but more manual.
527 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
528 # Negative values are rejected at compile time (_validate_review_counts)
529 require: int = 1
530 reviewers: list[str] = [] # Field(min_length=1)
531 # min_paths: int = 300
532 # min_lines: int = 3000
533 labels: list[str] = []
534 # really need author value too...?
536 @field_validator("reviewers", mode="after")
537 @classmethod
538 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
539 return _validate_team_refs(values)
541 def unsolvable_reason(self, author_username: str) -> str | None:
542 """
543 Explain why this LSC can never pass for a PR authored by this user,
544 or None if it can.
546 Messages must stay under ~130 chars: they flow into the git-host
547 commit status description, which the adapters slice to 140.
548 """
549 if "*" in self.reviewers:
550 # Anyone can review, so any require is satisfiable
551 return None
553 if not self.reviewers:
554 # An empty roster is "configuration required", which
555 # process_large_scale_change reports on its own
556 return None
558 # Count eligible reviewers (excluding author who can't self-approve)
559 eligible_reviewers = {r.lower() for r in self.reviewers} - {
560 author_username.lower()
561 }
562 if self.require > 0 and len(eligible_reviewers) < self.require:
563 if not eligible_reviewers:
564 return "the PR author is the only reviewer and cannot self-approve"
565 return f"require={self.require} but only {len(eligible_reviewers)} possible approvals (excluding author)"
567 return None
570class ConfigModel(BaseModel):
571 model_config = ConfigDict(extra="forbid")
573 # Nothing is technically required
574 extends: list[str] = []
575 template: bool = False
576 aliases: dict[str, list[str]] = {}
577 large_scale_change: LargeScaleChangeModel | None = None
578 scopes: list[ScopeModel] = []
580 @field_validator("scopes", mode="after")
581 @classmethod
582 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
583 seen: set[str] = set()
584 for scope in scopes:
585 if scope.name.lower() in seen:
586 raise ValueError(f"Duplicate scope name: {scope.name}")
587 seen.add(scope.name.lower())
589 return scopes
591 @field_validator("extends", mode="before")
592 @classmethod
593 def validate_extends(cls, extends: list[str]) -> list[str]:
594 for i, path in enumerate(extends):
595 basename = Path(path).name
596 if not basename.startswith(CONFIG_FILENAME_PREFIX):
597 raise ValueError(
598 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
599 )
600 return extends
602 def compiled_config(
603 self,
604 config_path: Path,
605 other_configs: ConfigModels,
606 teams: dict[str, list[str]] | None = None,
607 ) -> ConfigModel:
608 """
609 Resolve `extends` and replace aliases, returning the effective config.
611 Two phases: flatten the whole extends chain into one merged (raw,
612 unexpanded) config, then expand aliases once. Expanding after the full
613 merge is what makes transitive inheritance and cross-chain alias
614 scoping work — an alias defined anywhere in the chain resolves anywhere.
616 `teams` maps team refs (any case, without the leading `@`) to member
617 usernames, and is only consulted for user-list fields (reviewers,
618 alternates, authors, cc, large_scale_change.reviewers). Keys are
619 lowercased here, so callers don't need to normalize case themselves.
620 With `teams=None` (the default), `@org/team` references in those
621 fields pass through unexpanded — the offline/CLI mode, since this
622 library never calls out to GitHub/GitLab itself. That
623 partially-resolved config is only sanctioned for offline use;
624 anything that needs real reviewer usernames must pass a `teams`
625 mapping.
627 Pure function of the raw `self` and `other_configs` (it never reads its
628 own anchored output), so it is safe to call uncached.
629 """
631 if teams is not None:
632 teams = {key.lower(): members for key, members in teams.items()}
634 compiled_data = self._merged_data(config_path, other_configs)
636 # Expand aliases for any aliasable list fields. Team refs only expand
637 # in user-list fields — paths/code/labels keep a leading "@" literal.
638 for scope in compiled_data["scopes"]:
639 for field in [
640 "paths",
641 "code",
642 "authors",
643 "reviewers",
644 "alternates",
645 "cc",
646 "labels",
647 ]:
648 if field in scope:
649 scope[field] = _expand_aliases(
650 scope[field],
651 compiled_data["aliases"],
652 teams=teams,
653 expand_teams=field in USER_LIST_FIELDS,
654 )
656 # Apply compile-time "!" subtraction to roster fields.
657 # `_apply_negations` leaves a field with an unexpanded team ref
658 # (offline compile, i.e. teams=None) untouched.
659 for scope in compiled_data["scopes"]:
660 for field in ROSTER_FIELDS:
661 if field in scope:
662 scope[field] = _apply_negations(scope[field])
664 # The "*" wildcard is only meaningful in `reviewers` — everywhere
665 # else it's matched as a literal username and silently does
666 # nothing: in `alternates` the scope stays pending forever, in
667 # `authors` the scope never applies to any PR, in `cc` nobody is
668 # notified. Reject it at compile time (after alias expansion, so
669 # `$alias` indirection can't smuggle it in) rather than as a
670 # ScopeModel validator, because compiled configs stored inside
671 # old processing results must keep parsing.
672 # Messages must stay under ~130 chars: they become the git-host
673 # commit status description, which the adapters slice to 140.
674 for field, hint in (
675 (
676 "authors",
677 "remove it (a scope with no authors applies to any author)",
678 ),
679 (
680 "alternates",
681 "add it to reviewers instead (wildcard reviewers are never auto-requested)",
682 ),
683 ("cc", "remove it"),
684 ):
685 if "*" in scope.get(field, []):
686 raise ValueError(
687 f"Scope '{scope['name']}': \"*\" is not supported in "
688 f"{field} — {hint}"
689 )
691 _validate_review_counts(f"Scope '{scope['name']}'", scope)
693 if large_scale_change := compiled_data.get("large_scale_change"):
694 large_scale_change["reviewers"] = _expand_aliases(
695 large_scale_change["reviewers"],
696 compiled_data["aliases"],
697 teams=teams,
698 expand_teams=True,
699 )
700 large_scale_change["labels"] = _expand_aliases(
701 large_scale_change["labels"],
702 compiled_data["aliases"],
703 )
704 large_scale_change["reviewers"] = _apply_negations(
705 large_scale_change["reviewers"]
706 )
707 _validate_review_counts("large_scale_change", large_scale_change)
709 # Anchor each scope's paths at the directory tagged during flattening
710 # (after alias expansion, so any `$path-alias` is resolved first). The
711 # transient tag is popped so it never reaches the model.
712 for scope in compiled_data["scopes"]:
713 anchor_dir = scope.pop("_anchor_dir", "")
714 scope["paths"] = [_anchor_path(anchor_dir, p) for p in scope["paths"]]
716 # The compiled config is the self-contained effective config: extends
717 # are already merged in and aliases already expanded, so drop both. This
718 # keeps stored results lean and makes the compiled form standalone (it
719 # can never dangle on a missing extends target or re-expand differently).
720 compiled_data["extends"] = []
721 compiled_data["aliases"] = {}
723 return ConfigModel.from_data(
724 data=compiled_data,
725 path=config_path,
726 )
728 def _merged_data(
729 self,
730 config_path: Path,
731 other_configs: ConfigModels,
732 _in_progress: list[str] | None = None,
733 _seen: set[str] | None = None,
734 ) -> dict[str, Any]:
735 """
736 Flatten the `extends` chain into one merged, *unexpanded* config dict.
738 Parents are merged before this config (so a child can specialize), with
739 aliases unioned child-wins and the large-scale-change config taken from
740 the child if set else the first parent that defines one.
742 `_in_progress` is the current ancestor path, used to detect circular
743 extends. `_seen` is every config already merged into this flatten, used
744 to merge a shared ancestor only once (diamond dedup).
745 """
746 if _in_progress is None:
747 _in_progress = []
748 if _seen is None:
749 _seen = set()
751 config_path_str = str(config_path)
752 if config_path_str in _in_progress:
753 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
754 config_path_str
755 ]
756 raise ValueError(
757 f"Circular reference detected in extends: {' -> '.join(cycle)}"
758 )
759 _in_progress.append(config_path_str)
761 inherited_scopes: list[dict[str, Any]] = []
762 inherited_aliases: dict[str, list[str]] = {}
763 inherited_lsc: dict[str, Any] | None = None
765 for extend_path in self.extends:
766 resolved_path = _resolve_extends_path(config_path_str, extend_path)
767 if resolved_path not in other_configs:
768 raise ValueError(
769 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
770 )
771 if resolved_path in _seen:
772 # Already merged via another branch (diamond) — skip the dup.
773 continue
775 parent_data = other_configs[resolved_path]._merged_data(
776 Path(resolved_path), other_configs, _in_progress, _seen
777 )
778 inherited_scopes = inherited_scopes + parent_data["scopes"]
779 inherited_aliases = inherited_aliases | parent_data["aliases"]
780 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
782 merged = self.model_dump()
783 merged["scopes"] = inherited_scopes + merged["scopes"]
784 merged["aliases"] = inherited_aliases | merged["aliases"]
785 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
787 # Tag each scope with the directory its paths should anchor at. A scope's
788 # paths are relative to the config that owns it, so the first
789 # non-template config to consume a scope claims it: a non-template's own
790 # scopes (and any it inherits from a template) anchor at its directory,
791 # while a template defers to its consumer. `setdefault` means an
792 # already-tagged scope (from a non-template ancestor) keeps its anchor.
793 if not self.template:
794 base_dir = posixpath.dirname(config_path_str)
795 for scope in merged["scopes"]:
796 scope.setdefault("_anchor_dir", base_dir)
798 _seen.add(config_path_str)
799 _in_progress.pop()
801 return merged
803 @classmethod
804 def from_filesystem(cls, path: Path | str) -> ConfigModel:
805 with open(path, "rb") as f:
806 return cls.from_data(tomllib.load(f), path)
808 @classmethod
809 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
810 return cls.from_data(tomllib.loads(content), path)
812 @classmethod
813 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
814 return cls(**data)
817class _ConfigModelsBase(RootModel):
818 """Shared storage and accessors for a set of configs keyed by repo path."""
820 root: dict[str, ConfigModel]
822 @classmethod
823 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
824 """Build from a dict of already-constructed configs keyed by path."""
825 configs = cls(root={})
826 for path, config_model in models.items():
827 configs.root[str(Path(path))] = config_model
828 return configs
830 def get_config_models(self) -> dict[str, ConfigModel]:
831 return dict(self.root.items())
833 def __bool__(self) -> bool:
834 return bool(self.root)
836 def __getitem__(self, key: str) -> ConfigModel:
837 return self.root[key]
839 def __contains__(self, key: str) -> bool:
840 return key in self.root
842 def __len__(self) -> int:
843 return len(self.root)
846class ConfigModels(_ConfigModelsBase):
847 """Configs exactly as loaded from the repo — extends unresolved, aliases
848 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
850 @classmethod
851 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
852 """Load configs from a dict of parsed config data keyed by path."""
853 configs = cls(root={})
855 for path, config_data in data.items():
856 config = ConfigModel.from_data(config_data, Path(path))
857 configs.add_config(config, Path(path))
859 return configs
861 def add_config(self, config: ConfigModel, path: Path) -> None:
862 self.root[str(path)] = config
864 def team_refs(self) -> set[str]:
865 """Collect every team ref (lowercase, no leading `@`/`!`) that
866 `compiled(teams=...)` would actually try to expand: refs written
867 directly in a user-list field (scopes' `USER_LIST_FIELDS` and
868 `large_scale_change.reviewers`), plus any refs reachable from those
869 fields through `$alias`/`!$alias` chains.
871 Meant for callers that need to know which teams to fetch/sync before
872 calling `compiled(teams=...)`.
874 Implemented as an offline compile (`teams=None`): aliases expand but
875 team refs pass through unexpanded, so whatever refs remain in the
876 compiled user-list fields are — by construction — exactly the refs a
877 real compile will try to expand. A ref that only appears in a
878 non-user-list field (e.g. an npm-style `@vendor/pkg/**` in `paths`)
879 or inside an alias nothing references never survives into a compiled
880 user-list field, so it is never collected. Raises the same config
881 errors `compiled()` would (unknown alias, circular refs, ...), just
882 earlier.
883 """
884 # Cheap pre-check: a team ref can only enter a compile as a literal
885 # `@`/`!@` value in a user-list field or an alias value. Most repos
886 # have none, and skipping the compile keeps this near-free for them.
887 candidate_lists: list[list[str]] = []
888 for config in self.root.values():
889 candidate_lists.extend(config.aliases.values())
890 for scope in config.scopes:
891 for field in USER_LIST_FIELDS:
892 candidate_lists.append(getattr(scope, field))
893 if config.large_scale_change:
894 candidate_lists.append(config.large_scale_change.reviewers)
895 if not any(
896 _split_team_ref(value) is not None
897 for values in candidate_lists
898 for value in values
899 ):
900 return set()
902 refs: set[str] = set()
904 def collect_refs(values: list[str]) -> None:
905 for value in values:
906 if (split := _split_team_ref(value)) is not None:
907 refs.add(split[1].lower())
909 for config in self.compiled(teams=None).get_config_models().values():
910 if config.template:
911 continue
912 for scope in config.scopes:
913 for field in USER_LIST_FIELDS:
914 collect_refs(getattr(scope, field))
915 if config.large_scale_change:
916 collect_refs(config.large_scale_change.reviewers)
918 return refs
920 def compiled(
921 self, teams: dict[str, list[str]] | None = None
922 ) -> CompiledConfigModels:
923 """Resolve the whole set into its effective, PR-independent form.
925 Each non-template config is compiled once — extends merged, aliases
926 expanded, paths anchored. Templates are NOT compiled standalone: a
927 template scope may reference an alias the consuming config provides, and
928 its paths anchor at the consumer. They are carried through untouched
929 (folded into each consumer during that consumer's compile, and kept in
930 the set for display).
932 `teams` maps team refs (any case, without the leading `@`) to member
933 usernames; passed straight through to each config's `compiled_config`
934 (see there for case normalization and the `teams=None` vs provided
935 semantics).
937 The result is an immutable `CompiledConfigModels` — there is no way to
938 compile it again, so the non-idempotent path anchoring can never
939 double-apply.
940 """
941 effective: dict[str, ConfigModel] = {}
942 for path, config in self.root.items():
943 if config.template:
944 effective[path] = config
945 else:
946 effective[path] = config.compiled_config(
947 config_path=Path(path), other_configs=self, teams=teams
948 )
950 return CompiledConfigModels.from_config_models(effective)
953class CompiledConfigModels(_ConfigModelsBase):
954 """The effective configs used for matching: every non-template config is
955 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
957 def closest_config(self, file_path: Path) -> ConfigModel:
958 """Return the closest non-template config governing this file."""
959 for parent in file_path.parents:
960 parent_config_path = str(parent / CONFIG_FILENAME)
962 if parent_config_path in self.root:
963 config = self.root[parent_config_path]
965 if config.template:
966 # Skip templates
967 continue
969 return config
971 raise ValueError(f"No config found for {file_path}")
973 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
974 """The primary (repo-root) config's large-scale-change section, if any.
976 The primary was compiled by `compiled()`, so its reviewers/labels are
977 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
978 true` repo root is a misconfiguration (templates are meant to be
979 extended, not be the primary); it is passed through uncompiled, so its
980 LSC would read with aliases unexpanded.
981 """
982 if CONFIG_FILENAME in self.root:
983 if lsc := self.root[CONFIG_FILENAME].large_scale_change:
984 return lsc
986 return LargeScaleChangeModel()
988 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels:
989 """
990 Overlay PR-dependent scope gating: drop scopes that author rules disable
991 for this pull request.
993 This is the only PR-dependent step. The configs are already compiled, so
994 each config's scopes are self-contained and dropping one is a plain list
995 filter — no re-inheritance. Templates are passed through (they are never
996 matched directly; their scopes already live in each consumer).
997 """
998 effective: dict[str, ConfigModel] = {}
999 for config_path, config in self.root.items():
1000 if config.template:
1001 # Templates are never matched directly; pass them through.
1002 effective[config_path] = config
1003 continue
1005 kept_scopes = [
1006 scope
1007 for scope in config.scopes
1008 if scope.matches_author(author_username)
1009 ]
1010 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
1012 return CompiledConfigModels.from_config_models(effective)