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

1from __future__ import annotations 

2 

3import posixpath 

4import re 

5import tomllib 

6import warnings 

7 

8with warnings.catch_warnings(): 

9 warnings.simplefilter("ignore", DeprecationWarning) 

10 import sre_parse 

11from collections.abc import Generator 

12from enum import StrEnum 

13from pathlib import Path 

14from typing import Any, Self 

15 

16from pydantic import ( 

17 BaseModel, 

18 ConfigDict, 

19 Field, 

20 RootModel, 

21 field_validator, 

22 model_validator, 

23) 

24from wcmatch import glob 

25 

26from .checklists import Checklist 

27 

28CONFIG_FILENAME_PREFIX = "CODEREVIEW" 

29 

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

31 

32 

33def _has_nested_quantifiers(data: Any) -> bool: 

34 """Detect patterns like (a+)+ that cause catastrophic backtracking.""" 

35 for op, av in data: 

36 if op in _REPEAT_OPS: 

37 if _contains_quantifier(av[2]): 

38 return True 

39 elif op == sre_parse.SUBPATTERN: 

40 if _has_nested_quantifiers(av[-1]): 

41 return True 

42 elif op == sre_parse.BRANCH: 

43 if any(_has_nested_quantifiers(branch) for branch in av[1]): 

44 return True 

45 return False 

46 

47 

48def _contains_quantifier(data: Any) -> bool: 

49 for op, av in data: 

50 if op in _REPEAT_OPS: 

51 return True 

52 elif op == sre_parse.SUBPATTERN: 

53 if _contains_quantifier(av[-1]): 

54 return True 

55 elif op == sre_parse.BRANCH: 

56 if any(_contains_quantifier(branch) for branch in av[1]): 

57 return True 

58 return False 

59 

60 

61CONFIG_FILENAME = "CODEREVIEW.toml" 

62 

63 

64_TEAM_REF_SEGMENT = r"[a-zA-Z0-9][a-zA-Z0-9\-_.]*" 

65_TEAM_REF_RE = re.compile(rf"^{_TEAM_REF_SEGMENT}(/{_TEAM_REF_SEGMENT})+$") 

66 

67# Fields that hold reviewer identities (plain usernames, `$aliases`, and 

68# `@team` refs) rather than paths/code/labels. Team refs only expand here. 

69USER_LIST_FIELDS = ("authors", "reviewers", "alternates", "cc") 

70 

71# Roster fields where "!" performs compile-time subtraction (remove from the 

72# resolved list) rather than surviving as a match-time predicate. `authors` 

73# is deliberately excluded — its "!" entries are consumed by `matches_author`. 

74ROSTER_FIELDS = ("reviewers", "alternates", "cc") 

75 

76 

77def _split_team_ref(value: str) -> tuple[str, str] | None: 

78 """Split a value into `(prefix, ref)` if it has team-reference shape 

79 (`@org/team` or `!@org/team`), `prefix` being `"!"` or `""`. Returns 

80 `None` for anything else, so callers fall back to their own handling of 

81 plain values. 

82 """ 

83 if value.startswith("!@"): 

84 return "!", value[2:] 

85 if value.startswith("@"): 

86 return "", value[1:] 

87 return None 

88 

89 

90def is_unexpanded_ref(value: str) -> bool: 

91 """True if the value is a `$alias` or `@team` reference (optionally 

92 negated with a leading `!`) that has not been expanded — as opposed to a 

93 plain username. Only an offline compile (`teams=None`) leaves such values 

94 in user-list fields. 

95 """ 

96 return value.removeprefix("!").startswith(("$", "@")) 

97 

98 

99def _validate_team_refs(values: list[str]) -> list[str]: 

100 """Team references (`@org/team`, `!@org/team`) need at least two 

101 slash-separated segments. This only checks shape — membership is resolved 

102 later, at compile time, against the caller-provided `teams` mapping. 

103 

104 Any other value containing "@" is rejected too — that shape is reserved 

105 for team references (and, in future, email-style identifiers), so it 

106 can't be confused with a plain username. 

107 """ 

108 for value in values: 

109 split = _split_team_ref(value) 

110 if split is None: 

111 if "@" in value: 

112 raise ValueError( 

113 f"Invalid value '{value}': email addresses are not " 

114 "supported here — use the platform username" 

115 ) 

