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

483 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-07-28 18:11 -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: 

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: 

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 _anchor_path(base_dir: str, pattern: str) -> str: 

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

349 

350 Scope paths are written relative to the config they live in. A leading `/` 

351 makes a pattern repo-root-absolute (escape hatch); a leading `!` negation is 

352 preserved. With an empty `base_dir` (root config) the pattern is unchanged. 

353 """ 

354 negate = pattern.startswith("!") 

355 if negate: 

356 pattern = pattern[1:] 

357 

358 if pattern.startswith("/"): 

359 anchored = pattern.lstrip("/") 

360 elif base_dir: 

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

362 else: 

363 anchored = pattern 

364 

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

366 

367 

368class OwnershipChoices(StrEnum): 

369 EMPTY = "" 

370 APPEND = "append" 

371 GLOBAL = "global" 

372 

373 

374# A GitHub App has two names: the login it posts as (`name[bot]`) and the slug 

375# that owns its check runs (`name`). The suffix is how the engine tells a bot 

376# from a person -- a `[bot]` account never counts as a human and never sits in 

377# a roster, no declaration needed. 

378BOT_LOGIN_SUFFIX = "[bot]" 

379 

380 

381def is_bot_login(username: str) -> bool: 

382 """The engine rule, in one place: a `[bot]` account is never a person.""" 

383 return username.lower().endswith(BOT_LOGIN_SUFFIX) 

384 

385 

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

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

388 

389 Split on the FIRST `/`: App slugs cannot contain one, check names can. 

390 Both parts are stripped -- the validator checks THIS function's output, so 

391 a ref that validates is exactly a ref that matches at runtime. 

392 """ 

393 producer, _, name = ref.partition("/") 

394 return producer.strip().lower(), name.strip() 

395 

396 

397def _reject_bot_reviewers(values: list[str], *, where: str) -> None: 

398 """Reject any `[bot]` account listed in a roster surface (a scope's 

399 reviewers/alternates/cc, or large_scale_change.reviewers). 

400 

401 A bot is never a person, so it can never sit in a roster: bots that open 

402 pull requests are routed with `authors`, and bots that attest are 

403 referenced from `unless`. Checked after alias expansion so `$alias` 

404 indirection can't smuggle one in. `where` names the surface; the message 

405 becomes the git-host commit status, so it stays under ~130 chars. 

406 """ 

407 for entry in values: 

408 if is_bot_login(entry): 

409 raise ValueError( 

410 f"{where}: '{entry}' is a bot and cannot be listed as a reviewer" 

411 ) 

412 

413 

414def _first_case_insensitive_duplicate(values: Iterable[str]) -> str | None: 

415 """The first value whose lowercased form was already seen, else None.""" 

416 seen: set[str] = set() 

417 for value in values: 

418 if value.lower() in seen: 

419 return value 

420 seen.add(value.lower()) 

421 return None 

422 

423 

424class ScopeModel(BaseModel): 

425 model_config = ConfigDict(extra="forbid") 

426 

427 # Required fields 

428 name: str = Field(min_length=1) 

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

430 

431 # Optional fields 

432 

433 # Expanded version of lines could be dict 

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

435 code: list[str] = [] 

436 

437 # This only filtering field that can't be used with raw diff/files... 

438 # If we get into that, the others are: 

439 # - labels 

440 # - ref 

441 # - statuses 

442 # - dates 

443 # - body 

444 # - title 

445 # - other scopes 

446 # (this is how I ended up with expressions... 

447 # I'm not trying to build a general purpose workflow tool, 

448 # but I do need to support the legit use cases and AI/bot review is one, so is team hierarchy) 

449 authors: list[str] = [] 

450 

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

452 description: str = "" 

453 reviewers: list[str] = [] 

454 alternates: list[str] = [] 

455 cc: list[str] = [] 

456 

457 # Review scoring 

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

459 require: int = 0 

460 author_value: int = 0 

461 

462 # The checks that can attest this scope's review requirement away. Each ref 

463 # is "producer/check-name" -- the slug of the App that produces the check, 

