Coverage for src/pullapprove/config.py: 94%
496 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-10-13 21:10 -0500
1from __future__ import annotations
3import os
4import posixpath
5import re
6import tomllib
7import warnings
9with warnings.catch_warnings():
10 warnings.simplefilter("ignore", DeprecationWarning)
11 import sre_parse
12from collections.abc import Generator, Iterable
13from enum import StrEnum
14from pathlib import Path
15from typing import Any, Self
17from pydantic import (
18 BaseModel,
19 ConfigDict,
20 Field,
21 RootModel,
22 field_validator,
23 model_validator,
24)
25from wcmatch import glob
27from .checklists import Checklist
30def _resolve_config_filename_prefix() -> str:
31 """The name every config file starts with, for this process.
33 Defaults to CODEREVIEW. An instance can rename it with
34 PULLAPPROVE_CONFIG_PREFIX, which renames the config file itself
35 (REVIEW -> REVIEW.toml, REVIEW.template.toml). Because discovery matches
36 on this prefix, an instance only ever sees files named for its own prefix
37 -- so two instances can watch the same repo without seeing each other's
38 configs, which is how unreleased config features get tried out on a repo
39 that production also watches.
41 Read once at import: this is a per-process constant, not runtime state.
43 Two operational constraints this doesn't (and can't) enforce:
45 Prefixes must not be prefixes of one another. Matching is `startswith`
46 (see is_config_filename), and a suffix after the prefix is legitimate --
47 CODEREVIEW.template.toml and CODEREVIEW-BASE.toml are both real config
48 names -- so a REVIEW instance would also pick up REVIEW_DEV.toml. An
49 instance only knows its own prefix, so pick names that don't overlap
50 (REVIEW and DEV_REVIEW, not REVIEW and REVIEW_DEV).
52 Changing the prefix on a live instance means clearing its cached configs.
53 Config rows are keyed by (repo, sha) with no record of which prefix
54 discovered them, so a sha already processed under the old prefix keeps
55 serving from cache and the new prefix's files are never fetched.
56 """
57 # `or` rather than a get() default: an explicitly empty value
58 # (`PULLAPPROVE_CONFIG_PREFIX=` in a .env or compose file -- a normal way to
59 # write "unset") is not a missing key, and would name the config ".toml".
60 prefix = os.environ.get("PULLAPPROVE_CONFIG_PREFIX", "").strip() or "CODEREVIEW"
62 # A bad prefix would silently match nothing (i.e. every repo looks
63 # unconfigured), so refuse to start instead.
64 if prefix.endswith(".toml"):
65 raise ValueError(
66 f"PULLAPPROVE_CONFIG_PREFIX should be a name without an extension, not {prefix!r}. "
67 f"The config file is named after it (e.g. {prefix[: -len('.toml')]!r} -> {prefix!r})."
68 )
69 if "\\" in prefix or prefix != posixpath.basename(prefix):
70 raise ValueError(
71 f"PULLAPPROVE_CONFIG_PREFIX should be a filename prefix, not a path: {prefix!r}"
72 )
74 return prefix
77CONFIG_FILENAME_PREFIX = _resolve_config_filename_prefix()
78CONFIG_FILENAME = f"{CONFIG_FILENAME_PREFIX}.toml"
81def is_config_filename(basename: str) -> bool:
82 """Whether a filename is a config file (CODEREVIEW.toml,
83 CODEREVIEW.template.toml).
85 The prefix half is what isolates instances: an instance running a renamed
86 prefix (see PULLAPPROVE_CONFIG_PREFIX) never even discovers another
87 instance's configs. The extension half keeps a neighbor like CODEREVIEW.md
88 from being treated as one.
89 """
90 return basename.startswith(CONFIG_FILENAME_PREFIX) and basename.endswith(".toml")
93def is_config_path(path: str) -> bool:
94 """Whether a repo-relative path names a config file."""
95 return is_config_filename(posixpath.basename(path))
98_REPEAT_OPS = {sre_parse.MAX_REPEAT, sre_parse.MIN_REPEAT}
101def _has_nested_quantifiers(data: Any) -> bool:
102 """Detect patterns like (a+)+ that cause catastrophic backtracking."""
103 for op, av in data:
104 if op in _REPEAT_OPS:
105 if _contains_quantifier(av[2]):
106 return True
107 elif op == sre_parse.SUBPATTERN:
108 if _has_nested_quantifiers(av[-1]):
109 return True
110 elif op == sre_parse.BRANCH: # noqa: SIM102
111 if any(_has_nested_quantifiers(branch) for branch in av[1]):
112 return True
113 return False
116def _contains_quantifier(data: Any) -> bool:
117 for op, av in data:
118 if op in _REPEAT_OPS:
119 return True
120 elif op == sre_parse.SUBPATTERN:
121 if _contains_quantifier(av[-1]):
122 return True
123 elif op == sre_parse.BRANCH: # noqa: SIM102
124 if any(_contains_quantifier(branch) for branch in av[1]):
125 return True
126 return False
129_TEAM_REF_SEGMENT = r"[a-zA-Z0-9][a-zA-Z0-9\-_.]*"
130_TEAM_REF_RE = re.compile(rf"^{_TEAM_REF_SEGMENT}(/{_TEAM_REF_SEGMENT})+$")
132# Fields that hold reviewer identities (plain usernames, `$aliases`, and
133# `@team` refs) rather than paths/code/labels. Team refs only expand here.
134USER_LIST_FIELDS = ("authors", "reviewers", "alternates", "cc")
136# Roster fields where "!" performs compile-time subtraction (remove from the
137# resolved list) rather than surviving as a match-time predicate. `authors`
138# is deliberately excluded — its "!" entries are consumed by `matches_author`.
139ROSTER_FIELDS = ("reviewers", "alternates", "cc")
142def _split_team_ref(value: str) -> tuple[str, str] | None:
143 """Split a value into `(prefix, ref)` if it has team-reference shape
144 (`@org/team` or `!@org/team`), `prefix` being `"!"` or `""`. Returns
145 `None` for anything else, so callers fall back to their own handling of
146 plain values.
147 """
148 if value.startswith("!@"):
149 return "!", value[2:]
150 if value.startswith("@"):
151 return "", value[1:]
152 return None
155def is_unexpanded_ref(value: str) -> bool:
156 """True if the value is a `$alias` or `@team` reference (optionally
157 negated with a leading `!`) that has not been expanded — as opposed to a
158 plain username. Only an offline compile (`teams=None`) leaves such values
159 in user-list fields.
160 """
161 return value.removeprefix("!").startswith(("$", "@"))
164def _validate_team_refs(values: list[str]) -> list[str]:
165 """Team references (`@org/team`, `!@org/team`) need at least two
166 slash-separated segments. This only checks shape — membership is resolved
167 later, at compile time, against the caller-provided `teams` mapping.
169 Any other value containing "@" is rejected too — that shape is reserved
170 for team references (and, in future, email-style identifiers), so it
171 can't be confused with a plain username.
172 """
173 for value in values:
174 split = _split_team_ref(value)
175 if split is None:
176 if "@" in value:
177 raise ValueError(
178 f"Invalid value '{value}': email addresses are not "
179 "supported here — use the platform username"
180 )
181 continue
183 _prefix, ref = split
184 if not _TEAM_REF_RE.match(ref):
185 raise ValueError(
186 f"Invalid team reference '{value}': team references need the "
187 "'org/team' form"
188 )
189 return values
192def _expand_team_ref(
193 ref: str, teams: dict[str, list[str]] | None, prefix: str
194) -> list[str]:
195 """Expand a single team reference (without its `@`/`!@`) to member usernames.
197 `teams=None` is the offline/CLI mode: the library never calls out to
198 GitHub/GitLab itself, so with no mapping provided the reference passes
199 through unexpanded rather than erroring. With a `teams` mapping (even an
200 empty one), an unresolvable ref is a loud config error, matching how
201 unknown `$aliases` are handled.
202 """
203 if teams is None:
204 return [f"{prefix}@{ref}"]
206 members = teams.get(ref.lower())
207 if members is None:
208 raise ValueError(f"Unknown team: {prefix}@{ref}")
210 return [f"{prefix}{member}" for member in members]
213def _expand_aliases(
214 values: list[str],
215 aliases: dict[str, list[str]],
216 teams: dict[str, list[str]] | None = None,
217 expand_teams: bool = False,
218 _seen: set[str] | None = None,
219 _path: list[str] | None = None,
220) -> list[str]:
221 """Replace alias references in a list with their mapped values recursively.
223 Team references (`@org/team`) are only expanded when `expand_teams` is
224 True — user-list fields (reviewers, alternates, authors, cc). Elsewhere
225 (paths, code, labels) a leading `@` is a literal string, e.g. npm-style
226 scoped path patterns like `@vendor/pkg/**`. Teams are always leaf nodes —
227 their values are plain usernames, never other refs — so expanding one
228 never recurses further.
229 """
230 if _seen is None:
231 _seen = set()
232 if _path is None:
233 _path = []
235 expanded: list[str] = []
236 for value in values:
237 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
238 if value.startswith("!$"):
239 prefix = "!"
240 alias_ref = value[2:]
241 elif value.startswith("$"):
242 prefix = ""
243 alias_ref = value[1:]
244 elif expand_teams and (team_ref := _split_team_ref(value)) is not None:
245 team_prefix, ref = team_ref
246 expanded.extend(_expand_team_ref(ref=ref, teams=teams, prefix=team_prefix))
247 continue
248 else:
249 expanded.append(value)
250 continue
252 if alias_ref in _seen:
253 # Cycle detected, raise an error with the cycle path
254 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
255 raise ValueError(
256 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
257 )
258 if alias_ref in aliases:
259 _seen.add(alias_ref)
260 _path.append(alias_ref)
261 # Recursively expand the alias values
262 nested_expanded = _expand_aliases(
263 aliases[alias_ref],
264 aliases=aliases,
265 teams=teams,
266 expand_teams=expand_teams,
267 _seen=_seen,
268 _path=_path,
269 )
270 if prefix:
271 expanded.extend(prefix + v for v in nested_expanded)
272 else:
273 expanded.extend(nested_expanded)
274 _path.pop()
275 _seen.remove(alias_ref)
276 else:
277 # Unknown alias — surface it loudly instead of silently dropping the
278 # reference (a typo'd alias would otherwise vanish reviewers/paths).
279 raise ValueError(f"Unknown alias: {prefix}${alias_ref}")
281 # Remove duplicates while preserving order
282 return list(dict.fromkeys(expanded))
285def _apply_negations(values: list[str]) -> list[str]:
286 """Compile-time subtraction for roster fields (`ROSTER_FIELDS`): a "!name"
287 entry removes "name" from the resolved list instead of surviving as a
288 match-time rule (contrast `authors`, handled by `matches_author`).
290 A list still holding an unexpanded `@team` ref (offline compile, i.e.
291 teams=None) is returned untouched — it isn't fully resolved yet, so
292 subtraction can't run, and the partially-resolved config must round-trip
293 unchanged. `$aliases` must already be expanded by the caller.
295 The wildcard+negation error fires BEFORE that early return: it's a check
296 on the written form (`"*"` alongside any `"!"` entry, team ref or not),
297 so an offline compile must reject it the same way the server will — a
298 `pullapprove check` that passes locally can't then error in production.
299 """
300 if "*" in values and any(v.startswith("!") for v in values):
301 raise ValueError(
302 'Negation cannot be combined with "*" in reviewers/alternates/cc '
303 "(wildcard exclusion is not supported)"
304 )
306 if any(_split_team_ref(v) is not None for v in values):
307 return values
309 negations = {v[1:].lower() for v in values if v.startswith("!")}
310 return [v for v in values if not v.startswith("!") and v.lower() not in negations]
313def _validate_review_counts(label: str, data: dict[str, Any]) -> None:
314 """Compile-time rejection of negative require/request/author_value.
315 Shared by scopes and large_scale_change (which has no
316 request/author_value — the `.get` defaults pass trivially there).
318 These are compile-time checks (not field validators) because negative
319 values were previously accepted, so compiled configs stored inside old
320 processing results must keep parsing (where they keep their old
321 behavior: a negative require always passed, a negative request
322 requested nobody).
323 """
324 for field in ("require", "request", "author_value"):
325 if data.get(field, 0) < 0:
326 raise ValueError(f"{label}: {field} cannot be negative")
329def _resolve_extends_path(extending_path: str, extends_ref: str) -> str:
330 """Resolve an `extends` reference to a canonical repo-relative config key.
332 - `/x` is repo-root-relative.
333 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending
334 file's directory.
336 Raises if the reference escapes above the repo root.
337 """
338 if extends_ref.startswith("/"):
339 resolved = posixpath.normpath(extends_ref.lstrip("/"))
340 else:
341 base_dir = posixpath.dirname(extending_path)
342 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref))
344 if resolved == ".." or resolved.startswith("../"):
345 raise ValueError(
346 f"Invalid extends path: '{extends_ref}' points above the repo root"
347 )
349 return resolved
352def matches_path_patterns(*, path: Path, patterns: list[str]) -> bool:
353 """Whether `path` matches any of the config's path globs.
355 The one definition of the config's glob semantics.
356 """
357 # TODO paths shouldn't start with /
358 return glob.globmatch(
359 path,
360 patterns,
361 flags=glob.GLOBSTAR | glob.BRACE | glob.NEGATE | glob.IGNORECASE | glob.DOTGLOB,
362 )
365def _anchor_path(base_dir: str, pattern: str) -> str:
366 """Anchor a scope path glob at `base_dir` (the owning config's directory).
368 Scope paths are written relative to the config they live in. A leading `/`
369 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is
370 preserved. With an empty `base_dir` (root config) the pattern is unchanged.
371 """
372 negate = pattern.startswith("!")
373 if negate:
374 pattern = pattern[1:]
376 if pattern.startswith("/"):
377 anchored = pattern.lstrip("/")
378 elif base_dir:
379 anchored = f"{base_dir}/{pattern}"
380 else:
381 anchored = pattern
383 return f"!{anchored}" if negate else anchored
386class OwnershipChoices(StrEnum):
387 EMPTY = ""
388 APPEND = "append"
389 GLOBAL = "global"
392# A GitHub App has two names: the login it posts as (`name[bot]`) and the slug
393# that owns its check runs (`name`). The suffix is how the engine tells a bot
394# from a person -- a `[bot]` account never counts as a human and never sits in
395# a roster, no declaration needed.
396BOT_LOGIN_SUFFIX = "[bot]"
399def is_bot_login(username: str) -> bool:
400 """The engine rule, in one place: a `[bot]` account is never a person."""
401 return username.lower().endswith(BOT_LOGIN_SUFFIX)
404def _split_check_ref(ref: str) -> tuple[str, str]:
405 """An `unless` ref as (producer slug lowercased, check name).
407 Split on the FIRST `/`: App slugs cannot contain one, check names can.
408 Both parts are stripped -- the validator checks THIS function's output, so
409 a ref that validates is exactly a ref that matches at runtime.
410 """
411 producer, _, name = ref.partition("/")
412 return producer.strip().lower(), name.strip()
415def _reject_bot_reviewers(values: list[str], *, where: str) -> None:
416 """Reject any `[bot]` account listed in a roster surface (a scope's
417 reviewers/alternates/cc, or large_scale_change.reviewers).
419 A bot is never a person, so it can never sit in a roster: bots that open
420 pull requests are routed with `authors`, and bots that attest are
421 referenced from `unless`. Checked after alias expansion so `$alias`
422 indirection can't smuggle one in. `where` names the surface; the message
423 becomes the git-host commit status, so it stays under ~130 chars.
424 """
425 for entry in values:
426 if is_bot_login(entry):
427 raise ValueError(
428 f"{where}: '{entry}' is a bot and cannot be listed as a reviewer"
429 )
432def _first_case_insensitive_duplicate(values: Iterable[str]) -> str | None:
433 """The first value whose lowercased form was already seen, else None."""
434 seen: set[str] = set()
435 for value in values:
436 if value.lower() in seen:
437 return value
438 seen.add(value.lower())
439 return None
442class ScopeModel(BaseModel):
443 model_config = ConfigDict(extra="forbid")
445 # Required fields
446 name: str = Field(min_length=1)
447 paths: list[str] = Field(min_length=1)
449 # Optional fields
451 # Expanded version of lines could be dict
452 # with fnmatch, regex, exclude patterns, etc?
453 code: list[str] = []
455 # This only filtering field that can't be used with raw diff/files...
456 # If we get into that, the others are:
457 # - labels
458 # - ref
459 # - statuses
460 # - dates
461 # - body
462 # - title
463 # - other scopes
464 # (this is how I ended up with expressions...
465 # I'm not trying to build a general purpose workflow tool,
466 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
467 authors: list[str] = []
469 # (defaults should be the "empty" values)
470 description: str = ""
471 reviewers: list[str] = []
472 alternates: list[str] = []
473 cc: list[str] = []
475 # Review scoring
476 # Negative values are rejected at compile time (_validate_review_counts)
477 require: int = 0
478 author_value: int = 0
480 # The checks that can attest this scope's review requirement away. Each ref
481 # is "producer/check-name" -- the slug of the App that produces the check,
482 # then the check run's name, split on the first "/". When every listed
483 # check has completed successfully on the current head commit, the scope is
484 # waived: its requirement drops to zero and the evidence is stored on the
485 # result. Failed, skipped, pending, missing -- the requirement stands
486 # unchanged; passing is the only state that subtracts.
487 #
488 # Deliberately placed next to the `require` it undermines, so reading a
489 # scope always reveals whether its human review is removable. A list is
490 # implicitly "all must pass" -- OR, thresholds, and 2-of-3 belong in the
491 # producer, which can post one combined check.
492 unless: list[str] = []
494 # How scopes are combined
495 ownership: OwnershipChoices = OwnershipChoices.EMPTY
497 # Actionable items
498 request: int = 0
499 labels: list[str] = []
500 instructions: str = ""
502 # Approval checklist
503 checklist: Checklist | None = None
505 @field_validator("name", mode="after")
506 @classmethod
507 def validate_name(cls, name: str) -> str:
508 if "," in name:
509 raise ValueError("Scope name cannot contain commas")
510 return name
512 @field_validator(*USER_LIST_FIELDS, mode="after")
513 @classmethod
514 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
515 return _validate_team_refs(values)
517 @field_validator("unless", mode="after")
518 @classmethod
519 def validate_unless_refs(cls, values: list[str]) -> list[str]:
520 """A check name alone is worthless -- any workflow with `checks: write`
521 can post any name -- so a bare name is a parse error, not a default.
522 Messages stay under ~130 chars: they become the git-host commit status.
523 """
524 for ref in values:
525 producer, name = _split_check_ref(ref)
526 if "/" not in ref or not producer or not name:
527 raise ValueError(
528 f"unless: '{ref}' must be 'producer/check-name' -- the App "
529 "slug that produces the check, then the check's name"
530 )
531 # To most users PullApprove *is* "the check" on their PRs.
532 # PullApprove never creates these checks, it only reads them -- and
533 # a config waiting on our own status would deadlock politely.
534 if producer == "pullapprove":
535 raise ValueError(
536 f"unless: '{ref}' references PullApprove itself -- "
537 "PullApprove never creates checks, it only reads them"
538 )
539 return values
541 @field_validator("code", mode="after")
542 @classmethod
543 def validate_code_patterns(cls, code: list[str]) -> list[str]:
544 for pattern in code:
545 try:
546 parsed = sre_parse.parse(pattern)
547 except re.error as e:
548 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
549 if _has_nested_quantifiers(parsed):
550 raise ValueError(
551 f"Regex pattern '{pattern}' contains nested quantifiers, "
552 "which can cause catastrophic backtracking."
553 )
554 return code
556 @model_validator(mode="after")
557 def validate_reviewers_for_require(self) -> ScopeModel:
558 all_reviewers = self.reviewers + self.alternates
560 # Skip if wildcard - anyone can review
561 if "*" in all_reviewers:
562 return self
564 # Skip if aliases or team refs (possibly negated with "!") are not yet
565 # expanded. Re-validation after compilation only re-checks refs that
566 # actually resolve — an offline compile (teams=None) leaves refs
567 # unexpanded, so this same skip fires again there too.
568 if any(is_unexpanded_ref(r) for r in all_reviewers):
569 return self
571 if len(all_reviewers) < self.require:
572 raise ValueError(
573 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
574 )
575 return self
577 def counts_toward(self, username: str) -> bool:
578 """Whether this person's review counts for this scope.
580 Its named `reviewers` and its `alternates`, plus everybody when
581 `reviewers` holds the wildcard. A wildcard in `alternates` does not
582 count everybody: the wildcard describes the scope's `reviewers`, not
583 its backups.
585 Case-insensitive, like every other username comparison here.
586 """
587 if "*" in self.reviewers:
588 return True
589 username_lower = username.lower()
590 return any(
591 username_lower == r.lower() for r in self.reviewers + self.alternates
592 )
594 def author_points(self, author_username: str) -> int:
595 """Author points only count if the author is explicitly listed as a
596 reviewer (a wildcard is not converted to usernames)."""
597 if author_username.lower() in {r.lower() for r in self.reviewers}:
598 return self.author_value
599 return 0
601 def unless_refs(self) -> list[tuple[str, str]]:
602 """Each `unless` entry as (producer slug lowercased, check name)."""
603 return [_split_check_ref(ref) for ref in self.unless]
605 def unsolvable_reason(self, author_username: str) -> str | None:
606 """
607 Explain why this scope can never pass for a PR authored by this user,
608 or None if it can.
610 Messages must stay under ~130 chars: they flow into the git-host
611 commit status description, which the adapters slice to 140.
612 """
613 if "*" in self.reviewers:
614 # Anyone can review, so any require is satisfiable
615 return None
617 # Count eligible reviewers (excluding author who can't self-approve)
618 eligible_reviewers = {r.lower() for r in self.reviewers + self.alternates} - {
619 author_username.lower()
620 }
621 max_possible_points = len(eligible_reviewers) + self.author_points(
622 author_username
623 )
625 if self.require > 0 and max_possible_points < self.require:
626 if not eligible_reviewers:
627 return (
628 "PR author is the only reviewer/alternate and cannot self-approve"
629 )
630 return f"require={self.require} but only {max_possible_points} possible approvals (excluding author)"
632 return None
634 def ownership_marker(self) -> str:
635 """The glyph a non-default ownership puts in front of the scope name.
637 Split out from printed_name so a renderer that styles the marker apart
638 from the name doesn't have to know which glyph goes with which mode.
639 """
640 match self.ownership:
641 case OwnershipChoices.APPEND:
642 return "+"
643 case OwnershipChoices.GLOBAL:
644 return "*"
646 return ""
648 def printed_name(self) -> str:
649 return self.ownership_marker() + self.name
651 def __eq__(self, other: object) -> bool:
652 if not isinstance(other, ScopeModel):
653 return NotImplemented
654 return self.name == other.name
656 def matches_path(self, path: Path) -> bool:
657 return matches_path_patterns(path=path, patterns=self.paths)
659 def matches_code(self, code: str) -> Generator[dict[str, int]]:
660 patterns = getattr(self, "_code_regex_patterns", [])
661 if not patterns:
662 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
663 self._code_regex_patterns = patterns
665 for pattern in patterns:
666 for match in pattern.finditer(code):
667 start_index = match.start()
668 end_index = match.end()
670 start_line = code.count("\n", 0, start_index) + 1
671 start_col = start_index - code.rfind("\n", 0, start_index)
673 end_line = code.count("\n", 0, end_index) + 1
674 end_col = end_index - code.rfind("\n", 0, end_index)
676 yield {
677 "start_line": start_line,
678 "start_col": start_col,
679 "end_line": end_line,
680 "end_col": end_col,
681 }
683 def matches_author(self, author_username: str) -> bool:
684 if not self.authors:
685 # No authors specified, so assume it matches
686 return True
688 author_username_lower = author_username.lower()
690 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
691 authors = [a.lower() for a in self.authors if not a.startswith("!")]
693 if author_username_lower in negated_authors:
694 # If the author is in the negated list, return False
695 return False
697 if not authors:
698 # Negation-only: everyone not negated matches
699 return True
701 return author_username_lower in authors
704class LargeScaleChangeModel(BaseModel):
705 model_config = ConfigDict(extra="forbid")
707 # Note, an LSC only applies to diffs, not raw files,
708 # because we have to know what *changed*.
710 # Pretty similar to a scope, but more manual.
711 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
712 # Negative values are rejected at compile time (_validate_review_counts)
713 require: int = 1
714 reviewers: list[str] = [] # Field(min_length=1)
715 # min_paths: int = 300
716 # min_lines: int = 3000
717 labels: list[str] = []
718 # really need author value too...?
720 @field_validator("reviewers", mode="after")
721 @classmethod
722 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
723 return _validate_team_refs(values)
725 def unsolvable_reason(self, author_username: str) -> str | None:
726 """
727 Explain why this LSC can never pass for a PR authored by this user,
728 or None if it can.
730 Messages must stay under ~130 chars: they flow into the git-host
731 commit status description, which the adapters slice to 140.
732 """
733 if "*" in self.reviewers:
734 # Anyone can review, so any require is satisfiable
735 return None
737 if not self.reviewers:
738 # An empty roster is "configuration required", which
739 # process_large_scale_change reports on its own
740 return None
742 # Count eligible reviewers (excluding author who can't self-approve)
743 eligible_reviewers = {r.lower() for r in self.reviewers} - {
744 author_username.lower()
745 }
746 if self.require > 0 and len(eligible_reviewers) < self.require:
747 if not eligible_reviewers:
748 return "the PR author is the only reviewer and cannot self-approve"
749 return f"require={self.require} but only {len(eligible_reviewers)} possible approvals (excluding author)"
751 return None
754class ConfigModel(BaseModel):
755 model_config = ConfigDict(extra="forbid")
757 # Nothing is technically required
758 extends: list[str] = []
759 template: bool = False
760 aliases: dict[str, list[str]] = {}
761 large_scale_change: LargeScaleChangeModel | None = None
762 scopes: list[ScopeModel] = []
764 @field_validator("scopes", mode="after")
765 @classmethod
766 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
767 if dup := _first_case_insensitive_duplicate(scope.name for scope in scopes):
768 raise ValueError(f"Duplicate scope name: {dup}")
769 return scopes
771 @field_validator("extends", mode="before")
772 @classmethod
773 def validate_extends(cls, extends: list[str]) -> list[str]:
774 for i, path in enumerate(extends):
775 basename = Path(path).name
776 if not basename.startswith(CONFIG_FILENAME_PREFIX):
777 raise ValueError(
778 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
779 )
780 return extends
782 def compiled_config(
783 self,
784 config_path: Path,
785 other_configs: ConfigModels,
786 teams: dict[str, list[str]] | None = None,
787 ) -> ConfigModel:
788 """
789 Resolve `extends` and replace aliases, returning the effective config.
791 Two phases: flatten the whole extends chain into one merged (raw,
792 unexpanded) config, then expand aliases once. Expanding after the full
793 merge is what makes transitive inheritance and cross-chain alias
794 scoping work — an alias defined anywhere in the chain resolves anywhere.
796 `teams` maps team refs (any case, without the leading `@`) to member
797 usernames, and is only consulted for user-list fields (reviewers,
798 alternates, authors, cc, large_scale_change.reviewers). Keys are
799 lowercased here, so callers don't need to normalize case themselves.
800 With `teams=None` (the default), `@org/team` references in those
801 fields pass through unexpanded — the offline/CLI mode, since this
802 library never calls out to GitHub/GitLab itself. That
803 partially-resolved config is only sanctioned for offline use;
804 anything that needs real reviewer usernames must pass a `teams`
805 mapping.
807 Pure function of the raw `self` and `other_configs` (it never reads its
808 own anchored output), so it is safe to call uncached.
809 """
811 if teams is not None:
812 teams = {key.lower(): members for key, members in teams.items()}
814 compiled_data = self._merged_data(config_path, other_configs)
816 # Expand aliases for any aliasable list fields. Team refs only expand
817 # in user-list fields — paths/code/labels keep a leading "@" literal.
818 for scope in compiled_data["scopes"]:
819 for field in [
820 "paths",
821 "code",
822 "authors",
823 "reviewers",
824 "alternates",
825 "cc",
826 "labels",
827 ]:
828 if field in scope:
829 scope[field] = _expand_aliases(
830 scope[field],
831 compiled_data["aliases"],
832 teams=teams,
833 expand_teams=field in USER_LIST_FIELDS,
834 )
836 # Apply compile-time "!" subtraction to roster fields.
837 # `_apply_negations` leaves a field with an unexpanded team ref
838 # (offline compile, i.e. teams=None) untouched.
839 for scope in compiled_data["scopes"]:
840 for field in ROSTER_FIELDS:
841 if field in scope:
842 scope[field] = _apply_negations(scope[field])
843 _reject_bot_reviewers(
844 scope[field],
845 where=f"Scope '{scope['name']}' {field}",
846 )
848 # The "*" wildcard is only meaningful in `reviewers` — everywhere
849 # else it's matched as a literal username and silently does
850 # nothing: in `alternates` the scope stays pending forever, in
851 # `authors` the scope never applies to any PR, in `cc` nobody is
852 # notified. Reject it at compile time (after alias expansion, so
853 # `$alias` indirection can't smuggle it in) rather than as a
854 # ScopeModel validator, because compiled configs stored inside
855 # old processing results must keep parsing.
856 # Messages must stay under ~130 chars: they become the git-host
857 # commit status description, which the adapters slice to 140.
858 for field, hint in (
859 (
860 "authors",
861 "remove it (a scope with no authors applies to any author)",
862 ),
863 (
864 "alternates",
865 "add it to reviewers instead (wildcard reviewers are never auto-requested)",
866 ),
867 ("cc", "remove it"),
868 ):
869 if "*" in scope.get(field, []):
870 raise ValueError(
871 f"Scope '{scope['name']}': \"*\" is not supported in "
872 f"{field} — {hint}"
873 )
875 _validate_review_counts(f"Scope '{scope['name']}'", scope)
877 if large_scale_change := compiled_data.get("large_scale_change"):
878 large_scale_change["reviewers"] = _expand_aliases(
879 large_scale_change["reviewers"],
880 compiled_data["aliases"],
881 teams=teams,
882 expand_teams=True,
883 )
884 large_scale_change["labels"] = _expand_aliases(
885 large_scale_change["labels"],
886 compiled_data["aliases"],
887 )
888 large_scale_change["reviewers"] = _apply_negations(
889 large_scale_change["reviewers"]
890 )
891 _reject_bot_reviewers(
892 large_scale_change["reviewers"],
893 where="large_scale_change reviewers",
894 )
895 _validate_review_counts("large_scale_change", large_scale_change)
897 # Anchor each scope's paths at the directory tagged during flattening
898 # (after alias expansion, so any `$path-alias` is resolved first). The
899 # transient tag is popped so it never reaches the model.
900 # Paths are relative to the config that owns the scope, `/` for
901 # repo-root-absolute.
902 for entry in compiled_data["scopes"]:
903 anchor_dir = entry.pop("_anchor_dir", "")
904 entry["paths"] = [_anchor_path(anchor_dir, p) for p in entry["paths"]]
906 # The compiled config is the self-contained effective config: extends
907 # are already merged in and aliases already expanded, so drop both. This
908 # keeps stored results lean and makes the compiled form standalone (it
909 # can never dangle on a missing extends target or re-expand differently).
910 compiled_data["extends"] = []
911 compiled_data["aliases"] = {}
913 return ConfigModel.from_data(
914 data=compiled_data,
915 path=config_path,
916 )
918 def _merged_data(
919 self,
920 config_path: Path,
921 other_configs: ConfigModels,
922 _in_progress: list[str] | None = None,
923 _seen: set[str] | None = None,
924 ) -> dict[str, Any]:
925 """
926 Flatten the `extends` chain into one merged, *unexpanded* config dict.
928 Parents are merged before this config (so a child can specialize), with
929 aliases unioned child-wins and the large-scale-change config taken from
930 the child if set else the first parent that defines one.
932 `_in_progress` is the current ancestor path, used to detect circular
933 extends. `_seen` is every config already merged into this flatten, used
934 to merge a shared ancestor only once (diamond dedup).
935 """
936 if _in_progress is None:
937 _in_progress = []
938 if _seen is None:
939 _seen = set()
941 config_path_str = str(config_path)
942 if config_path_str in _in_progress:
943 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
944 config_path_str
945 ]
946 raise ValueError(
947 f"Circular reference detected in extends: {' -> '.join(cycle)}"
948 )
949 _in_progress.append(config_path_str)
951 inherited_scopes: list[dict[str, Any]] = []
952 inherited_aliases: dict[str, list[str]] = {}
953 inherited_lsc: dict[str, Any] | None = None
955 for extend_path in self.extends:
956 resolved_path = _resolve_extends_path(config_path_str, extend_path)
957 if resolved_path not in other_configs:
958 raise ValueError(
959 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
960 )
961 if resolved_path in _seen:
962 # Already merged via another branch (diamond) — skip the dup.
963 continue
965 parent_data = other_configs[resolved_path]._merged_data(
966 Path(resolved_path), other_configs, _in_progress, _seen
967 )
968 inherited_scopes = inherited_scopes + parent_data["scopes"]
969 inherited_aliases = inherited_aliases | parent_data["aliases"]
970 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
972 merged = self.model_dump()
973 merged["scopes"] = inherited_scopes + merged["scopes"]
974 merged["aliases"] = inherited_aliases | merged["aliases"]
975 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
977 # Tag each scope with the directory its paths should anchor at. A scope's
978 # paths are relative to the config that owns it, so the first
979 # non-template config to consume a scope claims it: a non-template's own
980 # scopes (and any it inherits from a template) anchor at its directory,
981 # while a template defers to its consumer. `setdefault` means an
982 # already-tagged scope (from a non-template ancestor) keeps its anchor.
983 if not self.template:
984 base_dir = posixpath.dirname(config_path_str)
985 for entry in merged["scopes"]:
986 entry.setdefault("_anchor_dir", base_dir)
988 _seen.add(config_path_str)
989 _in_progress.pop()
991 return merged
993 @classmethod
994 def from_filesystem(cls, path: Path | str) -> ConfigModel:
995 with open(path, "rb") as f:
996 return cls.from_data(tomllib.load(f), path)
998 @classmethod
999 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
1000 return cls.from_data(tomllib.loads(content), path)
1002 @classmethod
1003 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
1004 return cls(**data)
1007class _ConfigModelsBase(RootModel):
1008 """Shared storage and accessors for a set of configs keyed by repo path."""
1010 root: dict[str, ConfigModel]
1012 @classmethod
1013 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
1014 """Build from a dict of already-constructed configs keyed by path."""
1015 configs = cls(root={})
1016 for path, config_model in models.items():
1017 configs.root[str(Path(path))] = config_model
1018 return configs
1020 def get_config_models(self) -> dict[str, ConfigModel]:
1021 return dict(self.root.items())
1023 def __bool__(self) -> bool:
1024 return bool(self.root)
1026 def __getitem__(self, key: str) -> ConfigModel:
1027 return self.root[key]
1029 def __contains__(self, key: str) -> bool:
1030 return key in self.root
1032 def __len__(self) -> int:
1033 return len(self.root)
1035 def closest_config_path(self, file_path: Path) -> str | None:
1036 """The path of the closest non-template config governing this file,
1037 or None when nothing governs it."""
1038 for parent in file_path.parents:
1039 parent_config_path = str(parent / CONFIG_FILENAME)
1040 config = self.root.get(parent_config_path)
1041 if config is not None and not config.template:
1042 return parent_config_path
1043 return None
1046class ConfigModels(_ConfigModelsBase):
1047 """Configs exactly as loaded from the repo — extends unresolved, aliases
1048 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
1050 def declared_check_names(self) -> set[str]:
1051 """Every check-run name an effective (non-template) scope's `unless`
1052 references, across the whole config set.
1054 Collected from an offline compile (`teams=None`, the `team_refs`
1055 precedent) so a template's refs count only where a consumer actually
1056 inherits them -- an unconsumed template must not make every pull
1057 request fetch (or error on) checks that nothing evaluates. `unless`
1058 refs are literal strings (aliases and teams never expand inside
1059 them), so the offline compile is exact. Empty for the common case (no
1060 `unless` anywhere), which is what lets the processor skip fetching
1061 check runs entirely.
1062 """
1063 return {
1064 name
1065 for config in self.compiled().root.values()
1066 if not config.template
1067 for scope in config.scopes
1068 for _, name in scope.unless_refs()
1069 }
1071 @classmethod
1072 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
1073 """Load configs from a dict of parsed config data keyed by path."""
1074 configs = cls(root={})
1076 for path, config_data in data.items():
1077 config = ConfigModel.from_data(config_data, Path(path))
1078 configs.add_config(config, Path(path))
1080 return configs
1082 @classmethod
1083 def from_contents(cls, contents: dict[str, str]) -> ConfigModels:
1084 """Load configs from a dict of raw TOML content keyed by path."""
1085 configs = cls(root={})
1087 for path, content in contents.items():
1088 configs.add_config(ConfigModel.from_content(content, path), Path(path))
1090 return configs
1092 def add_config(self, config: ConfigModel, path: Path) -> None:
1093 self.root[str(path)] = config
1095 def team_refs(self) -> set[str]:
1096 """Collect every team ref (lowercase, no leading `@`/`!`) that
1097 `compiled(teams=...)` would actually try to expand: refs written
1098 directly in a user-list field (scopes' `USER_LIST_FIELDS` and
1099 `large_scale_change.reviewers`), plus any refs reachable from those
1100 fields through `$alias`/`!$alias` chains.
1102 Meant for callers that need to know which teams to fetch/sync before
1103 calling `compiled(teams=...)`.
1105 Implemented as an offline compile (`teams=None`): aliases expand but
1106 team refs pass through unexpanded, so whatever refs remain in the
1107 compiled user-list fields are — by construction — exactly the refs a
1108 real compile will try to expand. A ref that only appears in a
1109 non-user-list field (e.g. an npm-style `@vendor/pkg/**` in `paths`)
1110 or inside an alias nothing references never survives into a compiled
1111 user-list field, so it is never collected. Raises the same config
1112 errors `compiled()` would (unknown alias, circular refs, ...), just
1113 earlier.
1114 """
1115 # Cheap pre-check: a team ref can only enter a compile as a literal
1116 # `@`/`!@` value in a user-list field or an alias value. Most repos
1117 # have none, and skipping the compile keeps this near-free for them.
1118 candidate_lists: list[list[str]] = []
1119 for config in self.root.values():
1120 candidate_lists.extend(config.aliases.values())
1121 for scope in config.scopes:
1122 for field in USER_LIST_FIELDS:
1123 candidate_lists.append(getattr(scope, field))
1124 if config.large_scale_change:
1125 candidate_lists.append(config.large_scale_change.reviewers)
1126 if not any(
1127 _split_team_ref(value) is not None
1128 for values in candidate_lists
1129 for value in values
1130 ):
1131 return set()
1133 refs: set[str] = set()
1135 def collect_refs(values: list[str]) -> None:
1136 for value in values:
1137 if (split := _split_team_ref(value)) is not None:
1138 refs.add(split[1].lower())
1140 for config in self.compiled(teams=None).get_config_models().values():
1141 if config.template:
1142 continue
1143 for scope in config.scopes:
1144 for field in USER_LIST_FIELDS:
1145 collect_refs(getattr(scope, field))
1146 if config.large_scale_change:
1147 collect_refs(config.large_scale_change.reviewers)
1149 return refs
1151 def compiled(
1152 self, teams: dict[str, list[str]] | None = None
1153 ) -> CompiledConfigModels:
1154 """Resolve the whole set into its effective, PR-independent form.
1156 Each non-template config is compiled once — extends merged, aliases
1157 expanded, paths anchored. Templates are NOT compiled standalone: a
1158 template scope may reference an alias the consuming config provides, and
1159 its paths anchor at the consumer. They are carried through untouched
1160 (folded into each consumer during that consumer's compile, and kept in
1161 the set for display).
1163 `teams` maps team refs (any case, without the leading `@`) to member
1164 usernames; passed straight through to each config's `compiled_config`
1165 (see there for case normalization and the `teams=None` vs provided
1166 semantics).
1168 The result is an immutable `CompiledConfigModels` — there is no way to
1169 compile it again, so the non-idempotent path anchoring can never
1170 double-apply.
1171 """
1172 effective: dict[str, ConfigModel] = {}
1173 for path, config in self.root.items():
1174 if config.template:
1175 effective[path] = config
1176 else:
1177 effective[path] = config.compiled_config(
1178 config_path=Path(path), other_configs=self, teams=teams
1179 )
1181 return CompiledConfigModels.from_config_models(effective)
1184class CompiledConfigModels(_ConfigModelsBase):
1185 """The effective configs used for matching: every non-template config is
1186 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
1188 def closest_config(self, file_path: Path) -> ConfigModel:
1189 """Return the closest non-template config governing this file."""
1190 config_path = self.closest_config_path(file_path)
1191 if config_path is None:
1192 raise ValueError(f"No config found for {file_path}")
1193 return self.root[config_path]
1195 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
1196 """The primary (repo-root) config's large-scale-change section, if any.
1198 The primary was compiled by `compiled()`, so its reviewers/labels are
1199 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
1200 true` repo root is a misconfiguration (templates are meant to be
1201 extended, not be the primary); it is passed through uncompiled, so its
1202 LSC would read with aliases unexpanded.
1203 """
1204 if CONFIG_FILENAME in self.root and (
1205 lsc := self.root[CONFIG_FILENAME].large_scale_change
1206 ):
1207 return lsc
1209 return LargeScaleChangeModel()
1211 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels:
1212 """
1213 Overlay PR-dependent scope gating: drop scopes that author rules disable
1214 for this pull request.
1216 This is the only PR-dependent step. The configs are already compiled, so
1217 each config's scopes are self-contained and dropping one is a plain list
1218 filter — no re-inheritance. Templates are passed through (they are never
1219 matched directly; their scopes already live in each consumer).
1220 """
1221 effective: dict[str, ConfigModel] = {}
1222 for config_path, config in self.root.items():
1223 if config.template:
1224 # Templates are never matched directly; pass them through.
1225 effective[config_path] = config
1226 continue
1228 kept_scopes = [
1229 scope
1230 for scope in config.scopes
1231 if scope.matches_author(author_username)
1232 ]
1233 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
1235 return CompiledConfigModels.from_config_models(effective)