116 continue 

117 

118 _prefix, ref = split 

119 if not _TEAM_REF_RE.match(ref): 

120 raise ValueError( 

121 f"Invalid team reference '{value}': team references need the " 

122 "'org/team' form" 

123 ) 

124 return values 

125 

126 

127def _expand_team_ref( 

128 ref: str, teams: dict[str, list[str]] | None, prefix: str 

129) -> list[str]: 

130 """Expand a single team reference (without its `@`/`!@`) to member usernames. 

131 

132 `teams=None` is the offline/CLI mode: the library never calls out to 

133 GitHub/GitLab itself, so with no mapping provided the reference passes 

134 through unexpanded rather than erroring. With a `teams` mapping (even an 

135 empty one), an unresolvable ref is a loud config error, matching how 

136 unknown `$aliases` are handled. 

137 """ 

138 if teams is None: 

139 return [f"{prefix}@{ref}"] 

140 

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

142 if members is None: 

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

144 

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

146 

147 

148def _expand_aliases( 

149 values: list[str], 

150 aliases: dict[str, list[str]], 

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

152 expand_teams: bool = False, 

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

154 _path: list[str] | None = None, 

155) -> list[str]: 

156 """Replace alias references in a list with their mapped values recursively. 

157 

158 Team references (`@org/team`) are only expanded when `expand_teams` is 

159 True — user-list fields (reviewers, alternates, authors, cc). Elsewhere 

160 (paths, code, labels) a leading `@` is a literal string, e.g. npm-style 

161 scoped path patterns like `@vendor/pkg/**`. Teams are always leaf nodes — 

162 their values are plain usernames, never other refs — so expanding one 

163 never recurses further. 

164 """ 

165 if _seen is None: 

166 _seen = set() 

167 if _path is None: 

168 _path = [] 

169 

170 expanded: list[str] = [] 

171 for value in values: 

172 # Support negated aliases like "!$team" -> ["!alice", "!bob"] 

173 if value.startswith("!$"): 

174 prefix = "!" 

175 alias_ref = value[2:] 

176 elif value.startswith("$"): 

177 prefix = "" 

178 alias_ref = value[1:] 

179 elif expand_teams and (team_ref := _split_team_ref(value)) is not None: 

180 team_prefix, ref = team_ref 

181 expanded.extend(_expand_team_ref(ref=ref, teams=teams, prefix=team_prefix)) 

182 continue 

183 else: 

184 expanded.append(value) 

185 continue 

186 

187 if alias_ref in _seen: 

188 # Cycle detected, raise an error with the cycle path 

189 cycle_path = _path[_path.index(alias_ref) :] + [alias_ref] 

190 raise ValueError( 

191 f"Circular reference detected in aliases: {' -> '.join(cycle_path)}" 

192 ) 

193 if alias_ref in aliases: 

194 _seen.add(alias_ref) 

195 _path.append(alias_ref) 

196 # Recursively expand the alias values 

197 nested_expanded = _expand_aliases( 

198 aliases[alias_ref], 

199 aliases=aliases, 

200 teams=teams, 

201 expand_teams=expand_teams, 

202 _seen=_seen, 

203 _path=_path, 

204 ) 

205 if prefix: 

206 expanded.extend(prefix + v for v in nested_expanded) 

207 else: 

208 expanded.extend(nested_expanded) 

209 _path.pop() 

210 _seen.remove(alias_ref) 

211 else: 

212 # Unknown alias — surface it loudly instead of silently dropping the 

213 # reference (a typo'd alias would otherwise vanish reviewers/paths). 

214 raise ValueError(f"Unknown alias: {prefix}${alias_ref}") 

215 

216 # Remove duplicates while preserving order 

217 return list(dict.fromkeys(expanded)) 

218 

219 

220def _apply_negations(values: list[str]) -> list[str]: 

221 """Compile-time subtraction for roster fields (`ROSTER_FIELDS`): a "!name" 

222 entry removes "name" from the resolved list instead of surviving as a 

223 match-time rule (contrast `authors`, handled by `matches_author`). 

224 

225 A list still holding an unexpanded `@team` ref (offline compile, i.e. 

226 teams=None) is returned untouched — it isn't fully resolved yet, so 

227 subtraction can't run, and the partially-resolved config must round-trip 

228 unchanged. `$aliases` must already be expanded by the caller. 

229 

230 The wildcard+negation error fires BEFORE that early return: it's a check 

231 on the written form (`"*"` alongside any `"!"` entry, team ref or not), 

232 so an offline compile must reject it the same way the server will — a 

233 `pullapprove check` that passes locally can't then error in production. 

234 """ 