464 # then the check run's name, split on the first "/". When every listed 

465 # check has completed successfully on the current head commit, the scope is 

466 # waived: its requirement drops to zero and the evidence is stored on the 

467 # result. Failed, skipped, pending, missing -- the requirement stands 

468 # unchanged; passing is the only state that subtracts. 

469 # 

470 # Deliberately placed next to the `require` it undermines, so reading a 

471 # scope always reveals whether its human review is removable. A list is 

472 # implicitly "all must pass" -- OR, thresholds, and 2-of-3 belong in the 

473 # producer, which can post one combined check. 

474 unless: list[str] = [] 

475 

476 # How scopes are combined 

477 ownership: OwnershipChoices = OwnershipChoices.EMPTY 

478 

479 # Actionable items 

480 request: int = 0 

481 labels: list[str] = [] 

482 instructions: str = "" 

483 

484 # Approval checklist 

485 checklist: Checklist | None = None 

486 

487 @field_validator("name", mode="after") 

488 @classmethod 

489 def validate_name(cls, name: str) -> str: 

490 if "," in name: 

491 raise ValueError("Scope name cannot contain commas") 

492 return name 

493 

494 @field_validator(*USER_LIST_FIELDS, mode="after") 

495 @classmethod 

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

497 return _validate_team_refs(values) 

498 

499 @field_validator("unless", mode="after") 

500 @classmethod 

501 def validate_unless_refs(cls, values: list[str]) -> list[str]: 

502 """A check name alone is worthless -- any workflow with `checks: write` 

503 can post any name -- so a bare name is a parse error, not a default. 

504 Messages stay under ~130 chars: they become the git-host commit status. 

505 """ 

506 for ref in values: 

507 producer, name = _split_check_ref(ref) 

508 if "/" not in ref or not producer or not name: 

509 raise ValueError( 

510 f"unless: '{ref}' must be 'producer/check-name' -- the App " 

511 "slug that produces the check, then the check's name" 

512 ) 

513 # To most users PullApprove *is* "the check" on their PRs. 

514 # PullApprove never creates these checks, it only reads them -- and 

515 # a config waiting on our own status would deadlock politely. 

516 if producer == "pullapprove": 

517 raise ValueError( 

518 f"unless: '{ref}' references PullApprove itself -- " 

519 "PullApprove never creates checks, it only reads them" 

520 ) 

521 return values 

522 

523 @field_validator("code", mode="after") 

524 @classmethod 

525 def validate_code_patterns(cls, code: list[str]) -> list[str]: 

526 for pattern in code: 

527 try: 

528 parsed = sre_parse.parse(pattern) 

529 except re.error as e: 

530 raise ValueError(f"Invalid regex pattern '{pattern}': {e}") from None 

531 if _has_nested_quantifiers(parsed): 

532 raise ValueError( 

533 f"Regex pattern '{pattern}' contains nested quantifiers, " 

534 "which can cause catastrophic backtracking." 

535 ) 

536 return code 

537 

538 @model_validator(mode="after") 

539 def validate_reviewers_for_require(self) -> ScopeModel: 

540 all_reviewers = self.reviewers + self.alternates 

541 

542 # Skip if wildcard - anyone can review 

543 if "*" in all_reviewers: 

544 return self 

545 

546 # Skip if aliases or team refs (possibly negated with "!") are not yet 

547 # expanded. Re-validation after compilation only re-checks refs that 

548 # actually resolve — an offline compile (teams=None) leaves refs 

549 # unexpanded, so this same skip fires again there too. 

550 if any(is_unexpanded_ref(r) for r in all_reviewers): 

551 return self 

552 

553 if len(all_reviewers) < self.require: 

554 raise ValueError( 

555 f"has require={self.require} but only {len(all_reviewers)} reviewers/alternates specified" 

556 ) 

557 return self 

558 

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

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

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

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

563 return self.author_value 

564 return 0 

565 

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

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

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

569 

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

571 """ 

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

573 or None if it can. 

574 

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

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

577 """ 

