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

485 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-07-28 20:18 -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 ownership_marker(self) -> str: 

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

601 

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

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

604 """ 

605 match self.ownership: 

606 case OwnershipChoices.APPEND: 

607 return "+" 

608 case OwnershipChoices.GLOBAL: 

609 return "*" 

610 

611 return "" 

612 

613 def printed_name(self) -> str: 

614 return self.ownership_marker() + self.name 

615 

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

617 return self.name == other.name 

618 

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

620 # TODO paths shouldn't start with / 

621 return glob.globmatch( 

622 path, 

623 self.paths, 

624 flags=glob.GLOBSTAR 

625 | glob.BRACE 

626 | glob.NEGATE 

627 | glob.IGNORECASE 

628 | glob.DOTGLOB, 

629 ) 

630 

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

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

633 if not patterns: 

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

635 self._code_regex_patterns = patterns 

636 

637 for pattern in patterns: 

638 for match in pattern.finditer(code): 

639 start_index = match.start() 

640 end_index = match.end() 

641 

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

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

644 

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

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

647 

648 yield { 

649 "start_line": start_line, 

650 "start_col": start_col, 

651 "end_line": end_line, 

652 "end_col": end_col, 

653 } 

654 

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

656 if not self.authors: 

657 # No authors specified, so assume it matches 

658 return True 

659 

660 author_username_lower = author_username.lower() 

661 

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

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

664 

665 if author_username_lower in negated_authors: 

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

667 return False 

668 

669 if not authors: 

670 # Negation-only: everyone not negated matches 

671 return True 

672 

673 if author_username_lower in authors: 

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

675 return True 

676 

677 return False 

678 

679 

680class LargeScaleChangeModel(BaseModel): 

681 model_config = ConfigDict(extra="forbid") 

682 

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

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

685 

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

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

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

689 require: int = 1 

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

691 # min_paths: int = 300 

692 # min_lines: int = 3000 

693 labels: list[str] = [] 

694 # really need author value too...? 

695 

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

697 @classmethod 

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

699 return _validate_team_refs(values) 

700 

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

702 """ 

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

704 or None if it can. 

705 

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

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

708 """ 

709 if "*" in self.reviewers: 

710 # Anyone can review, so any require is satisfiable 

711 return None 

712 

713 if not self.reviewers: 

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

715 # process_large_scale_change reports on its own 

716 return None 

717 

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

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

720 author_username.lower() 

721 } 

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

723 if not eligible_reviewers: 

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

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

726 

727 return None 

728 

729 

730class ConfigModel(BaseModel): 

731 model_config = ConfigDict(extra="forbid") 

732 

733 # Nothing is technically required 

734 extends: list[str] = [] 

735 template: bool = False 

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

737 large_scale_change: LargeScaleChangeModel | None = None 

738 scopes: list[ScopeModel] = [] 

739 

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

741 @classmethod 

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

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

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

745 return scopes 

746 

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

748 @classmethod 

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

750 for i, path in enumerate(extends): 

751 basename = Path(path).name 

752 if not basename.startswith(CONFIG_FILENAME_PREFIX): 

753 raise ValueError( 

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

755 ) 

756 return extends 

757 

758 def compiled_config( 

759 self, 

760 config_path: Path, 

761 other_configs: ConfigModels, 

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

763 ) -> ConfigModel: 

764 """ 

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

766 

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

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

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

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

771 

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

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

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

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

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

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

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

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

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

781 mapping. 

782 

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

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

785 """ 

786 

787 if teams is not None: 

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

789 

790 compiled_data = self._merged_data(config_path, other_configs) 

791 

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

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

794 for scope in compiled_data["scopes"]: 

795 for field in [ 

796 "paths", 

797 "code", 

798 "authors", 

799 "reviewers", 

800 "alternates", 

801 "cc", 

802 "labels", 

803 ]: 

804 if field in scope: 

805 scope[field] = _expand_aliases( 

806 scope[field], 

807 compiled_data["aliases"], 

808 teams=teams, 

809 expand_teams=field in USER_LIST_FIELDS, 

810 ) 

811 

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

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

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

815 for scope in compiled_data["scopes"]: 

816 for field in ROSTER_FIELDS: 

817 if field in scope: 

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

819 _reject_bot_reviewers( 

820 scope[field], 

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

822 ) 

823 

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

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

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

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

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

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

830 # ScopeModel validator, because compiled configs stored inside 

831 # old processing results must keep parsing. 

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

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

834 for field, hint in ( 

835 ( 

836 "authors", 

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

838 ), 

839 ( 

840 "alternates", 

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

842 ), 

843 ("cc", "remove it"), 

844 ): 

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

846 raise ValueError( 

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

848 f"{field}{hint}" 

849 ) 

850 

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

852 

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

854 large_scale_change["reviewers"] = _expand_aliases( 

855 large_scale_change["reviewers"], 

856 compiled_data["aliases"], 

857 teams=teams, 

858 expand_teams=True, 

859 ) 

860 large_scale_change["labels"] = _expand_aliases( 

861 large_scale_change["labels"], 

862 compiled_data["aliases"], 

863 ) 

864 large_scale_change["reviewers"] = _apply_negations( 

865 large_scale_change["reviewers"] 

866 ) 

867 _reject_bot_reviewers( 

868 large_scale_change["reviewers"], 

869 where="large_scale_change reviewers", 

870 ) 

871 _validate_review_counts("large_scale_change", large_scale_change) 

872 

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

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

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

876 for scope in compiled_data["scopes"]: 

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

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

879 

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

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

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

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

884 compiled_data["extends"] = [] 

885 compiled_data["aliases"] = {} 

886 

887 return ConfigModel.from_data( 

888 data=compiled_data, 

889 path=config_path, 

890 ) 

891 

892 def _merged_data( 

893 self, 

894 config_path: Path, 

895 other_configs: ConfigModels, 

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

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

898 ) -> dict[str, Any]: 

899 """ 

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