235 if "*" in values and any(v.startswith("!") for v in values): 

236 raise ValueError( 

237 'Negation cannot be combined with "*" in reviewers/alternates/cc ' 

238 "(wildcard exclusion is not supported)" 

239 ) 

240 

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

242 return values 

243 

244 negations = {v[1:].lower() for v in values if v.startswith("!")} 

245 return [v for v in values if not v.startswith("!") and v.lower() not in negations] 

246 

247 

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

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

250 

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

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

253 file's directory. 

254 

255 Raises if the reference escapes above the repo root. 

256 """ 

257 if extends_ref.startswith("/"): 

258 resolved = posixpath.normpath(extends_ref.lstrip("/")) 

259 else: 

260 base_dir = posixpath.dirname(extending_path) 

261 resolved = posixpath.normpath(posixpath.join(base_dir, extends_ref)) 

262 

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

264 raise ValueError( 

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

266 ) 

267 

268 return resolved 

269 

270 

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

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

273 

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

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

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

277 """ 

278 negate = pattern.startswith("!") 

279 if negate: 

280 pattern = pattern[1:] 

281 

282 if pattern.startswith("/"): 

283 anchored = pattern.lstrip("/") 

284 elif base_dir: 

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

286 else: 

287 anchored = pattern 

288 

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

290 

291 

292class OwnershipChoices(StrEnum): 

293 EMPTY = "" 

294 APPEND = "append" 

295 GLOBAL = "global" 

296 

297 

298class ScopeModel(BaseModel): 

299 model_config = ConfigDict(extra="forbid") 

300 

301 # Required fields 

302 name: str = Field(min_length=1) 

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

304 

305 # Optional fields 

306 

307 # Expanded version of lines could be dict 

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

309 code: list[str] = [] 

310 

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

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

313 # - labels 

314 # - ref 

315 # - statuses 

316 # - dates 

317 # - body 

318 # - title 

319 # - other scopes 

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

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

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

323 authors: list[str] = [] 

324 

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

326 description: str = "" 

327 reviewers: list[str] = [] 

328 alternates: list[str] = [] 

329 cc: list[str] = [] 

330 

331 # Review scoring 

332 require: int = 0 

333 author_value: int = 0 

334 

335 # How scopes are combined 

336 ownership: OwnershipChoices = OwnershipChoices.EMPTY 

337 

338 # Actionable items 

339 request: int = 0 

340 labels: list[str] = [] 

341 instructions: str = "" 

342 

343 # Approval checklist 

344 checklist: Checklist | None = None 

345 

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

347 @classmethod 

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

349 if "," in name: 

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

351 return name 

352 

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

354 @classmethod 

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

356 return _validate_team_refs(values) 

357 

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

359 @classmethod 

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

361 for pattern in code: 

362 try: 

363 parsed = sre_parse.parse(pattern) 

364 except re.error as e: 

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

366 if _has_nested_quantifiers(parsed): 

367 raise ValueError( 

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

369 "which can cause catastrophic backtracking." 

370 ) 

371 return code 

372 

373 @model_validator(mode="after") 

374 def validate_reviewers_for_require(self) -> ScopeModel: 

375 all_reviewers = self.reviewers + self.alternates 

376 

377 # Skip if wildcard - anyone can review 

378 if "*" in all_reviewers: 

379 return self 

380 

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

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

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

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

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

386 return self 

387 

388 if len(all_reviewers) < self.require: 

389 raise ValueError( 

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

391 ) 

392 return self 

393 

394 def printed_name(self) -> str: 

395 match self.ownership: 

396 case OwnershipChoices.APPEND: 

397 return "+" + self.name 

398 case OwnershipChoices.GLOBAL: 

399 return "*" + self.name 

400 

401 return self.name 

402 

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

404 return self.name == other.name 

405 

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

407 # TODO paths shouldn't start with / 