578 if "*" in self.reviewers: 

579 # Anyone can review, so any require is satisfiable 

580 return None 

581 

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

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

584 author_username.lower() 

585 } 

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

587 author_username 

588 ) 

589 

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

591 if not eligible_reviewers: 

592 return ( 

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

594 ) 

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

596 

597 return None 

598 

599 def printed_name(self) -> str: 

600 match self.ownership: 

601 case OwnershipChoices.APPEND: 

602 return "+" + self.name 

603 case OwnershipChoices.GLOBAL: 

604 return "*" + self.name 

605 

606 return self.name 

607 

608 def __eq__(self, other: Any) -> bool: 

609 return self.name == other.name 

610 

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

612 # TODO paths shouldn't start with / 

613 return glob.globmatch( 

614 path, 

615 self.paths, 

616 flags=glob.GLOBSTAR 

617 | glob.BRACE 

618 | glob.NEGATE 

619 | glob.IGNORECASE 

620 | glob.DOTGLOB, 

621 ) 

622 

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

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

625 if not patterns: 

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

627 self._code_regex_patterns = patterns 

628 

629 for pattern in patterns: 

630 for match in pattern.finditer(code): 

631 start_index = match.start() 

632 end_index = match.end() 

633 

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

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

636 

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

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

639 

640 yield { 

641 "start_line": start_line, 

642 "start_col": start_col, 

643 "end_line": end_line, 

644 "end_col": end_col, 

645 } 

646 

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

648 if not self.authors: 

649 # No authors specified, so assume it matches 

650 return True 

651 

652 author_username_lower = author_username.lower() 

653 

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

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

656 

657 if author_username_lower in negated_authors: 

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

659 return False 

660 

661 if not authors: 

662 # Negation-only: everyone not negated matches 

663 return True 

664 

665 if author_username_lower in authors: 

666 # If the author is in the authors list, return True 

667 return True 

668 

669 return False 

670 

671 

672class LargeScaleChangeModel(BaseModel): 

673 model_config = ConfigDict(extra="forbid") 

674 

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

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

677 

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

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

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

681 require: int = 1 

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

683 # min_paths: int = 300 

684 # min_lines: int = 3000 

685 labels: list[str] = [] 

686 # really need author value too...? 

687 

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

689 @classmethod 

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

691 return _validate_team_refs(values) 

692 

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

694 """ 

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

696 or None if it can. 

697 

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

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

700 """ 

701 if "*" in self.reviewers: 

702 # Anyone can review, so any require is satisfiable 

703 return None 

704 

705 if not self.reviewers: 

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

707 # process_large_scale_change reports on its own 

708 return None 

709 

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

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

712 author_username.lower() 

713 } 

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

715 if not eligible_reviewers: 

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

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

718 

719 return None 

720 

721 

722class ConfigModel(BaseModel): 

723 model_config = ConfigDict(extra="forbid") 

724 

725 # Nothing is technically required 

726 extends: list[str] = [] 

727 template: bool = False 

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

729 large_scale_change: LargeScaleChangeModel | None = None 

730 scopes: list[ScopeModel] = [] 

731 

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

733 @classmethod 

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

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

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

737 return scopes 

738 

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

740 @classmethod 

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

742 for i, path in enumerate(extends): 

743 basename = Path(path).name 

744 if not basename.startswith(CONFIG_FILENAME_PREFIX): 

745 raise ValueError( 

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

747 ) 

748 return extends 

749 

750 def compiled_config( 

751 self, 

752 config_path: Path, 

753 other_configs: ConfigModels, 

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

755 ) -> ConfigModel: 

756 """ 

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

758 

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

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

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

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

763 

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

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

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

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

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

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

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

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

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

773 mapping. 

774 

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

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

777 """ 

778 

779 if teams is not None: 

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

781 

782 compiled_data = self._merged_data(config_path, other_configs) 

783 

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

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

786 for scope in compiled_data["scopes"]: 