901 

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

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

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

905 

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

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

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

909 """ 

910 if _in_progress is None: 

911 _in_progress = [] 

912 if _seen is None: 

913 _seen = set() 

914 

915 config_path_str = str(config_path) 

916 if config_path_str in _in_progress: 

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

918 config_path_str 

919 ] 

920 raise ValueError( 

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

922 ) 

923 _in_progress.append(config_path_str) 

924 

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

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

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

928 

929 for extend_path in self.extends: 

930 resolved_path = _resolve_extends_path(config_path_str, extend_path) 

931 if resolved_path not in other_configs: 

932 raise ValueError( 

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

934 ) 

935 if resolved_path in _seen: 

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

937 continue 

938 

939 parent_data = other_configs[resolved_path]._merged_data( 

940 Path(resolved_path), other_configs, _in_progress, _seen 

941 ) 

942 inherited_scopes = inherited_scopes + parent_data["scopes"] 

943 inherited_aliases = inherited_aliases | parent_data["aliases"] 

944 inherited_lsc = inherited_lsc or parent_data["large_scale_change"] 

945 

946 merged = self.model_dump() 

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

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

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

950 

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

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

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

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

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

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

957 if not self.template: 

958 base_dir = posixpath.dirname(config_path_str) 

959 for scope in merged["scopes"]: 

960 scope.setdefault("_anchor_dir", base_dir) 

961 

962 _seen.add(config_path_str) 

963 _in_progress.pop() 

964 

965 return merged 

966 

967 @classmethod 

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

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

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

971 

972 @classmethod 

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

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

975 

976 @classmethod 

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

978 return cls(**data) 

979 

980 

981class _ConfigModelsBase(RootModel): 

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

983 

984 root: dict[str, ConfigModel] 

985 

986 @classmethod 

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

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

989 configs = cls(root={}) 

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

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

992 return configs 

993 

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

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

996 

997 def __bool__(self) -> bool: 

998 return bool(self.root) 

999 

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

1001 return self.root[key] 

1002 

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

1004 return key in self.root 

1005 

1006 def __len__(self) -> int: 

1007 return len(self.root) 

1008 

1009 

1010class ConfigModels(_ConfigModelsBase): 

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

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

1013 

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

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

1016 references, across the whole config set. 

1017 

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

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

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

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

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

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

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

1025 check runs entirely. 

1026 """ 

1027 return { 

1028 name 

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

1030 if not config.template 

1031 for scope in config.scopes 

1032 for _, name in scope.unless_refs() 

1033 } 

1034 

1035 @classmethod 

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

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

1038 configs = cls(root={}) 

1039 

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

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

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

1043 

1044 return configs 

1045 

1046 @classmethod 

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

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

1049 configs = cls(root={}) 

1050 

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

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

1053 

1054 return configs 

1055 

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

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

1058 

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

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

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

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

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

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

1065 

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

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

1068 

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

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

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

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

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

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

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

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

1077 earlier. 

1078 """ 

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

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

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

1082 candidate_lists: list[list[str]] = [] 

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

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

1085 for scope in config.scopes: 

1086 for field in USER_LIST_FIELDS: 

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

1088 if config.large_scale_change: 

1089 candidate_lists.append(config.large_scale_change.reviewers) 

1090 if not any( 

1091 _split_team_ref(value) is not None 

1092 for values in candidate_lists 

1093 for value in values 

1094 ): 

1095 return set() 

1096 

1097 refs: set[str] = set() 

1098 

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

1100 for value in values: 

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

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

1103 

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

1105 if config.template: 

1106 continue 

1107 for scope in config.scopes: 

1108 for field in USER_LIST_FIELDS: 

1109 collect_refs(getattr(scope, field)) 

1110 if config.large_scale_change: 

1111 collect_refs(config.large_scale_change.reviewers) 

1112 

1113 return refs 

1114 

1115 def compiled( 

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

1117 ) -> CompiledConfigModels: 

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

1119 

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

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

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

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

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

1125 the set for display). 

1126 

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

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

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

1130 semantics). 

1131 

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

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

1134 double-apply. 

1135 """ 

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

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

1138 if config.template: 

1139 effective[path] = config 

1140 else: 

1141 effective[path] = config.compiled_config( 

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

1143 ) 

1144 

1145 return CompiledConfigModels.from_config_models(effective) 

1146 

1147 

1148class CompiledConfigModels(_ConfigModelsBase): 

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

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

1151 

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

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

1154 for parent in file_path.parents: 

1155 parent_config_path = str(parent / CONFIG_FILENAME) 

1156 

1157 if parent_config_path in self.root: 

1158 config = self.root[parent_config_path] 

1159 

1160 if config.template: 

1161 # Skip templates 

1162 continue 

1163 

1164 return config 

1165 

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

1167 

1168 def get_default_large_scale_change(self) -> LargeScaleChangeModel: 

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

1170 

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

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

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

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

1175 LSC would read with aliases unexpanded. 

1176 """ 

1177 if CONFIG_FILENAME in self.root: 

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

1179 return lsc 

1180 

1181 return LargeScaleChangeModel() 

1182 

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

1184 """ 

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

1186 for this pull request. 

1187 

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

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

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

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

1192 """ 

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

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

1195 if config.template: 

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

1197 effective[config_path] = config 

1198 continue 

1199 

1200 kept_scopes = [ 

1201 scope 

1202 for scope in config.scopes 

1203 if scope.matches_author(author_username) 

1204 ] 

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

1206 

1207 return CompiledConfigModels.from_config_models(effective)