Coverage for src/pullapprove/config.py: 94%

489 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-09-09 12:02 -0500

1from __future__ import annotations 

2 

3import os 

4import posixpath 

5import re 

6import tomllib 

7import warnings 

8 

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 

16 

17from pydantic import ( 

18 BaseModel, 

19 ConfigDict, 

20 Field, 

21 RootModel, 

22 field_validator, 

23 model_validator, 

24) 

25from wcmatch import glob 

26 

27from .checklists import Checklist 

28 

29 

30def _resolve_config_filename_prefix() -> str: 

31 """The name every config file starts with, for this process. 

32 

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. 

40 

41 Read once at import: this is a per-process constant, not runtime state. 

42 

43 Two operational constraints this doesn't (and can't) enforce: 

44 

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). 

51 

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" 

61 

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 ) 

73 

74 return prefix 

75 

76 

77CONFIG_FILENAME_PREFIX = _resolve_config_filename_prefix() 

78CONFIG_FILENAME = f"{CONFIG_FILENAME_PREFIX}.toml" 

79 

80 

81def is_config_filename(basename: str) -> bool: 

82 """Whether a filename is a config file (CODEREVIEW.toml, 

83 CODEREVIEW.template.toml). 

84 

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") 

91 

92 

93_REPEAT_OPS = {sre_parse.MAX_REPEAT, sre_parse.MIN_REPEAT} 

94 

95 

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 

109 

110 

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 

122 

123 

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})+$") 

126 

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") 

130 

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") 

135 

136 

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 

148 

149 

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(("$", "@")) 

157 

158 

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. 

163 

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 

177 

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 

185 

186 

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. 

191 

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}"] 

200 

201 members = teams.get(ref.lower()) 

202 if members is None: 

203 raise ValueError(f"Unknown team: {prefix}@{ref}") 

204 

205 return [f"{prefix}{member}" for member in members] 

206 

207 

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. 

217 

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 = [] 

229 

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 

246 

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}") 

275 

276 # Remove duplicates while preserving order 

277 return list(dict.fromkeys(expanded)) 

278 