787 for field in [ 

788 "paths", 

789 "code", 

790 "authors", 

791 "reviewers", 

792 "alternates", 

793 "cc", 

794 "labels", 

795 ]: 

796 if field in scope: 

797 scope[field] = _expand_aliases( 

798 scope[field], 

799 compiled_data["aliases"], 

800 teams=teams, 

801 expand_teams=field in USER_LIST_FIELDS, 

802 ) 

803 

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

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

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

807 for scope in compiled_data["scopes"]: 

808 for field in ROSTER_FIELDS: 

809 if field in scope: 

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

811 _reject_bot_reviewers( 

812 scope[field], 

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

814 ) 

815 

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

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

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

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

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

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

822 # ScopeModel validator, because compiled configs stored inside 

823 # old processing results must keep parsing. 

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

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

826 for field, hint in ( 

827 ( 

828 "authors", 

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

830 ), 

831 ( 

832 "alternates", 

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

834 ), 

835 ("cc", "remove it"), 

836 ): 

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

838 raise ValueError( 

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

840 f"{field}{hint}" 

841 ) 

842 

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

844 

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

846 large_scale_change["reviewers"] = _expand_aliases( 

847 large_scale_change["reviewers"], 

848 compiled_data["aliases"], 

849 teams=teams, 

850 expand_teams=True, 

851 ) 

852 large_scale_change["labels"] = _expand_aliases( 

853 large_scale_change["labels"], 

854 compiled_data["aliases"], 

855 ) 

856 large_scale_change["reviewers"] = _apply_negations( 

857 large_scale_change["reviewers"] 

858 ) 

859 _reject_bot_reviewers( 

860 large_scale_change["reviewers"], 

861 where="large_scale_change reviewers", 

862 ) 

863 _validate_review_counts("large_scale_change", large_scale_change) 

864 

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

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

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

868 for scope in compiled_data["scopes"]: 

869 anchor_dir = scope.pop("_anchor_dir", "") 

870 scope["paths"] = [_anchor_path(anchor_dir, p) for p in scope["paths"]] 

871 

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

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

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

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

876 compiled_data["extends"] = [] 

877 compiled_data["aliases"] = {} 

878 

879 return ConfigModel.from_data( 

880 data=compiled_data, 

881 path=config_path, 

882 ) 

883 

884 def _merged_data( 

885 self, 

886 config_path: Path, 

887 other_configs: ConfigModels, 

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

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

890 ) -> dict[str, Any]: 

891 """ 

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

893 

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

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

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

897 

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

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

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

901 """ 

902 if _in_progress is None: 

903 _in_progress = [] 

904 if _seen is None: 

905 _seen = set() 

906 

907 config_path_str = str(config_path) 

908 if config_path_str in _in_progress: 

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

910 config_path_str 

911 ] 

912 raise ValueError( 

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

914 ) 

915 _in_progress.append(config_path_str) 

916 

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

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

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

920 

921 for extend_path in self.extends: 

922 resolved_path = _resolve_extends_path(config_path_str, extend_path) 

923 if resolved_path not in other_configs: 

924 raise ValueError( 

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

926 ) 

927 if resolved_path in _seen: 

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

929 continue 

930 

931 parent_data = other_configs[resolved_path]._merged_data( 

932 Path(resolved_path), other_configs, _in_progress, _seen 

933 ) 

934 inherited_scopes = inherited_scopes + parent_data["scopes"] 

935 inherited_aliases = inherited_aliases | parent_data["aliases"] 

936 inherited_lsc = inherited_lsc or parent_data["large_scale_change"] 

937 

938 merged = self.model_dump() 

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

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

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

942 

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

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

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

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

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

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

949 if not self.template: 

950 base_dir = posixpath.dirname(config_path_str) 

951 for scope in merged["scopes"]: 

952 scope.setdefault("_anchor_dir", base_dir) 

953 

954 _seen.add(config_path_str) 

955 _in_progress.pop() 

956 

957 return merged 

958 

959 @classmethod 

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

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

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

963 

964 @classmethod 

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

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

967 

968 @classmethod 

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