408 return glob.globmatch( 

409 path, 

410 self.paths, 

411 flags=glob.GLOBSTAR 

412 | glob.BRACE 

413 | glob.NEGATE 

414 | glob.IGNORECASE 

415 | glob.DOTGLOB, 

416 ) 

417 

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

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

420 if not patterns: 

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

422 self._code_regex_patterns = patterns 

423 

424 for pattern in patterns: 

425 for match in pattern.finditer(code): 

426 start_index = match.start() 

427 end_index = match.end() 

428 

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

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

431 

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

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

434 

435 yield { 

436 "start_line": start_line, 

437 "start_col": start_col, 

438 "end_line": end_line, 

439 "end_col": end_col, 

440 } 

441 

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

443 if not self.authors: 

444 # No authors specified, so assume it matches 

445 return True 

446 

447 author_username_lower = author_username.lower() 

448 

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

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

451 

452 if author_username_lower in negated_authors: 

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

454 return False 

455 

456 if not authors: 

457 # Negation-only: everyone not negated matches 

458 return True 

459 

460 if author_username_lower in authors: 

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

462 return True 

463 

464 return False 

465 

466 

467class LargeScaleChangeModel(BaseModel): 

468 model_config = ConfigDict(extra="forbid") 

469 

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

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

472 

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

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

475 require: int = 1 

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

477 # min_paths: int = 300 

478 # min_lines: int = 3000 

479 labels: list[str] = [] 

480 # really need author value too...? 

481 

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

483 @classmethod 

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

485 return _validate_team_refs(values) 

486 

487 

488class ConfigModel(BaseModel): 

489 model_config = ConfigDict(extra="forbid") 

490 

491 # Nothing is technically required 

492 extends: list[str] = [] 

493 template: bool = False 

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

495 large_scale_change: LargeScaleChangeModel | None = None 

496 scopes: list[ScopeModel] = [] 

497 

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

499 @classmethod 

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

501 seen: set[str] = set() 

502 for scope in scopes: 

503 if scope.name.lower() in seen: 

504 raise ValueError(f"Duplicate scope name: {scope.name}") 

505 seen.add(scope.name.lower()) 

506 

507 return scopes 

508 

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

510 @classmethod 

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

512 for i, path in enumerate(extends): 

513 basename = Path(path).name 

514 if not basename.startswith(CONFIG_FILENAME_PREFIX): 

515 raise ValueError( 

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

517 ) 

518 return extends 

519 

520 def compiled_config( 

521 self, 

522 config_path: Path, 

523 other_configs: ConfigModels, 

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

525 ) -> ConfigModel: 

526 """ 

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

528 

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

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

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

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

533 

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

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

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

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

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

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

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

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

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

543 mapping. 

544 

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

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

547 """ 

548 

549 if teams is not None: 

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

551 

552 compiled_data = self._merged_data(config_path, other_configs) 

553 

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

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

556 for scope in compiled_data["scopes"]: 

557 for field in [ 

558 "paths", 

559 "code", 

560 "authors", 

561 "reviewers", 

562 "alternates", 

563 "cc", 

564 "labels", 

565 ]: 

566 if field in scope: 

567 scope[field] = _expand_aliases( 

568 scope[field], 

569 compiled_data["aliases"], 

570 teams=teams, 

571 expand_teams=field in USER_LIST_FIELDS, 

572 ) 

573 

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

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

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

577 for scope in compiled_data["scopes"]: 

578 for field in ROSTER_FIELDS: 

579 if field in scope: 

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

581 

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

583 large_scale_change["reviewers"] = _expand_aliases( 

584 large_scale_change["reviewers"], 

585 compiled_data["aliases"], 

586 teams=teams, 

587 expand_teams=True, 

588 ) 

589 large_scale_change["labels"] = _expand_aliases( 

590 large_scale_change["labels"], 

591 compiled_data["aliases"], 

592 ) 

593 large_scale_change["reviewers"] = _apply_negations( 

594 large_scale_change["reviewers"] 

595 ) 

596 

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

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

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

600 for scope in compiled_data["scopes"]: 

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

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

603 

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

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

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

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

608 compiled_data["extends"] = [] 

609 compiled_data["aliases"] = {} 

610 

611 return ConfigModel.from_data( 

612 data=compiled_data, 

613 path=config_path, 

614 ) 

615 