279 

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`). 

284 

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. 

289 

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 ) 

300 

301 if any(_split_team_ref(v) is not None for v in values): 

302 return values 

303 

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] 

306 

307 

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). 

312 

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") 

322 

323 

324def _resolve_extends_path(extending_path: str, extends_ref: str) -> str: 

325 """Resolve an `extends` reference to a canonical repo-relative config key. 

326 

327 - `/x` is repo-root-relative. 

328 - everything else (`../x`, `dir/x`, bare `x`) is relative to the extending 

329 file's directory. 

330 

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)) 

338 

339 if resolved == ".." or resolved.startswith("../"): 

340 raise ValueError( 

341 f"Invalid extends path: '{extends_ref}' points above the repo root" 

342 ) 

343 

344 return resolved 

345 

346 

347def matches_path_patterns(*, path: Path, patterns: list[str]) -> bool: 

348 """Whether `path` matches any of the config's path globs. 

349 

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 ) 

358 

359 

360def _anchor_path(base_dir: str, pattern: str) -> str: 

361 """Anchor a scope path glob at `base_dir` (the owning config's directory). 

362 

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:] 

370 

371 if pattern.startswith("/"): 

372 anchored = pattern.lstrip("/") 

373 elif base_dir: 

374 anchored = f"{base_dir}/{pattern}" 

375 else: 

376 anchored = pattern 

377 

378 return f"!{anchored}" if negate else anchored 

379 

380 

381class OwnershipChoices(StrEnum): 

382 EMPTY = "" 

383 APPEND = "append" 

384 GLOBAL = "global" 

385 

386 

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]" 

392 

393 

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) 

397 

398 

399def _split_check_ref(ref: str) -> tuple[str, str]: 

400 """An `unless` ref as (producer slug lowercased, check name). 

401 

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() 

408 

409 

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). 

413 

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 ) 

425 

426 

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 

435 

436 

437class ScopeModel(BaseModel): 

438 model_config = ConfigDict(extra="forbid") 

439 

440 # Required fields 

441 name: str = Field(min_length=1) 

442 paths: list[str] = Field(min_length=1) 

443 

444 # Optional fields 

445 

446 # Expanded version of lines could be dict 

447 # with fnmatch, regex, exclude patterns, etc? 

448 code: list[str] = [] 

449 

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] = [] 

463 

464 # (defaults should be the "empty" values) 

465 description: str = "" 

466 reviewers: list[str] = [] 

467 alternates: list[str] = [] 

468 cc: list[str] = [] 

469 

470 # Review scoring 

471 # Negative values are rejected at compile time (_validate_review_counts) 

472 require: int = 0 

473 author_value: int = 0 

474 

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] = [] 

488 

489 # How scopes are combined 

490 ownership: OwnershipChoices = OwnershipChoices.EMPTY 

491 

492 # Actionable items 

493 request: int = 0 

494 labels: list[str] = [] 

495 instructions: str = "" 

496 

497 # Approval checklist 

498 checklist: Checklist | None = None 

499 

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 

506 

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) 

511 

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 

535 

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 

550 

551 @model_validator(mode="after") 

552 def validate_reviewers_for_require(self) -> ScopeModel: 

553 all_reviewers = self.reviewers + self.alternates 

554 

555 # Skip if wildcard - anyone can review 

556 if "*" in all_reviewers: 

557 return self 

558 

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 

565 

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 

571 

572 def author_points(self, author_username: str) -> int: 

573 """Author points only count if the author is explicitly listed as a 

574 reviewer (a wildcard is not converted to usernames).""" 

575 if author_username.lower() in {r.lower() for r in self.reviewers}: 

576 return self.author_value 

577 return 0 

578 

579 def unless_refs(self) -> list[tuple[str, str]]: 

580 """Each `unless` entry as (producer slug lowercased, check name).""" 

581 return [_split_check_ref(ref) for ref in self.unless] 

582 

583 def unsolvable_reason(self, author_username: str) -> str | None: 

584 """ 

585 Explain why this scope can never pass for a PR authored by this user, 

586 or None if it can. 

587 

588 Messages must stay under ~130 chars: they flow into the git-host 

589 commit status description, which the adapters slice to 140. 

590 """ 

591 if "*" in self.reviewers: 

592 # Anyone can review, so any require is satisfiable 

593 return None 

594 

595 # Count eligible reviewers (excluding author who can't self-approve) 

596 eligible_reviewers = {r.lower() for r in self.reviewers + self.alternates} - { 

597 author_username.lower() 

598 } 

599 max_possible_points = len(eligible_reviewers) + self.author_points( 

600 author_username 

601 ) 

602 

603 if self.require > 0 and max_possible_points < self.require: 

604 if not eligible_reviewers: 

605 return ( 

606 "PR author is the only reviewer/alternate and cannot self-approve" 

607 ) 

608 return f"require={self.require} but only {max_possible_points} possible approvals (excluding author)" 

609 

610 return None 

611 

612 def ownership_marker(self) -> str: 

613 """The glyph a non-default ownership puts in front of the scope name. 

614 

615 Split out from printed_name so a renderer that styles the marker apart 

616 from the name doesn't have to know which glyph goes with which mode. 

617 """ 

618 match self.ownership: 

619 case OwnershipChoices.APPEND: 

620 return "+" 

621 case OwnershipChoices.GLOBAL: 

622 return "*" 

623 

624 return "" 

625 

626 def printed_name(self) -> str: 

627 return self.ownership_marker() + self.name 

628 

629 def __eq__(self, other: object) -> bool: 

630 if not isinstance(other, ScopeModel): 

631 return NotImplemented 

632 return self.name == other.name 

633 

634 def matches_path(self, path: Path) -> bool: 

635 return matches_path_patterns(path=path, patterns=self.paths) 

636 

637 def matches_code(self, code: str) -> Generator[dict[str, int]]: 

638 patterns = getattr(self, "_code_regex_patterns", []) 

639 if not patterns: 

640 patterns = [re.compile(pattern, re.MULTILINE) for pattern in self.code] 

641 self._code_regex_patterns = patterns 

642 

643 for pattern in patterns: 

644 for match in pattern.finditer(code): 

645 start_index = match.start() 

646 end_index = match.end() 

647 

648 start_line = code.count("\n", 0, start_index) + 1 

649 start_col = start_index - code.rfind("\n", 0, start_index) 

650 

651 end_line = code.count("\n", 0, end_index) + 1 

652 end_col = end_index - code.rfind("\n", 0, end_index) 

653 

654 yield { 

655 "start_line": start_line, 

656 "start_col": start_col, 

657 "end_line": end_line, 

658 "end_col": end_col, 

659 } 

660 

661 def matches_author(self, author_username: str) -> bool: 

662 if not self.authors: 

663 # No authors specified, so assume it matches 

664 return True 

665 

666 author_username_lower = author_username.lower() 

667 

668 negated_authors = [a[1:].lower() for a in self.authors if a.startswith("!")] 

669 authors = [a.lower() for a in self.authors if not a.startswith("!")] 

670 

671 if author_username_lower in negated_authors: 

672 # If the author is in the negated list, return False 

673 return False 

674 

675 if not authors: 

676 # Negation-only: everyone not negated matches 

677 return True 

678 

679 return author_username_lower in authors 

680 

681 

682class LargeScaleChangeModel(BaseModel): 

683 model_config = ConfigDict(extra="forbid") 

684 

685 # Note, an LSC only applies to diffs, not raw files, 

686 # because we have to know what *changed*. 

687 

688 # Pretty similar to a scope, but more manual. 

689 # There has to be at least one reviewer. So if a LSC config is not defined, an LSC PR error until you add one. 

690 # Negative values are rejected at compile time (_validate_review_counts) 

691 require: int = 1 

692 reviewers: list[str] = [] # Field(min_length=1) 

693 # min_paths: int = 300 

694 # min_lines: int = 3000 

695 labels: list[str] = [] 

696 # really need author value too...? 

697 

698 @field_validator("reviewers", mode="after") 

699 @classmethod 

700 def validate_team_ref_shape(cls, values: list[str]) -> list[str]: 

701 return _validate_team_refs(values) 

702 

703 def unsolvable_reason(self, author_username: str) -> str | None: 

704 """ 

705 Explain why this LSC can never pass for a PR authored by this user, 

706 or None if it can. 

707 

708 Messages must stay under ~130 chars: they flow into the git-host 

709 commit status description, which the adapters slice to 140. 

710 """ 

711 if "*" in self.reviewers: 

712 # Anyone can review, so any require is satisfiable 

713 return None 

714 

715 if not self.reviewers: 

716 # An empty roster is "configuration required", which 

717 # process_large_scale_change reports on its own 

718 return None 

719 

720 # Count eligible reviewers (excluding author who can't self-approve) 

721 eligible_reviewers = {r.lower() for r in self.reviewers} - { 

722 author_username.lower() 

723 } 

724 if self.require > 0 and len(eligible_reviewers) < self.require: 

725 if not eligible_reviewers: 

726 return "the PR author is the only reviewer and cannot self-approve" 

727 return f"require={self.require} but only {len(eligible_reviewers)} possible approvals (excluding author)" 

728 

729 return None 

730 

731 

732class ConfigModel(BaseModel): 

733 model_config = ConfigDict(extra="forbid") 

734 

735 # Nothing is technically required 

736 extends: list[str] = [] 

737 template: bool = False 

738 aliases: dict[str, list[str]] = {} 

739 large_scale_change: LargeScaleChangeModel | None = None 

740 scopes: list[ScopeModel] = [] 

741 

742 @field_validator("scopes", mode="after") 

743 @classmethod 

744 def validate_unique_scope_names(cls, scopes: list[ScopeModel]) -> list[ScopeModel]: 

745 if dup := _first_case_insensitive_duplicate(scope.name for scope in scopes): 

746 raise ValueError(f"Duplicate scope name: {dup}") 

747 return scopes 

748 

749 @field_validator("extends", mode="before") 

750 @classmethod 

751 def validate_extends(cls, extends: list[str]) -> list[str]: 

752 for i, path in enumerate(extends): 

753 basename = Path(path).name 

754 if not basename.startswith(CONFIG_FILENAME_PREFIX): 

755 raise ValueError( 

756 f"Invalid extends path: {path}. It should start with '{CONFIG_FILENAME_PREFIX}'." 

757 ) 

758 return extends 

759 

760 def compiled_config( 

761 self, 

762 config_path: Path, 

763 other_configs: ConfigModels, 

764 teams: dict[str, list[str]] | None = None, 

765 ) -> ConfigModel: 

766 """ 

767 Resolve `extends` and replace aliases, returning the effective config. 

768 

769 Two phases: flatten the whole extends chain into one merged (raw, 

770 unexpanded) config, then expand aliases once. Expanding after the full 

771 merge is what makes transitive inheritance and cross-chain alias 

772 scoping work — an alias defined anywhere in the chain resolves anywhere. 

773 

774 `teams` maps team refs (any case, without the leading `@`) to member 

775 usernames, and is only consulted for user-list fields (reviewers, 

776 alternates, authors, cc, large_scale_change.reviewers). Keys are 

777 lowercased here, so callers don't need to normalize case themselves. 

778 With `teams=None` (the default), `@org/team` references in those 

779 fields pass through unexpanded — the offline/CLI mode, since this 

780 library never calls out to GitHub/GitLab itself. That 

781 partially-resolved config is only sanctioned for offline use; 

782 anything that needs real reviewer usernames must pass a `teams` 

783 mapping. 

784 

785 Pure function of the raw `self` and `other_configs` (it never reads its 

786 own anchored output), so it is safe to call uncached. 

787 """ 

788 

789 if teams is not None: 

790 teams = {key.lower(): members for key, members in teams.items()} 

791 

792 compiled_data = self._merged_data(config_path, other_configs) 

793 

794 # Expand aliases for any aliasable list fields. Team refs only expand 

795 # in user-list fields — paths/code/labels keep a leading "@" literal. 

796 for scope in compiled_data["scopes"]: 

797 for field in [ 

798 "paths", 

799 "code", 

800 "authors", 

801 "reviewers", 

802 "alternates", 

803 "cc", 

804 "labels", 

805 ]: 

806 if field in scope: 

807 scope[field] = _expand_aliases( 

808 scope[field], 

809 compiled_data["aliases"], 

810 teams=teams, 

811 expand_teams=field in USER_LIST_FIELDS, 

812 ) 

813 

814 # Apply compile-time "!" subtraction to roster fields. 

815 # `_apply_negations` leaves a field with an unexpanded team ref 

816 # (offline compile, i.e. teams=None) untouched. 

817 for scope in compiled_data["scopes"]: 

818 for field in ROSTER_FIELDS: 

819 if field in scope: 

820 scope[field] = _apply_negations(scope[field]) 

821 _reject_bot_reviewers( 

822 scope[field], 

823 where=f"Scope '{scope['name']}' {field}", 

824 ) 

825 

826 # The "*" wildcard is only meaningful in `reviewers` — everywhere 

827 # else it's matched as a literal username and silently does 

828 # nothing: in `alternates` the scope stays pending forever, in 

829 # `authors` the scope never applies to any PR, in `cc` nobody is 

830 # notified. Reject it at compile time (after alias expansion, so 

831 # `$alias` indirection can't smuggle it in) rather than as a 

832 # ScopeModel validator, because compiled configs stored inside 

833 # old processing results must keep parsing. 

834 # Messages must stay under ~130 chars: they become the git-host 

835 # commit status description, which the adapters slice to 140. 

836 for field, hint in ( 

837 ( 

838 "authors", 

839 "remove it (a scope with no authors applies to any author)", 

840 ), 

841 ( 

842 "alternates", 

843 "add it to reviewers instead (wildcard reviewers are never auto-requested)", 

844 ), 

845 ("cc", "remove it"), 

846 ): 

847 if "*" in scope.get(field, []): 

848 raise ValueError( 

849 f"Scope '{scope['name']}': \"*\" is not supported in " 

850 f"{field} — {hint}" 

851 ) 

852 

853 _validate_review_counts(f"Scope '{scope['name']}'", scope) 

854 

855 if large_scale_change := compiled_data.get("large_scale_change"): 

856 large_scale_change["reviewers"] = _expand_aliases( 

857 large_scale_change["reviewers"], 

858 compiled_data["aliases"], 

859 teams=teams, 

860 expand_teams=True, 

861 ) 

862 large_scale_change["labels"] = _expand_aliases( 

863 large_scale_change["labels"], 

864 compiled_data["aliases"], 

865 ) 

866 large_scale_change["reviewers"] = _apply_negations( 

867 large_scale_change["reviewers"] 

868 ) 

869 _reject_bot_reviewers( 

870 large_scale_change["reviewers"], 

871 where="large_scale_change reviewers", 

872 ) 

873 _validate_review_counts("large_scale_change", large_scale_change) 

874 

875 # Anchor each scope's paths at the directory tagged during flattening 

876 # (after alias expansion, so any `$path-alias` is resolved first). The 

877 # transient tag is popped so it never reaches the model. 

878 # Paths are relative to the config that owns the scope, `/` for 

879 # repo-root-absolute. 

880 for entry in compiled_data["scopes"]: 

881 anchor_dir = entry.pop("_anchor_dir", "") 

882 entry["paths"] = [_anchor_path(anchor_dir, p) for p in entry["paths"]] 

883 

884 # The compiled config is the self-contained effective config: extends 

885 # are already merged in and aliases already expanded, so drop both. This 

886 # keeps stored results lean and makes the compiled form standalone (it 

887 # can never dangle on a missing extends target or re-expand differently). 

888 compiled_data["extends"] = [] 

889 compiled_data["aliases"] = {} 

890 

891 return ConfigModel.from_data( 

892 data=compiled_data, 

893 path=config_path, 

894 ) 

895 

896 def _merged_data( 

897 self, 

898 config_path: Path, 

899 other_configs: ConfigModels, 

900 _in_progress: list[str] | None = None, 

901 _seen: set[str] | None = None, 

902 ) -> dict[str, Any]: 

903 """ 

904 Flatten the `extends` chain into one merged, *unexpanded* config dict. 

905 

906 Parents are merged before this config (so a child can specialize), with 

907 aliases unioned child-wins and the large-scale-change config taken from 

908 the child if set else the first parent that defines one. 

909 

910 `_in_progress` is the current ancestor path, used to detect circular 

911 extends. `_seen` is every config already merged into this flatten, used 

912 to merge a shared ancestor only once (diamond dedup). 

913 """ 

914 if _in_progress is None: 

915 _in_progress = [] 

916 if _seen is None: 

917 _seen = set() 

918 

919 config_path_str = str(config_path) 

920 if config_path_str in _in_progress: 

921 cycle = _in_progress[_in_progress.index(config_path_str) :] + [ 

922 config_path_str 

923 ] 

924 raise ValueError( 

925 f"Circular reference detected in extends: {' -> '.join(cycle)}" 

926 ) 

927 _in_progress.append(config_path_str) 

928 

929 inherited_scopes: list[dict[str, Any]] = [] 

930 inherited_aliases: dict[str, list[str]] = {} 

931 inherited_lsc: dict[str, Any] | None = None 

932 

933 for extend_path in self.extends: 

934 resolved_path = _resolve_extends_path(config_path_str, extend_path) 

935 if resolved_path not in other_configs: 

936 raise ValueError( 

937 f"Config not found: '{extend_path}' (resolved to '{resolved_path}')" 

938 ) 

939 if resolved_path in _seen: 

940 # Already merged via another branch (diamond) — skip the dup. 

941 continue 

942 

943 parent_data = other_configs[resolved_path]._merged_data( 

944 Path(resolved_path), other_configs, _in_progress, _seen 

945 ) 

946 inherited_scopes = inherited_scopes + parent_data["scopes"] 

947 inherited_aliases = inherited_aliases | parent_data["aliases"] 

948 inherited_lsc = inherited_lsc or parent_data["large_scale_change"] 

949 

950 merged = self.model_dump() 

951 merged["scopes"] = inherited_scopes + merged["scopes"] 

952 merged["aliases"] = inherited_aliases | merged["aliases"] 

953 merged["large_scale_change"] = merged["large_scale_change"] or inherited_lsc 

954 

955 # Tag each scope with the directory its paths should anchor at. A scope's 

956 # paths are relative to the config that owns it, so the first 

957 # non-template config to consume a scope claims it: a non-template's own 

958 # scopes (and any it inherits from a template) anchor at its directory, 

959 # while a template defers to its consumer. `setdefault` means an 

960 # already-tagged scope (from a non-template ancestor) keeps its anchor. 

961 if not self.template: 

962 base_dir = posixpath.dirname(config_path_str) 

963 for entry in merged["scopes"]: 

964 entry.setdefault("_anchor_dir", base_dir) 

965 

966 _seen.add(config_path_str) 

967 _in_progress.pop() 

968 

969 return merged 

970 

971 @classmethod 

972 def from_filesystem(cls, path: Path | str) -> ConfigModel: 

973 with open(path, "rb") as f: 

974 return cls.from_data(tomllib.load(f), path) 

975 

976 @classmethod 

977 def from_content(cls, content: str, path: Path | str) -> ConfigModel: 

978 return cls.from_data(tomllib.loads(content), path) 

979 

980 @classmethod 

981 def from_data(cls, data: dict[str, Any], path: Path | str) -> ConfigModel: 

982 return cls(**data) 

983 

984 

985class _ConfigModelsBase(RootModel): 

986 """Shared storage and accessors for a set of configs keyed by repo path.""" 

987 

988 root: dict[str, ConfigModel] 

989 

990 @classmethod 

991 def from_config_models(cls, models: dict[str, ConfigModel]) -> Self: 

992 """Build from a dict of already-constructed configs keyed by path.""" 

993 configs = cls(root={}) 

994 for path, config_model in models.items(): 

995 configs.root[str(Path(path))] = config_model 

996 return configs 

997 

998 def get_config_models(self) -> dict[str, ConfigModel]: 

999 return dict(self.root.items()) 

1000 

1001 def __bool__(self) -> bool: 

1002 return bool(self.root) 

1003 

1004 def __getitem__(self, key: str) -> ConfigModel: 

1005 return self.root[key] 

1006 

1007 def __contains__(self, key: str) -> bool: 

1008 return key in self.root 

1009 

1010 def __len__(self) -> int: 

1011 return len(self.root) 

1012 

1013 

1014class ConfigModels(_ConfigModelsBase): 

1015 """Configs exactly as loaded from the repo — extends unresolved, aliases 

1016 unexpanded, paths unanchored. Build the set up, then call `compiled()`.""" 

1017 

1018 def declared_check_names(self) -> set[str]: 

1019 """Every check-run name an effective (non-template) scope's `unless` 

1020 references, across the whole config set. 

1021 

1022 Collected from an offline compile (`teams=None`, the `team_refs` 

1023 precedent) so a template's refs count only where a consumer actually 

1024 inherits them -- an unconsumed template must not make every pull 

1025 request fetch (or error on) checks that nothing evaluates. `unless` 

1026 refs are literal strings (aliases and teams never expand inside 

1027 them), so the offline compile is exact. Empty for the common case (no 

1028 `unless` anywhere), which is what lets the processor skip fetching 

1029 check runs entirely. 

1030 """ 

1031 return { 

1032 name 

1033 for config in self.compiled().root.values() 

1034 if not config.template 

1035 for scope in config.scopes 

1036 for _, name in scope.unless_refs() 

1037 } 

1038 

1039 @classmethod 

1040 def from_configs_data(cls, data: dict[str, Any]) -> ConfigModels: 

1041 """Load configs from a dict of parsed config data keyed by path.""" 

1042 configs = cls(root={}) 

1043 

1044 for path, config_data in data.items(): 

1045 config = ConfigModel.from_data(config_data, Path(path)) 

1046 configs.add_config(config, Path(path)) 

1047 

1048 return configs 

1049 

1050 @classmethod 

1051 def from_contents(cls, contents: dict[str, str]) -> ConfigModels: 

1052 """Load configs from a dict of raw TOML content keyed by path.""" 

1053 configs = cls(root={}) 

1054 

1055 for path, content in contents.items(): 

1056 configs.add_config(ConfigModel.from_content(content, path), Path(path)) 

1057 

1058 return configs 

1059 

1060 def add_config(self, config: ConfigModel, path: Path) -> None: 

1061 self.root[str(path)] = config 

1062 

1063 def team_refs(self) -> set[str]: 

1064 """Collect every team ref (lowercase, no leading `@`/`!`) that 

1065 `compiled(teams=...)` would actually try to expand: refs written 

1066 directly in a user-list field (scopes' `USER_LIST_FIELDS` and 

1067 `large_scale_change.reviewers`), plus any refs reachable from those 

1068 fields through `$alias`/`!$alias` chains. 

1069 

1070 Meant for callers that need to know which teams to fetch/sync before 

1071 calling `compiled(teams=...)`. 

1072 

1073 Implemented as an offline compile (`teams=None`): aliases expand but 

1074 team refs pass through unexpanded, so whatever refs remain in the 

1075 compiled user-list fields are — by construction — exactly the refs a 

1076 real compile will try to expand. A ref that only appears in a 

1077 non-user-list field (e.g. an npm-style `@vendor/pkg/**` in `paths`) 

1078 or inside an alias nothing references never survives into a compiled 

1079 user-list field, so it is never collected. Raises the same config 

1080 errors `compiled()` would (unknown alias, circular refs, ...), just 

1081 earlier. 

1082 """ 

1083 # Cheap pre-check: a team ref can only enter a compile as a literal 

1084 # `@`/`!@` value in a user-list field or an alias value. Most repos 

1085 # have none, and skipping the compile keeps this near-free for them. 

1086 candidate_lists: list[list[str]] = [] 

1087 for config in self.root.values(): 

1088 candidate_lists.extend(config.aliases.values()) 

1089 for scope in config.scopes: 

1090 for field in USER_LIST_FIELDS: 

1091 candidate_lists.append(getattr(scope, field)) 

1092 if config.large_scale_change: 

1093 candidate_lists.append(config.large_scale_change.reviewers) 

1094 if not any( 

1095 _split_team_ref(value) is not None 

1096 for values in candidate_lists 

1097 for value in values 

1098 ): 

1099 return set() 

1100 

1101 refs: set[str] = set() 

1102 

1103 def collect_refs(values: list[str]) -> None: 

1104 for value in values: 

1105 if (split := _split_team_ref(value)) is not None: 

1106 refs.add(split[1].lower()) 

1107 

1108 for config in self.compiled(teams=None).get_config_models().values(): 

1109 if config.template: 

1110 continue 

1111 for scope in config.scopes: 

1112 for field in USER_LIST_FIELDS: 

1113 collect_refs(getattr(scope, field)) 

1114 if config.large_scale_change: 

1115 collect_refs(config.large_scale_change.reviewers) 

1116 

1117 return refs 

1118 

1119 def compiled( 

1120 self, teams: dict[str, list[str]] | None = None 

1121 ) -> CompiledConfigModels: 

1122 """Resolve the whole set into its effective, PR-independent form. 

1123 

1124 Each non-template config is compiled once — extends merged, aliases 

1125 expanded, paths anchored. Templates are NOT compiled standalone: a 

1126 template scope may reference an alias the consuming config provides, and 

1127 its paths anchor at the consumer. They are carried through untouched 

1128 (folded into each consumer during that consumer's compile, and kept in 

1129 the set for display). 

1130 

1131 `teams` maps team refs (any case, without the leading `@`) to member 

1132 usernames; passed straight through to each config's `compiled_config` 

1133 (see there for case normalization and the `teams=None` vs provided 

1134 semantics). 

1135 

1136 The result is an immutable `CompiledConfigModels` — there is no way to 

1137 compile it again, so the non-idempotent path anchoring can never 

1138 double-apply. 

1139 """ 

1140 effective: dict[str, ConfigModel] = {} 

1141 for path, config in self.root.items(): 

1142 if config.template: 

1143 effective[path] = config 

1144 else: 

1145 effective[path] = config.compiled_config( 

1146 config_path=Path(path), other_configs=self, teams=teams 

1147 ) 

1148 

1149 return CompiledConfigModels.from_config_models(effective) 

1150 

1151 

1152class CompiledConfigModels(_ConfigModelsBase): 

1153 """The effective configs used for matching: every non-template config is 

1154 fully resolved. Produced by `ConfigModels.compiled()`; never recompiled.""" 

1155 

1156 def closest_config_path(self, file_path: Path) -> str | None: 

1157 """The path of the closest non-template config governing this file, 

1158 or None when nothing governs it.""" 

1159 for parent in file_path.parents: 

1160 parent_config_path = str(parent / CONFIG_FILENAME) 

1161 config = self.root.get(parent_config_path) 

1162 if config is not None and not config.template: 

1163 return parent_config_path 

1164 return None 

1165 

1166 def closest_config(self, file_path: Path) -> ConfigModel: 

1167 """Return the closest non-template config governing this file.""" 

1168 config_path = self.closest_config_path(file_path) 

1169 if config_path is None: 

1170 raise ValueError(f"No config found for {file_path}") 

1171 return self.root[config_path] 

1172 

1173 def get_default_large_scale_change(self) -> LargeScaleChangeModel: 

1174 """The primary (repo-root) config's large-scale-change section, if any. 

1175 

1176 The primary was compiled by `compiled()`, so its reviewers/labels are 

1177 already alias-expanded (e.g. ["$backend"] -> usernames). A `template = 

1178 true` repo root is a misconfiguration (templates are meant to be 

1179 extended, not be the primary); it is passed through uncompiled, so its 

1180 LSC would read with aliases unexpanded. 

1181 """ 

1182 if CONFIG_FILENAME in self.root and ( 

1183 lsc := self.root[CONFIG_FILENAME].large_scale_change 

1184 ): 

1185 return lsc 

1186 

1187 return LargeScaleChangeModel() 

1188 

1189 def filter_for_pullrequest(self, author_username: str) -> CompiledConfigModels: 

1190 """ 

1191 Overlay PR-dependent scope gating: drop scopes that author rules disable 

1192 for this pull request. 

1193 

1194 This is the only PR-dependent step. The configs are already compiled, so 

1195 each config's scopes are self-contained and dropping one is a plain list 

1196 filter — no re-inheritance. Templates are passed through (they are never 

1197 matched directly; their scopes already live in each consumer). 

1198 """ 

1199 effective: dict[str, ConfigModel] = {} 

1200 for config_path, config in self.root.items(): 

1201 if config.template: 

1202 # Templates are never matched directly; pass them through. 

1203 effective[config_path] = config 

1204 continue 

1205 

1206 kept_scopes = [ 

1207 scope 

1208 for scope in config.scopes 

1209 if scope.matches_author(author_username) 

1210 ] 

1211 effective[config_path] = config.model_copy(update={"scopes": kept_scopes}) 

1212 

1213 return CompiledConfigModels.from_config_models(effective)