970 return cls(**data) 

971 

972 

973class _ConfigModelsBase(RootModel): 

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

975 

976 root: dict[str, ConfigModel] 

977 

978 @classmethod 

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

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

981 configs = cls(root={}) 

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

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

984 return configs 

985 

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

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

988 

989 def __bool__(self) -> bool: 

990 return bool(self.root) 

991 

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

993 return self.root[key] 

994 

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

996 return key in self.root 

997 

998 def __len__(self) -> int: 

999 return len(self.root) 

1000 

1001 

1002class ConfigModels(_ConfigModelsBase): 

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

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

1005 

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

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

1008 references, across the whole config set. 

1009 

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

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

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

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

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

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

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

1017 check runs entirely. 

1018 """ 

1019 return { 

1020 name 

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

1022 if not config.template 

1023 for scope in config.scopes 

1024 for _, name in scope.unless_refs() 

1025 } 

1026 

1027 @classmethod 

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

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

1030 configs = cls(root={}) 

1031 

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

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

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

1035 

1036 return configs 

1037 

1038 @classmethod 

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

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

1041 configs = cls(root={}) 

1042 

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

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

1045 

1046 return configs 

1047 

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

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

1050 

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

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

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

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

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

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

1057 

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

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

1060 

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

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

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

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

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

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

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

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

1069 earlier. 

1070 """ 

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

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

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

1074 candidate_lists: list[list[str]] = [] 

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

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

1077 for scope in config.scopes: 

1078 for field in USER_LIST_FIELDS: 

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

1080 if config.large_scale_change: 

1081 candidate_lists.append(config.large_scale_change.reviewers) 

1082 if not any( 

1083 _split_team_ref(value) is not None 

1084 for values in candidate_lists 

1085 for value in values 

1086 ): 

1087 return set() 

1088 

1089 refs: set[str] = set() 

1090 

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

1092 for value in values: 

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

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

1095 

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

1097 if config.template: 

1098 continue 

1099 for scope in config.scopes: 

1100 for field in USER_LIST_FIELDS: 

1101 collect_refs(getattr(scope, field)) 

1102 if config.large_scale_change: 

1103 collect_refs(config.large_scale_change.reviewers) 

1104 

1105 return refs 

1106 

1107 def compiled( 

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

1109 ) -> CompiledConfigModels: 

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

1111 

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

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

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

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

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

1117 the set for display). 

1118 

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

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

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

1122 semantics). 

1123 

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

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

1126 double-apply. 

1127 """ 

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

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

1130 if config.template: 

1131 effective[path] = config 

1132 else: 

1133 effective[path] = config.compiled_config( 

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

1135 ) 

1136 

1137 return CompiledConfigModels.from_config_models(effective) 

1138 

1139 

1140class CompiledConfigModels(_ConfigModelsBase): 

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

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

1143 

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

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

1146 for parent in file_path.parents: 

1147 parent_config_path = str(parent / CONFIG_FILENAME) 

1148 

1149 if parent_config_path in self.root: 

1150 config = self.root[parent_config_path] 

1151 

1152 if config.template: 

1153 # Skip templates 

1154 continue 

1155 

1156 return config 

1157 

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

1159 

1160 def get_default_large_scale_change(self) -> LargeScaleChangeModel: 

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

1162 

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

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

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

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

1167 LSC would read with aliases unexpanded. 

1168 """ 

1169 if CONFIG_FILENAME in self.root: 

1170 if lsc := self.root[CONFIG_FILENAME].large_scale_change: 

1171 return lsc 

1172 

1173 return LargeScaleChangeModel() 

1174 

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

1176 """ 

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

1178 for this pull request. 

1179 

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

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

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

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

1184 """ 

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

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

1187 if config.template: 

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

1189 effective[config_path] = config 

1190 continue 

1191 

1192 kept_scopes = [ 

1193 scope 

1194 for scope in config.scopes 

1195 if scope.matches_author(author_username) 

1196 ] 

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

1198 

1199 return CompiledConfigModels.from_config_models(effective)