Coverage for src/pullapprove/config.py: 94%
494 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-09-17 14:55 -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")
93_REPEAT_OPS = {sre_parse.MAX_REPEAT, sre_parse.MIN_REPEAT}
96def _has_nested_quantifiers(data: Any) -> bool:
97 """Detect patterns like (a+)+ that cause catastrophic backtracking."""
98 for op, av in data:
99 if op in _REPEAT_OPS:
100 if _contains_quantifier(av[2]):
101 return True
102 elif op == sre_parse.SUBPATTERN:
103 if _has_nested_quantifiers(av[-1]):
104 return True
105 elif op == sre_parse.BRANCH: # noqa: SIM102
106 if any(_has_nested_quantifiers(branch) for branch in av[1]):
107 return True
108 return False
111def _contains_quantifier(data: Any) -> bool:
112 for op, av in data:
113 if op in _REPEAT_OPS:
114 return True
115 elif op == sre_parse.SUBPATTERN:
116 if _contains_quantifier(av[-1]):
117 return True
118 elif op == sre_parse.BRANCH: # noqa: SIM102
119 if any(_contains_quantifier(branch) for branch in av[1]):
120 return True
121 return False
124_TEAM_REF_SEGMENT = r"[a-zA-Z0-9][a-zA-Z0-9\-_.]*"
125_TEAM_REF_RE = re.compile(rf"^{_TEAM_REF_SEGMENT}(/{_TEAM_REF_SEGMENT})+$")
127# Fields that hold reviewer identities (plain usernames, `$aliases`, and
128# `@team` refs) rather than paths/code/labels. Team refs only expand here.
129USER_LIST_FIELDS = ("authors", "reviewers", "alternates", "cc")
131# Roster fields where "!" performs compile-time subtraction (remove from the
132# resolved list) rather than surviving as a match-time predicate. `authors`
133# is deliberately excluded — its "!" entries are consumed by `matches_author`.
134ROSTER_FIELDS = ("reviewers", "alternates", "cc")
137def _split_team_ref(value: str) -> tuple[str, str] | None:
138 """Split a value into `(prefix, ref)` if it has team-reference shape
139 (`@org/team` or `!@org/team`), `prefix` being `"!"` or `""`. Returns
140 `None` for anything else, so callers fall back to their own handling of
141 plain values.
142 """
143 if value.startswith("!@"):
144 return "!", value[2:]
145 if value.startswith("@"):
146 return "", value[1:]
147 return None
150def is_unexpanded_ref(value: str) -> bool:
151 """True if the value is a `$alias` or `@team` reference (optionally
152 negated with a leading `!`) that has not been expanded — as opposed to a
153 plain username. Only an offline compile (`teams=None`) leaves such values
154 in user-list fields.
155 """
156 return value.removeprefix("!").startswith(("$", "@"))
159def _validate_team_refs(values: list[str]) -> list[str]:
160 """Team references (`@org/team`, `!@org/team`) need at least two
161 slash-separated segments. This only checks shape — membership is resolved
162 later, at compile time, against the caller-provided `teams` mapping.
164 Any other value containing "@" is rejected too — that shape is reserved
165 for team references (and, in future, email-style identifiers), so it
166 can't be confused with a plain username.
167 """
168 for value in values:
169 split = _split_team_ref(value)
170 if split is None:
171 if "@" in value:
172 raise ValueError(
173 f"Invalid value '{value}': email addresses are not "
174 "supported here — use the platform username"
175 )
176 continue
178 _prefix, ref = split
179 if not _TEAM_REF_RE.match(ref):
180 raise ValueError(
181 f"Invalid team reference '{value}': team references need the "
182 "'org/team' form"
183 )
184 return values
187def _expand_team_ref(
188 ref: str, teams: dict[str, list[str]] | None, prefix: str
189) -> list[str]:
190 """Expand a single team reference (without its `@`/`!@`) to member usernames.
192 `teams=None` is the offline/CLI mode: the library never calls out to
193 GitHub/GitLab itself, so with no mapping provided the reference passes
194 through unexpanded rather than erroring. With a `teams` mapping (even an
195 empty one), an unresolvable ref is a loud config error, matching how
196 unknown `$aliases` are handled.
197 """
198 if teams is None:
199 return [f"{prefix}@{ref}"]
201 members = teams.get(ref.lower())
202 if members is None:
203 raise ValueError(f"Unknown team: {prefix}@{ref}")
205 return [f"{prefix}{member}" for member in members]
208def _expand_aliases(
209 values: list[str],
210 aliases: dict[str, list[str]],
211 teams: dict[str, list[str]] | None = None,
212 expand_teams: bool = False,
213 _seen: set[str] | None = None,
214 _path: list[str] | None = None,
215) -> list[str]:
216 """Replace alias references in a list with their mapped values recursively.
218 Team references (`@org/team`) are only expanded when `expand_teams` is
219 True — user-list fields (reviewers, alternates, authors, cc). Elsewhere
220 (paths, code, labels) a leading `@` is a literal string, e.g. npm-style
221 scoped path patterns like `@vendor/pkg/**`. Teams are always leaf nodes —
222 their values are plain usernames, never other refs — so expanding one
223 never recurses further.
224 """
225 if _seen is None:
226 _seen = set()
227 if _path is None:
228 _path = []
230 expanded: list[str] = []
231 for value in values:
232 # Support negated aliases like "!$team" -> ["!alice", "!bob"]
233 if value.startswith("!$"):
234 prefix = "!"
235 alias_ref = value[2:]
236 elif value.startswith("$"):
237 prefix = ""
238 alias_ref = value[1:]
239 elif expand_teams and (team_ref := _split_team_ref(value)) is not None:
240 team_prefix, ref = team_ref
241 expanded.extend(_expand_team_ref(ref=ref, teams=teams, prefix=team_prefix))
242 continue
243 else:
244 expanded.append(value)
245 continue
247 if alias_ref in _seen:
248 # Cycle detected, raise an error with the cycle path
249 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref]
250 raise ValueError(
251 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}"
252 )
253 if alias_ref in aliases:
254 _seen.add(alias_ref)
255 _path.append(alias_ref)
256 # Recursively expand the alias values
257 nested_expanded = _expand_aliases(
258 aliases[alias_ref],
259 aliases=aliases,
260 teams=teams,
261 expand_teams=expand_teams,
262 _seen=_seen,
263 _path=_path,
264 )
265 if prefix:
266 expanded.extend(prefix + v for v in nested_expanded)
267 else:
268 expanded.extend(nested_expanded)
269 _path.pop()
270 _seen.remove(alias_ref)
271 else:
272 # Unknown alias — surface it loudly instead of silently dropping the
273 # reference (a typo'd alias would otherwise vanish reviewers/paths).
274 raise ValueError(f"Unknown alias: {prefix}${alias_ref}")
276 # Remove duplicates while preserving order
277 return list(dict.fromkeys(expanded))
280def _apply_negations(values: list[str]) -> list[str]:
281 """Compile-time subtraction for roster fields (`ROSTER_FIELDS`): a "!name"
282 entry removes "name" from the resolved list instead of surviving as a
283 match-time rule (contrast `authors`, handled by `matches_author`).
285 A list still holding an unexpanded `@team` ref (offline compile, i.e.
286 teams=None) is returned untouched — it isn't fully resolved yet, so
287 subtraction can't run, and the partially-resolved config must round-trip
288 unchanged. `$aliases` must already be expanded by the caller.
290 The wildcard+negation error fires BEFORE that early return: it's a check
291 on the written form (`"*"` alongside any `"!"` entry, team ref or not),
292 so an offline compile must reject it the same way the server will — a
293 `pullapprove check` that passes locally can't then error in production.
294 """
295 if "*" in values and any(v.startswith("!") for v in values):
296 raise ValueError(
297 'Negation cannot be combined with "*" in reviewers/alternates/cc '
298 "(wildcard exclusion is not supported)"
299 )
301 if any(_split_team_ref(v) is not None for v in values):
302 return values
304 negations = {v[1:].lower() for v in values if v.startswith("!")}
305 return [v for v in values if not v.startswith("!") and v.lower() not in negations]
308def _validate_review_counts(label: str, data: dict[str, Any]) -> None:
309 """Compile-time rejection of negative require/request/author_value.
310 Shared by scopes and large_scale_change (which has no
311 request/author_value — the `.get` defaults pass trivially there).
313 These are compile-time checks (not field validators) because negative
314 values were previously accepted, so compiled configs stored inside old
315 processing results must keep parsing (where they keep their old
316 behavior: a negative require always passed, a negative request
317 requested nobody).
318 """
319 for field in ("require", "request", "author_value"):
320 if data.get(field, 0) < 0:
321 raise ValueError(f"{label}: {field} cannot be negative")
324def _resolve_extends_path(extending_path: str, extends_ref: str) -> str:
325 """Resolve an `extends` reference to a canonical repo-relative config key.
327 - `/x` is repo-root-relative.
328 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending
329 file's directory.
331 Raises if the reference escapes above the repo root.
332 """
333 if extends_ref.startswith("/"):
334 resolved = posixpath.normpath(extends_ref.lstrip("/"))
335 else:
336 base_dir = posixpath.dirname(extending_path)
337 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref))
339 if resolved == ".." or resolved.startswith("../"):
340 raise ValueError(
341 f"Invalid extends path: '{extends_ref}' points above the repo root"
342 )
344 return resolved
347def matches_path_patterns(*, path: Path, patterns: list[str]) -> bool:
348 """Whether `path` matches any of the config's path globs.
350 The one definition of the config's glob semantics.
351 """
352 # TODO paths shouldn't start with /
353 return glob.globmatch(
354 path,
355 patterns,
356 flags=glob.GLOBSTAR | glob.BRACE | glob.NEGATE | glob.IGNORECASE | glob.DOTGLOB,
357 )
360def _anchor_path(base_dir: str, pattern: str) -> str:
361 """Anchor a scope path glob at `base_dir` (the owning config's directory).
363 Scope paths are written relative to the config they live in. A leading `/`
364 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is
365 preserved. With an empty `base_dir` (root config) the pattern is unchanged.
366 """
367 negate = pattern.startswith("!")
368 if negate:
369 pattern = pattern[1:]
371 if pattern.startswith("/"):
372 anchored = pattern.lstrip("/")
373 elif base_dir:
374 anchored = f"{base_dir}/{pattern}"
375 else:
376 anchored = pattern
378 return f"!{anchored}" if negate else anchored
381class OwnershipChoices(StrEnum):
382 EMPTY = ""
383 APPEND = "append"
384 GLOBAL = "global"
387# A GitHub App has two names: the login it posts as (`name[bot]`) and the slug
388# that owns its check runs (`name`). The suffix is how the engine tells a bot
389# from a person -- a `[bot]` account never counts as a human and never sits in
390# a roster, no declaration needed.
391BOT_LOGIN_SUFFIX = "[bot]"
394def is_bot_login(username: str) -> bool:
395 """The engine rule, in one place: a `[bot]` account is never a person."""
396 return username.lower().endswith(BOT_LOGIN_SUFFIX)
399def _split_check_ref(ref: str) -> tuple[str, str]:
400 """An `unless` ref as (producer slug lowercased, check name).
402 Split on the FIRST `/`: App slugs cannot contain one, check names can.
403 Both parts are stripped -- the validator checks THIS function's output, so
404 a ref that validates is exactly a ref that matches at runtime.
405 """
406 producer, _, name = ref.partition("/")
407 return producer.strip().lower(), name.strip()
410def _reject_bot_reviewers(values: list[str], *, where: str) -> None:
411 """Reject any `[bot]` account listed in a roster surface (a scope's
412 reviewers/alternates/cc, or large_scale_change.reviewers).
414 A bot is never a person, so it can never sit in a roster: bots that open
415 pull requests are routed with `authors`, and bots that attest are
416 referenced from `unless`. Checked after alias expansion so `$alias`
417 indirection can't smuggle one in. `where` names the surface; the message
418 becomes the git-host commit status, so it stays under ~130 chars.
419 """
420 for entry in values:
421 if is_bot_login(entry):
422 raise ValueError(
423 f"{where}: '{entry}' is a bot and cannot be listed as a reviewer"
424 )
427def _first_case_insensitive_duplicate(values: Iterable[str]) -> str | None:
428 """The first value whose lowercased form was already seen, else None."""
429 seen: set[str] = set()
430 for value in values:
431 if value.lower() in seen:
432 return value
433 seen.add(value.lower())
434 return None
437class ScopeModel(BaseModel):
438 model_config = ConfigDict(extra="forbid")
440 # Required fields
441 name: str = Field(min_length=1)
442 paths: list[str] = Field(min_length=1)
444 # Optional fields
446 # Expanded version of lines could be dict
447 # with fnmatch, regex, exclude patterns, etc?
448 code: list[str] = []
450 # This only filtering field that can't be used with raw diff/files...
451 # If we get into that, the others are:
452 # - labels
453 # - ref
454 # - statuses
455 # - dates
456 # - body
457 # - title
458 # - other scopes
459 # (this is how I ended up with expressions...
460 # I'm not trying to build a general purpose workflow tool,
461 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy)
462 authors: list[str] = []
464 # (defaults should be the "empty" values)
465 description: str = ""
466 reviewers: list[str] = []
467 alternates: list[str] = []
468 cc: list[str] = []
470 # Review scoring
471 # Negative values are rejected at compile time (_validate_review_counts)
472 require: int = 0
473 author_value: int = 0
475 # The checks that can attest this scope's review requirement away. Each ref
476 # is "producer/check-name" -- the slug of the App that produces the check,
477 # then the check run's name, split on the first "/". When every listed
478 # check has completed successfully on the current head commit, the scope is
479 # waived: its requirement drops to zero and the evidence is stored on the
480 # result. Failed, skipped, pending, missing -- the requirement stands
481 # unchanged; passing is the only state that subtracts.
482 #
483 # Deliberately placed next to the `require` it undermines, so reading a
484 # scope always reveals whether its human review is removable. A list is
485 # implicitly "all must pass" -- OR, thresholds, and 2-of-3 belong in the
486 # producer, which can post one combined check.
487 unless: list[str] = []
489 # How scopes are combined
490 ownership: OwnershipChoices = OwnershipChoices.EMPTY
492 # Actionable items
493 request: int = 0
494 labels: list[str] = []
495 instructions: str = ""
497 # Approval checklist
498 checklist: Checklist | None = None
500 @field_validator("name", mode="after")
501 @classmethod
502 def validate_name(cls, name: str) -> str:
503 if "," in name:
504 raise ValueError("Scope name cannot contain commas")
505 return name
507 @field_validator(*USER_LIST_FIELDS, mode="after")
508 @classmethod
509 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
510 return _validate_team_refs(values)
512 @field_validator("unless", mode="after")
513 @classmethod
514 def validate_unless_refs(cls, values: list[str]) -> list[str]:
515 """A check name alone is worthless -- any workflow with `checks: write`
516 can post any name -- so a bare name is a parse error, not a default.
517 Messages stay under ~130 chars: they become the git-host commit status.
518 """
519 for ref in values:
520 producer, name = _split_check_ref(ref)
521 if "/" not in ref or not producer or not name:
522 raise ValueError(
523 f"unless: '{ref}' must be 'producer/check-name' -- the App "
524 "slug that produces the check, then the check's name"
525 )
526 # To most users PullApprove *is* "the check" on their PRs.
527 # PullApprove never creates these checks, it only reads them -- and
528 # a config waiting on our own status would deadlock politely.
529 if producer == "pullapprove":
530 raise ValueError(
531 f"unless: '{ref}' references PullApprove itself -- "
532 "PullApprove never creates checks, it only reads them"
533 )
534 return values
536 @field_validator("code", mode="after")
537 @classmethod
538 def validate_code_patterns(cls, code: list[str]) -> list[str]:
539 for pattern in code:
540 try:
541 parsed = sre_parse.parse(pattern)
542 except re.error as e:
543 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None
544 if _has_nested_quantifiers(parsed):
545 raise ValueError(
546 f"Regex pattern '{pattern}' contains nested quantifiers, "
547 "which can cause catastrophic backtracking."
548 )
549 return code
551 @model_validator(mode="after")
552 def validate_reviewers_for_require(self) -> ScopeModel:
553 all_reviewers = self.reviewers + self.alternates
555 # Skip if wildcard - anyone can review
556 if "*" in all_reviewers:
557 return self
559 # Skip if aliases or team refs (possibly negated with "!") are not yet
560 # expanded. Re-validation after compilation only re-checks refs that
561 # actually resolve — an offline compile (teams=None) leaves refs
562 # unexpanded, so this same skip fires again there too.
563 if any(is_unexpanded_ref(r) for r in all_reviewers):
564 return self
566 if len(all_reviewers) < self.require:
567 raise ValueError(
568 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified"
569 )
570 return self
572 def counts_toward(self, username: str) -> bool:
573 """Whether this person's review counts for this scope.
575 Its named `reviewers` and its `alternates`, plus everybody when
576 `reviewers` holds the wildcard. A wildcard in `alternates` does not
577 count everybody: the wildcard describes the scope's `reviewers`, not
578 its backups.
580 Case-insensitive, like every other username comparison here.
581 """
582 if "*" in self.reviewers:
583 return True
584 username_lower = username.lower()
585 return any(
586 username_lower == r.lower() for r in self.reviewers + self.alternates
587 )
589 def author_points(self, author_username: str) -> int:
590 """Author points only count if the author is explicitly listed as a
591 reviewer (a wildcard is not converted to usernames)."""
592 if author_username.lower() in {r.lower() for r in self.reviewers}:
593 return self.author_value
594 return 0
596 def unless_refs(self) -> list[tuple[str, str]]:
597 """Each `unless` entry as (producer slug lowercased, check name)."""
598 return [_split_check_ref(ref) for ref in self.unless]
600 def unsolvable_reason(self, author_username: str) -> str | None:
601 """
602 Explain why this scope can never pass for a PR authored by this user,
603 or None if it can.
605 Messages must stay under ~130 chars: they flow into the git-host
606 commit status description, which the adapters slice to 140.
607 """
608 if "*" in self.reviewers:
609 # Anyone can review, so any require is satisfiable
610 return None
612 # Count eligible reviewers (excluding author who can't self-approve)
613 eligible_reviewers = {r.lower() for r in self.reviewers + self.alternates} - {
614 author_username.lower()
615 }
616 max_possible_points = len(eligible_reviewers) + self.author_points(
617 author_username
618 )
620 if self.require > 0 and max_possible_points < self.require:
621 if not eligible_reviewers:
622 return (
623 "PR author is the only reviewer/alternate and cannot self-approve"
624 )
625 return f"require={self.require} but only {max_possible_points} possible approvals (excluding author)"
627 return None
629 def ownership_marker(self) -> str:
630 """The glyph a non-default ownership puts in front of the scope name.
632 Split out from printed_name so a renderer that styles the marker apart
633 from the name doesn't have to know which glyph goes with which mode.
634 """
635 match self.ownership:
636 case OwnershipChoices.APPEND:
637 return "+"
638 case OwnershipChoices.GLOBAL:
639 return "*"
641 return ""
643 def printed_name(self) -> str:
644 return self.ownership_marker() + self.name
646 def __eq__(self, other: object) -> bool:
647 if not isinstance(other, ScopeModel):
648 return NotImplemented
649 return self.name == other.name
651 def matches_path(self, path: Path) -> bool:
652 return matches_path_patterns(path=path, patterns=self.paths)
654 def matches_code(self, code: str) -> Generator[dict[str, int]]:
655 patterns = getattr(self, "_code_regex_patterns", [])
656 if not patterns:
657 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code]
658 self._code_regex_patterns = patterns
660 for pattern in patterns:
661 for match in pattern.finditer(code):
662 start_index = match.start()
663 end_index = match.end()
665 start_line = code.count("\n", 0, start_index) + 1
666 start_col = start_index - code.rfind("\n", 0, start_index)
668 end_line = code.count("\n", 0, end_index) + 1
669 end_col = end_index - code.rfind("\n", 0, end_index)
671 yield {
672 "start_line": start_line,
673 "start_col": start_col,
674 "end_line": end_line,
675 "end_col": end_col,
676 }
678 def matches_author(self, author_username: str) -> bool:
679 if not self.authors:
680 # No authors specified, so assume it matches
681 return True
683 author_username_lower = author_username.lower()
685 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")]
686 authors = [a.lower() for a in self.authors if not a.startswith("!")]
688 if author_username_lower in negated_authors:
689 # If the author is in the negated list, return False
690 return False
692 if not authors:
693 # Negation-only: everyone not negated matches
694 return True
696 return author_username_lower in authors
699class LargeScaleChangeModel(BaseModel):
700 model_config = ConfigDict(extra="forbid")
702 # Note, an LSC only applies to diffs, not raw files,
703 # because we have to know what *changed*.
705 # Pretty similar to a scope, but more manual.
706 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one.
707 # Negative values are rejected at compile time (_validate_review_counts)
708 require: int = 1
709 reviewers: list[str] = [] # Field(min_length=1)
710 # min_paths: int = 300
711 # min_lines: int = 3000
712 labels: list[str] = []
713 # really need author value too...?
715 @field_validator("reviewers", mode="after")
716 @classmethod
717 def validate_team_ref_shape(cls, values: list[str]) -> list[str]:
718 return _validate_team_refs(values)
720 def unsolvable_reason(self, author_username: str) -> str | None:
721 """
722 Explain why this LSC can never pass for a PR authored by this user,
723 or None if it can.
725 Messages must stay under ~130 chars: they flow into the git-host
726 commit status description, which the adapters slice to 140.
727 """
728 if "*" in self.reviewers:
729 # Anyone can review, so any require is satisfiable
730 return None
732 if not self.reviewers:
733 # An empty roster is "configuration required", which
734 # process_large_scale_change reports on its own
735 return None
737 # Count eligible reviewers (excluding author who can't self-approve)
738 eligible_reviewers = {r.lower() for r in self.reviewers} - {
739 author_username.lower()
740 }
741 if self.require > 0 and len(eligible_reviewers) < self.require:
742 if not eligible_reviewers:
743 return "the PR author is the only reviewer and cannot self-approve"
744 return f"require={self.require} but only {len(eligible_reviewers)} possible approvals (excluding author)"
746 return None
749class ConfigModel(BaseModel):
750 model_config = ConfigDict(extra="forbid")
752 # Nothing is technically required
753 extends: list[str] = []
754 template: bool = False
755 aliases: dict[str, list[str]] = {}
756 large_scale_change: LargeScaleChangeModel | None = None
757 scopes: list[ScopeModel] = []
759 @field_validator("scopes", mode="after")
760 @classmethod
761 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]:
762 if dup := _first_case_insensitive_duplicate(scope.name for scope in scopes):
763 raise ValueError(f"Duplicate scope name: {dup}")
764 return scopes
766 @field_validator("extends", mode="before")
767 @classmethod
768 def validate_extends(cls, extends: list[str]) -> list[str]:
769 for i, path in enumerate(extends):
770 basename = Path(path).name
771 if not basename.startswith(CONFIG_FILENAME_PREFIX):
772 raise ValueError(
773 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'."
774 )
775 return extends
777 def compiled_config(
778 self,
779 config_path: Path,
780 other_configs: ConfigModels,
781 teams: dict[str, list[str]] | None = None,
782 ) -> ConfigModel:
783 """
784 Resolve `extends` and replace aliases, returning the effective config.
786 Two phases: flatten the whole extends chain into one merged (raw,
787 unexpanded) config, then expand aliases once. Expanding after the full
788 merge is what makes transitive inheritance and cross-chain alias
789 scoping work — an alias defined anywhere in the chain resolves anywhere.
791 `teams` maps team refs (any case, without the leading `@`) to member
792 usernames, and is only consulted for user-list fields (reviewers,
793 alternates, authors, cc, large_scale_change.reviewers). Keys are
794 lowercased here, so callers don't need to normalize case themselves.
795 With `teams=None` (the default), `@org/team` references in those
796 fields pass through unexpanded — the offline/CLI mode, since this
797 library never calls out to GitHub/GitLab itself. That
798 partially-resolved config is only sanctioned for offline use;
799 anything that needs real reviewer usernames must pass a `teams`
800 mapping.
802 Pure function of the raw `self` and `other_configs` (it never reads its
803 own anchored output), so it is safe to call uncached.
804 """
806 if teams is not None:
807 teams = {key.lower(): members for key, members in teams.items()}
809 compiled_data = self._merged_data(config_path, other_configs)
811 # Expand aliases for any aliasable list fields. Team refs only expand
812 # in user-list fields — paths/code/labels keep a leading "@" literal.
813 for scope in compiled_data["scopes"]:
814 for field in [
815 "paths",
816 "code",
817 "authors",
818 "reviewers",
819 "alternates",
820 "cc",
821 "labels",
822 ]:
823 if field in scope:
824 scope[field] = _expand_aliases(
825 scope[field],
826 compiled_data["aliases"],
827 teams=teams,
828 expand_teams=field in USER_LIST_FIELDS,
829 )
831 # Apply compile-time "!" subtraction to roster fields.
832 # `_apply_negations` leaves a field with an unexpanded team ref
833 # (offline compile, i.e. teams=None) untouched.
834 for scope in compiled_data["scopes"]:
835 for field in ROSTER_FIELDS:
836 if field in scope:
837 scope[field] = _apply_negations(scope[field])
838 _reject_bot_reviewers(
839 scope[field],
840 where=f"Scope '{scope['name']}' {field}",
841 )
843 # The "*" wildcard is only meaningful in `reviewers` — everywhere
844 # else it's matched as a literal username and silently does
845 # nothing: in `alternates` the scope stays pending forever, in
846 # `authors` the scope never applies to any PR, in `cc` nobody is
847 # notified. Reject it at compile time (after alias expansion, so
848 # `$alias` indirection can't smuggle it in) rather than as a
849 # ScopeModel validator, because compiled configs stored inside
850 # old processing results must keep parsing.
851 # Messages must stay under ~130 chars: they become the git-host
852 # commit status description, which the adapters slice to 140.
853 for field, hint in (
854 (
855 "authors",
856 "remove it (a scope with no authors applies to any author)",
857 ),
858 (
859 "alternates",
860 "add it to reviewers instead (wildcard reviewers are never auto-requested)",
861 ),
862 ("cc", "remove it"),
863 ):
864 if "*" in scope.get(field, []):
865 raise ValueError(
866 f"Scope '{scope['name']}': \"*\" is not supported in "
867 f"{field} — {hint}"
868 )
870 _validate_review_counts(f"Scope '{scope['name']}'", scope)
872 if large_scale_change := compiled_data.get("large_scale_change"):
873 large_scale_change["reviewers"] = _expand_aliases(
874 large_scale_change["reviewers"],
875 compiled_data["aliases"],
876 teams=teams,
877 expand_teams=True,
878 )
879 large_scale_change["labels"] = _expand_aliases(
880 large_scale_change["labels"],
881 compiled_data["aliases"],
882 )
883 large_scale_change["reviewers"] = _apply_negations(
884 large_scale_change["reviewers"]
885 )
886 _reject_bot_reviewers(
887 large_scale_change["reviewers"],
888 where="large_scale_change reviewers",
889 )
890 _validate_review_counts("large_scale_change", large_scale_change)
892 # Anchor each scope's paths at the directory tagged during flattening
893 # (after alias expansion, so any `$path-alias` is resolved first). The
894 # transient tag is popped so it never reaches the model.
895 # Paths are relative to the config that owns the scope, `/` for
896 # repo-root-absolute.
897 for entry in compiled_data["scopes"]:
898 anchor_dir = entry.pop("_anchor_dir", "")
899 entry["paths"] = [_anchor_path(anchor_dir, p) for p in entry["paths"]]
901 # The compiled config is the self-contained effective config: extends
902 # are already merged in and aliases already expanded, so drop both. This
903 # keeps stored results lean and makes the compiled form standalone (it
904 # can never dangle on a missing extends target or re-expand differently).
905 compiled_data["extends"] = []
906 compiled_data["aliases"] = {}
908 return ConfigModel.from_data(
909 data=compiled_data,
910 path=config_path,
911 )
913 def _merged_data(
914 self,
915 config_path: Path,
916 other_configs: ConfigModels,
917 _in_progress: list[str] | None = None,
918 _seen: set[str] | None = None,
919 ) -> dict[str, Any]:
920 """
921 Flatten the `extends` chain into one merged, *unexpanded* config dict.
923 Parents are merged before this config (so a child can specialize), with
924 aliases unioned child-wins and the large-scale-change config taken from
925 the child if set else the first parent that defines one.
927 `_in_progress` is the current ancestor path, used to detect circular
928 extends. `_seen` is every config already merged into this flatten, used
929 to merge a shared ancestor only once (diamond dedup).
930 """
931 if _in_progress is None:
932 _in_progress = []
933 if _seen is None:
934 _seen = set()
936 config_path_str = str(config_path)
937 if config_path_str in _in_progress:
938 cycle = _in_progress[_in_progress.index(config_path_str) :] + [
939 config_path_str
940 ]
941 raise ValueError(
942 f"Circular reference detected in extends: {' -> '.join(cycle)}"
943 )
944 _in_progress.append(config_path_str)
946 inherited_scopes: list[dict[str, Any]] = []
947 inherited_aliases: dict[str, list[str]] = {}
948 inherited_lsc: dict[str, Any] | None = None
950 for extend_path in self.extends:
951 resolved_path = _resolve_extends_path(config_path_str, extend_path)
952 if resolved_path not in other_configs:
953 raise ValueError(
954 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')"
955 )
956 if resolved_path in _seen:
957 # Already merged via another branch (diamond) — skip the dup.
958 continue
960 parent_data = other_configs[resolved_path]._merged_data(
961 Path(resolved_path), other_configs, _in_progress, _seen
962 )
963 inherited_scopes = inherited_scopes + parent_data["scopes"]
964 inherited_aliases = inherited_aliases | parent_data["aliases"]
965 inherited_lsc = inherited_lsc or parent_data["large_scale_change"]
967 merged = self.model_dump()
968 merged["scopes"] = inherited_scopes + merged["scopes"]
969 merged["aliases"] = inherited_aliases | merged["aliases"]
970 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc
972 # Tag each scope with the directory its paths should anchor at. A scope's
973 # paths are relative to the config that owns it, so the first
974 # non-template config to consume a scope claims it: a non-template's own
975 # scopes (and any it inherits from a template) anchor at its directory,
976 # while a template defers to its consumer. `setdefault` means an
977 # already-tagged scope (from a non-template ancestor) keeps its anchor.
978 if not self.template:
979 base_dir = posixpath.dirname(config_path_str)
980 for entry in merged["scopes"]:
981 entry.setdefault("_anchor_dir", base_dir)
983 _seen.add(config_path_str)
984 _in_progress.pop()
986 return merged
988 @classmethod
989 def from_filesystem(cls, path: Path | str) -> ConfigModel:
990 with open(path, "rb") as f:
991 return cls.from_data(tomllib.load(f), path)
993 @classmethod
994 def from_content(cls, content: str, path: Path | str) -> ConfigModel:
995 return cls.from_data(tomllib.loads(content), path)
997 @classmethod
998 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel:
999 return cls(**data)
1002class _ConfigModelsBase(RootModel):
1003 """Shared storage and accessors for a set of configs keyed by repo path."""
1005 root: dict[str, ConfigModel]
1007 @classmethod
1008 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self:
1009 """Build from a dict of already-constructed configs keyed by path."""
1010 configs = cls(root={})
1011 for path, config_model in models.items():
1012 configs.root[str(Path(path))] = config_model
1013 return configs
1015 def get_config_models(self) -> dict[str, ConfigModel]:
1016 return dict(self.root.items())
1018 def __bool__(self) -> bool:
1019 return bool(self.root)
1021 def __getitem__(self, key: str) -> ConfigModel:
1022 return self.root[key]
1024 def __contains__(self, key: str) -> bool:
1025 return key in self.root
1027 def __len__(self) -> int:
1028 return len(self.root)
1031class ConfigModels(_ConfigModelsBase):
1032 """Configs exactly as loaded from the repo — extends unresolved, aliases
1033 unexpanded, paths unanchored. Build the set up, then call `compiled()`."""
1035 def declared_check_names(self) -> set[str]:
1036 """Every check-run name an effective (non-template) scope's `unless`
1037 references, across the whole config set.
1039 Collected from an offline compile (`teams=None`, the `team_refs`
1040 precedent) so a template's refs count only where a consumer actually
1041 inherits them -- an unconsumed template must not make every pull
1042 request fetch (or error on) checks that nothing evaluates. `unless`
1043 refs are literal strings (aliases and teams never expand inside
1044 them), so the offline compile is exact. Empty for the common case (no
1045 `unless` anywhere), which is what lets the processor skip fetching
1046 check runs entirely.
1047 """
1048 return {
1049 name
1050 for config in self.compiled().root.values()
1051 if not config.template
1052 for scope in config.scopes
1053 for _, name in scope.unless_refs()
1054 }
1056 @classmethod
1057 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels:
1058 """Load configs from a dict of parsed config data keyed by path."""
1059 configs = cls(root={})
1061 for path, config_data in data.items():
1062 config = ConfigModel.from_data(config_data, Path(path))
1063 configs.add_config(config, Path(path))
1065 return configs
1067 @classmethod
1068 def from_contents(cls, contents: dict[str, str]) -> ConfigModels:
1069 """Load configs from a dict of raw TOML content keyed by path."""
1070 configs = cls(root={})
1072 for path, content in contents.items():
1073 configs.add_config(ConfigModel.from_content(content, path), Path(path))
1075 return configs
1077 def add_config(self, config: ConfigModel, path: Path) -> None:
1078 self.root[str(path)] = config
1080 def team_refs(self) -> set[str]:
1081 """Collect every team ref (lowercase, no leading `@`/`!`) that
1082 `compiled(teams=...)` would actually try to expand: refs written
1083 directly in a user-list field (scopes' `USER_LIST_FIELDS` and
1084 `large_scale_change.reviewers`), plus any refs reachable from those
1085 fields through `$alias`/`!$alias` chains.
1087 Meant for callers that need to know which teams to fetch/sync before
1088 calling `compiled(teams=...)`.
1090 Implemented as an offline compile (`teams=None`): aliases expand but
1091 team refs pass through unexpanded, so whatever refs remain in the
1092 compiled user-list fields are — by construction — exactly the refs a
1093 real compile will try to expand. A ref that only appears in a
1094 non-user-list field (e.g. an npm-style `@vendor/pkg/**` in `paths`)
1095 or inside an alias nothing references never survives into a compiled
1096 user-list field, so it is never collected. Raises the same config
1097 errors `compiled()` would (unknown alias, circular refs, ...), just
1098 earlier.
1099 """
1100 # Cheap pre-check: a team ref can only enter a compile as a literal
1101 # `@`/`!@` value in a user-list field or an alias value. Most repos
1102 # have none, and skipping the compile keeps this near-free for them.
1103 candidate_lists: list[list[str]] = []
1104 for config in self.root.values():
1105 candidate_lists.extend(config.aliases.values())
1106 for scope in config.scopes:
1107 for field in USER_LIST_FIELDS:
1108 candidate_lists.append(getattr(scope, field))
1109 if config.large_scale_change:
1110 candidate_lists.append(config.large_scale_change.reviewers)
1111 if not any(
1112 _split_team_ref(value) is not None
1113 for values in candidate_lists
1114 for value in values
1115 ):
1116 return set()
1118 refs: set[str] = set()
1120 def collect_refs(values: list[str]) -> None:
1121 for value in values:
1122 if (split := _split_team_ref(value)) is not None:
1123 refs.add(split[1].lower())
1125 for config in self.compiled(teams=None).get_config_models().values():
1126 if config.template:
1127 continue
1128 for scope in config.scopes:
1129 for field in USER_LIST_FIELDS:
1130 collect_refs(getattr(scope, field))
1131 if config.large_scale_change:
1132 collect_refs(config.large_scale_change.reviewers)
1134 return refs
1136 def compiled(
1137 self, teams: dict[str, list[str]] | None = None
1138 ) -> CompiledConfigModels:
1139 """Resolve the whole set into its effective, PR-independent form.
1141 Each non-template config is compiled once — extends merged, aliases
1142 expanded, paths anchored. Templates are NOT compiled standalone: a
1143 template scope may reference an alias the consuming config provides, and
1144 its paths anchor at the consumer. They are carried through untouched
1145 (folded into each consumer during that consumer's compile, and kept in
1146 the set for display).
1148 `teams` maps team refs (any case, without the leading `@`) to member
1149 usernames; passed straight through to each config's `compiled_config`
1150 (see there for case normalization and the `teams=None` vs provided
1151 semantics).
1153 The result is an immutable `CompiledConfigModels` — there is no way to
1154 compile it again, so the non-idempotent path anchoring can never
1155 double-apply.
1156 """
1157 effective: dict[str, ConfigModel] = {}
1158 for path, config in self.root.items():
1159 if config.template:
1160 effective[path] = config
1161 else:
1162 effective[path] = config.compiled_config(
1163 config_path=Path(path), other_configs=self, teams=teams
1164 )
1166 return CompiledConfigModels.from_config_models(effective)
1169class CompiledConfigModels(_ConfigModelsBase):
1170 """The effective configs used for matching: every non-template config is
1171 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled."""
1173 def closest_config_path(self, file_path: Path) -> str | None:
1174 """The path of the closest non-template config governing this file,
1175 or None when nothing governs it."""
1176 for parent in file_path.parents:
1177 parent_config_path = str(parent / CONFIG_FILENAME)
1178 config = self.root.get(parent_config_path)
1179 if config is not None and not config.template:
1180 return parent_config_path
1181 return None
1183 def closest_config(self, file_path: Path) -> ConfigModel:
1184 """Return the closest non-template config governing this file."""
1185 config_path = self.closest_config_path(file_path)
1186 if config_path is None:
1187 raise ValueError(f"No config found for {file_path}")
1188 return self.root[config_path]
1190 def get_default_large_scale_change(self) -> LargeScaleChangeModel:
1191 """The primary (repo-root) config's large-scale-change section, if any.
1193 The primary was compiled by `compiled()`, so its reviewers/labels are
1194 already alias-expanded (e.g. ["$backend"] -> usernames). A `template =
1195 true` repo root is a misconfiguration (templates are meant to be
1196 extended, not be the primary); it is passed through uncompiled, so its
1197 LSC would read with aliases unexpanded.
1198 """
1199 if CONFIG_FILENAME in self.root and (
1200 lsc := self.root[CONFIG_FILENAME].large_scale_change
1201 ):
1202 return lsc
1204 return LargeScaleChangeModel()
1206 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels:
1207 """
1208 Overlay PR-dependent scope gating: drop scopes that author rules disable
1209 for this pull request.
1211 This is the only PR-dependent step. The configs are already compiled, so
1212 each config's scopes are self-contained and dropping one is a plain list
1213 filter — no re-inheritance. Templates are passed through (they are never
1214 matched directly; their scopes already live in each consumer).
1215 """
1216 effective: dict[str, ConfigModel] = {}
1217 for config_path, config in self.root.items():
1218 if config.template:
1219 # Templates are never matched directly; pass them through.
1220 effective[config_path] = config
1221 continue
1223 kept_scopes = [
1224 scope
1225 for scope in config.scopes
1226 if scope.matches_author(author_username)
1227 ]
1228 effective[config_path] = config.model_copy(update={"scopes": kept_scopes})
1230 return CompiledConfigModels.from_config_models(effective)