616 def _merged_data( 

617 self, 

618 config_path: Path, 

619 other_configs: ConfigModels, 

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

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

622 ) -> dict[str, Any]: 

623 """ 

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

625 

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

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

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

629 

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

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

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

633 """ 

634 if _in_progress is None: 

635 _in_progress = [] 

636 if _seen is None: 

637 _seen = set() 

638 

639 config_path_str = str(config_path) 

640 if config_path_str in _in_progress: 

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

642 config_path_str 

643 ] 

644 raise ValueError( 

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

646 ) 

647 _in_progress.append(config_path_str) 

648 

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

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

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

652 

653 for extend_path in self.extends: 

654 resolved_path = _resolve_extends_path(config_path_str, extend_path) 

655 if resolved_path not in other_configs: 

656 raise ValueError( 

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

658 ) 

659 if resolved_path in _seen: 

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

661 continue 

662 

663 parent_data = other_configs[resolved_path]._merged_data( 

664 Path(resolved_path), other_configs, _in_progress, _seen 

665 ) 

666 inherited_scopes = inherited_scopes + parent_data["scopes"] 

667 inherited_aliases = inherited_aliases | parent_data["aliases"] 

668 inherited_lsc = inherited_lsc or parent_data["large_scale_change"] 

669 

670 merged = self.model_dump() 

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

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

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

674 

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

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

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

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

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

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

681 if not self.template: 

682 base_dir = posixpath.dirname(config_path_str) 

683 for scope in merged["scopes"]: 

684 scope.setdefault("_anchor_dir", base_dir) 

685 

686 _seen.add(config_path_str) 

687 _in_progress.pop() 

688 

689 return merged 

690 

691 @classmethod 

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

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

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

695 

696 @classmethod 

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

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

699 

700 @classmethod 

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

702 return cls(**data) 

703 

704 

705class _ConfigModelsBase(RootModel): 

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

707 

708 root: dict[str, ConfigModel] 

709 

710 @classmethod 

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

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

713 configs = cls(root={}) 

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

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

716 return configs 

717 

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

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

720 

721 def __bool__(self) -> bool: 

722 return bool(self.root) 

723 

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

725 return self.root[key] 

726 

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

728 return key in self.root 

729 

730 def __len__(self) -> int: 

731 return len(self.root) 

732 

733 

734class ConfigModels(_ConfigModelsBase): 

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

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

737 

738 @classmethod 

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

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

741 configs = cls(root={}) 

742 

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

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

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

746 

747 return configs 

748 

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

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

751 

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

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

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

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

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

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

758 

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

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

761 

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

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

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

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

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

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

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

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

770 earlier. 

771 """ 

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

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

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

775 candidate_lists: list[list[str]] = [] 

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

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

778 for scope in config.scopes: 

779 for field in USER_LIST_FIELDS: 

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

781 if config.large_scale_change: 

782 candidate_lists.append(config.large_scale_change.reviewers) 

783 if not any( 

784 _split_team_ref(value) is not None 

785 for values in candidate_lists 

786 for value in values 

787 ): 

788 return set() 

789 

790 refs: set[str] = set() 

791 

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

793 for value in values: 

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

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

796 

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

798 if config.template: 

799 continue 

800 for scope in config.scopes: 

801 for field in USER_LIST_FIELDS: 

802 collect_refs(getattr(scope, field)) 

803 if config.large_scale_change: 

804 collect_refs(config.large_scale_change.reviewers) 

805 

806 return refs 

807 

808 def compiled( 

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

810 ) -> CompiledConfigModels: 

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

812 

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

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

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

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

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

818 the set for display). 

819 

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

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

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

823 semantics). 

824 

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

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

827 double-apply. 

828 """ 

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

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

831 if config.template: 

832 effective[path] = config 

833 else: 

834 effective[path] = config.compiled_config( 

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

836 ) 

837 

838 return CompiledConfigModels.from_config_models(effective) 

839 

840 

841class CompiledConfigModels(_ConfigModelsBase): 

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

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

844 

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

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

847 for parent in file_path.parents: 

848 parent_config_path = str(parent / CONFIG_FILENAME) 

849 

850 if parent_config_path in self.root: 

851 config = self.root[parent_config_path] 

852 

853 if config.template: 

854 # Skip templates 

855 continue 

856 

857 return config 

858