Coverage for src/pullapprove/cli.py: 70%

251 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-08-17 21:57 -0500

1from __future__ import annotations 

2 

3import os 

4import subprocess 

5import sys 

6from collections import Counter, deque 

7from pathlib import Path 

8from textwrap import dedent 

9 

10import click 

11from pydantic import ValidationError 

12 

13from . import git 

14from .config import CONFIG_FILENAME, ConfigModel, ConfigModels, is_config_filename 

15from .diff import DiffFile, DiffHunk, filter_diff_text, iter_file_hunks 

16from .matches import match_diff, match_files 

17from .printer import MatchesPrinter 

18from .text import plural 

19from .trust import TRUST_FAMILIES, Trust, trust_diff, trust_label 

20 

21 

22@click.group() 

23@click.version_option(package_name="pullapprove") 

24@click.pass_context 

25def cli(ctx: click.Context) -> None: 

26 pass 

27 

28 

29@cli.command() 

30@click.option("--filename", default=CONFIG_FILENAME, help="Configuration filename") 

31def init(filename: str) -> None: 

32 """Create a new config file""" 

33 config_path = Path(filename) 

34 if config_path.exists(): 

35 click.secho(f"{filename} already exists!", fg="red") 

36 sys.exit(1) 

37 

38 # Could we use blame to guess? 

39 # go straight to agent? 

40 # gh auth status can give us the user? or ask what's their username? 

41 # keep it simple - agent can do more when I get to it 

42 

43 contents = f""" 

44 [[scopes]] 

45 name = "default" 

46 paths = ["**/*"] 

47 request = 1 

48 require = 1 

49 reviewers = ["<YOU>"] 

50 

51 [[scopes]] 

52 name = "pullapprove" 

53 paths = ["**/{CONFIG_FILENAME}"] 

54 request = 1 

55 require = 1 

56 reviewers = ["<YOU>"] 

57 """ 

58 config_path.write_text(dedent(contents).strip() + "\n") 

59 click.secho(f"Created {filename}") 

60 

61 

62@cli.command() 

63@click.option("--quiet", is_flag=True) 

64def check(quiet: bool) -> ConfigModels: 

65 """ 

66 Validate configuration files 

67 """ 

68 

69 if not quiet: 

70 if Path(".pullapprove.yml").exists(): 

71 click.secho( 

72 f"{click.style('[Warning]', fg='yellow')} This repo still contains a PullApprove v3 config file (.pullapprove.yml). Consider migrating it to PullApprove v5." 

73 ) 

74 if Path("CODEOWNERS").exists(): 

75 click.secho( 

76 f"{click.style('[Warning]', fg='yellow')} This repo still contains a CODEOWNERS file. Consider migrating it to PullApprove v5." 

77 ) 

78 if Path("docs/CODEOWNERS").exists(): 

79 click.secho( 

80 f"{click.style('[Warning]', fg='yellow')} This repo still contains a CODEOWNERS file (docs/CODEOWNERS). Consider migrating it to PullApprove v5." 

81 ) 

82 if Path(".github/CODEOWNERS").exists(): 

83 click.secho( 

84 f"{click.style('[Warning]', fg='yellow')} This repo still contains a CODEOWNERS file (.github/CODEOWNERS). Consider migrating it to PullApprove v5." 

85 ) 

86 

87 errors = {} 

88 configs = ConfigModels(root={}) 

89 

90 for root, _, files in os.walk("."): 

91 for f in files: 

92 if is_config_filename(f): 

93 config_path = Path(root) / f 

94 

95 if not quiet: 

96 click.echo(config_path, nl=False) 

97 try: 

98 configs.add_config( 

99 ConfigModel.from_filesystem(config_path), config_path 

100 ) 

101 

102 if not quiet: 

103 click.secho(" -> OK", fg="green") 

104 except ValidationError as e: 

105 if not quiet: 

106 click.secho(" -> ERROR", fg="red") 

107 

108 errors[config_path] = e 

109 

110 for path, error in errors.items(): 

111 click.secho(str(path), fg="red") 

112 print(error) 

113 

114 if errors: 

115 raise click.Abort("Configuration validation failed.") 

116 

117 # Compile the whole set offline to catch what per-file validation 

118 # can't: unknown aliases, missing/circular extends, and invalid 

119 # combinations like wildcard+negation. teams=None leaves `@team` refs 

120 # unexpanded — the same partial resolution `match`/`coverage` use — 

121 # so anything that errors here would also error in production. 

122 try: 

123 configs.compiled() 

124 except (ValidationError, ValueError) as e: 

125 click.secho(f"ERROR: {e}", fg="red") 

126 raise click.Abort("Configuration validation failed.") 

127 

128 if not configs and not quiet: 

129 click.secho(f"No {CONFIG_FILENAME} files found.", fg="red") 

130 sys.exit(1) 

131 

132 return configs 

133 

134 

135@cli.command() 

136@click.option("--changed", is_flag=True, help="Show only changed files") 

137@click.option("--staged", is_flag=True, help="Show only staged files") 

138@click.option("--diff", is_flag=True, help="Show diff content with matches") 

139@click.option( 

140 "--by-scope", is_flag=True, help="Organize output by scope instead of by path" 

141) 

142@click.option( 

143 "--scope", 

144 multiple=True, 

145 help="Filter to show only files matching these scopes (can be used multiple times)", 

146) 

147@click.argument("paths", nargs=-1, type=click.Path()) 

148@click.pass_context 

149def match( 

150 ctx: click.Context, 

151 changed: bool, 

152 staged: bool, 

153 diff: bool, 

154 by_scope: bool, 

155 scope: tuple[str, ...], 

156 paths: tuple[str, ...], 

157) -> None: 

158 """ 

159 Show files and their matching scopes 

160 

161 If PATHS are provided, only those specific paths will be matched. 

162 Directories will be recursively expanded to include all files within them. 

163 Otherwise, all files in the repository will be matched. 

164 """ 

165 configs = ctx.invoke(check, quiet=True).compiled() 

166 

167 if not configs: 

168 click.secho("No valid configurations found.", fg="red") 

169 raise click.Abort("No configurations to check.") 

170 

171 if paths: 

172 # When specific paths are provided, match only those paths 

173 if diff or staged or changed: 

174 click.secho( 

175 "Cannot use --diff, --staged, or --changed with specific paths.", 

176 fg="red", 

177 ) 

178 raise click.Abort("Conflicting options.") 

179 

180 # Get all git-tracked files and filter by provided paths 

181 all_git_files = set(git.git_ls_files(Path("."))) 

182 expanded_paths = [] 

183 

184 for path_str in paths: 

185 path = Path(path_str) 

186 if path.is_dir(): 

187 # Filter git files that are within this directory 

188 dir_prefix = str(path) + "/" 

189 for git_file in all_git_files: 

190 if git_file.startswith(dir_prefix) or git_file == str(path): 

191 expanded_paths.append(git_file) 

192 elif path.is_file() or str(path) in all_git_files: 

193 # Include if it's a git-tracked file 

194 if str(path) in all_git_files: 

195 expanded_paths.append(str(path)) 

196 else: 

197 click.secho(f"File not tracked by git: {path}", fg="yellow") 

198 else: 

199 click.secho( 

200 f"Path does not exist or not tracked by git: {path}", fg="yellow" 

201 ) 

202 

203 matches = match_files(configs, iter(expanded_paths)) 

204 all_files = expanded_paths 

205 elif diff or staged: 

206 # Use git diff for these options 

207 diff_args = [] 

208 if staged: 

209 diff_args.append("--staged") 

210 

211 diff_stream = git.git_diff_stream(Path("."), *diff_args) 

212 try: 

213 diff_results = match_diff(configs, diff_stream) 

214 except subprocess.CalledProcessError as exc: 

215 # git failed (not a repo, bad state) — it already explained itself on 

216 # stderr, so exit cleanly instead of dumping a traceback. 

217 raise click.ClickException("Could not read a diff from git.") from exc 

218 matches = diff_results.matches 

219 # For diff mode, we only show files in the diff 

220 all_files = None 

221 elif changed: 

222 iterator = git.git_ls_changes(Path(".")) 

223 matches = match_files(configs, iterator) 

224 # For changed mode, we only show changed files 

225 all_files = None 

226 else: 

227 # For normal mode, show all files to see gaps 

228 iterator = git.git_ls_files(Path(".")) 

229 matches = match_files(configs, iterator) 

230 # Get all files again for the printer 

231 all_files = list(git.git_ls_files(Path("."))) 

232 

233 printer = MatchesPrinter(matches, all_files=all_files) 

234 if by_scope: 

235 printer.print_by_scope(scope_filter=scope) 

236 else: 

237 printer.print_by_path(scope_filter=scope) 

238 

239 

240@cli.command() 

241@click.option( 

242 "--check", 

243 "check_flag", 

244 is_flag=True, 

245 help="Exit with non-zero status if coverage is incomplete", 

246) 

247@click.argument("path", type=click.Path(exists=True), default=".") 

248@click.pass_context 

249def coverage(ctx: click.Context, path: str, check_flag: bool) -> None: 

250 """ 

251 Calculate file coverage for review scopes 

252 """ 

253 configs = ctx.invoke(check, quiet=True).compiled() 

254 

255 num_matched = 0 

256 num_total = 0 

257 uncovered_files = [] 

258 

259 # First, get all files to know the total count for progress bar 

260 all_files = list(git.git_ls_files(Path(path))) 

261 

262 if not all_files: 

263 click.echo("No files found") 

264 return 

265 

266 # Process files with progress bar 

267 with click.progressbar( 

268 all_files, label="Analyzing coverage", show_percent=True, show_pos=True 

269 ) as files: 

270 # Use match_files to get proper scope matching including code patterns 

271 results = match_files(configs, iter(files)) 

272 

273 # Count files with and without scope matches 

274 for path_str, path_match in results.paths.items(): 

275 if path_match.scopes: 

276 num_matched += 1 

277 else: 

278 uncovered_files.append(path_str) 

279 num_total += 1 

280 

281 # Also count files that weren't in the results (no scope matches at all) 

282 for f in all_files: 

283 if f not in results.paths: 

284 uncovered_files.append(f) 

285 num_total += 1 

286 

287 percentage = (num_matched / num_total) * 100 

288 

289 # Display coverage statistics 

290 if num_matched == num_total: 

291 click.secho(f"\n✓ {num_matched}/{num_total} files covered (100.0%)", fg="green") 

292 else: 

293 # Show uncovered files 

294 if uncovered_files: 

295 click.echo("\nUncovered files:") 

296 for file in sorted(uncovered_files)[:10]: # Show first 10 

297 click.echo(f" - {file}") 

298 if len(uncovered_files) > 10: 

299 click.echo(f" ...and {len(uncovered_files) - 10} more") 

300 

301 click.secho( 

302 f"\n{num_matched}/{num_total} files covered ({percentage:.1f}%)", 

303 fg="yellow", 

304 ) 

305 

306 if check_flag and num_matched != num_total: 

307 sys.exit(1) 

308 

309 

310@cli.command( 

311 "diff", 

312 context_settings={ 

313 "ignore_unknown_options": True, 

314 # Everything from the first git argument onward is git's, verbatim — 

315 # including a `--` separator, which click otherwise swallows. Without 

316 # this, `pullapprove diff HEAD -- deleted.py` reaches git as 

317 # `diff HEAD deleted.py`, and git refuses a path it can't also read as 

318 # a revision. The cost is that PullApprove's own flags have to come 

319 # first, which the help text says. 

320 "allow_interspersed_args": False, 

321 }, 

322) 

323@click.option( 

324 "--hide", 

325 "hide_families", 

326 multiple=True, 

327 help="Hide only these label families, e.g. formatting (repeatable). " 

328 "Default: hide every labeled hunk.", 

329) 

330@click.option( 

331 "--no-pager", 

332 is_flag=True, 

333 help="Write to stdout instead of paging (paging is the default at a terminal).", 

334) 

335@click.option( 

336 "--quiet", is_flag=True, help="Don't print the hidden-hunk summary to stderr." 

337) 

338@click.argument("git_diff_args", nargs=-1, type=click.UNPROCESSED) 

339def diff( 

340 hide_families: tuple[str, ...], 

341 no_pager: bool, 

342 quiet: bool, 

343 git_diff_args: tuple[str, ...], 

344) -> None: 

345 """ 

346 Show a diff with mechanically-trivial hunks hidden — a trust-aware `git diff`. 

347 

348 Runs `git diff`, forwarding any extra arguments (`--staged`, revisions, 

349 pathspecs); with no arguments, piped stdin is read instead. Trusted hunks — 

350 lockfiles, whitespace, formatting, comments, imports, type annotations — are 

351 dropped, and the result stays a valid unified diff. At a terminal the output 

352 is paged through git's pager (like `git diff`); when piped it goes straight 

353 to stdout: 

354 

355 pullapprove diff # paged, an alternative to `git diff` 

356 pullapprove diff --staged # forwarded to git 

357 pullapprove diff main -- app/ # `--` and pathspecs reach git intact 

358 git diff | pullapprove diff # paged 

359 git diff | pullapprove diff --no-pager | delta 

360 

361 PullApprove's own options (--hide, --no-pager, --quiet) must come 

362 before the first git argument. Everything from there on is handed to git 

363 untouched, so `--` and pathspecs mean to git exactly what they always do. 

364 """ 

365 # --hide takes a family (e.g. formatting) or a full label (formatting:style). 

366 # Reject typos up front so a misspelled filter can't silently hide everything. 

367 known = TRUST_FAMILIES | {label.value for label in Trust} 

368 unknown = [name for name in hide_families if name not in known] 

369 if unknown: 

370 raise click.BadParameter( 

371 f"unknown trust {'family' if len(unknown) == 1 else 'families'}: " 

372 f"{', '.join(unknown)}. " 

373 f"Valid families: {', '.join(sorted(TRUST_FAMILIES))}.", 

374 param_hint="--hide", 

375 ) 

376 

377 diff_text = _read_diff(git_diff_args) 

378 

379 def is_hidden(label: Trust) -> bool: 

380 if not hide_families: 

381 return True 

382 return label in hide_families or label.family in hide_families 

383 

384 # Classify each hunk, queueing one keep/hide decision per hunk in document 

385 # order. `filter_diff_text` re-parses the same lines with the same header 

386 # parsers, so its `keep_hunk` calls arrive in that same order and the queue 

387 # joins the two walks — no coordinate keys needed (concatenated multi-commit 

388 # input like `git log -p` can repeat identical coordinates). The shared 

389 # `split("\n")` is what guarantees the alignment: `str.splitlines` also 

390 # breaks on `\r` and other separators, which could let the two walks see 

391 # different lines. 

392 keep: deque[bool] = deque() 

393 counts: Counter[str] = Counter() 

394 for file, hunk in iter_file_hunks(diff_text.split("\n")): 

395 label = trust_label(file, hunk) 

396 if label and is_hidden(label): 

397 keep.append(False) 

398 counts[label] += 1 

399 else: 

400 keep.append(True) 

401 

402 filtered = filter_diff_text( 

403 diff_text, lambda path, old_line, new_line: keep.popleft() 

404 ) 

405 

406 # Page at a terminal like git does; pipe straight through when redirected. 

407 if no_pager or not sys.stdout.isatty(): 

408 click.echo(filtered, nl=False) 

409 else: 

410 git.page(filtered) 

411 

412 if not quiet and counts: 

413 total = sum(counts.values()) 

414 breakdown = ", ".join(f"{label} ({n})" for label, n in sorted(counts.items())) 

415 click.secho( 

416 f"Hid {total} {plural(total, 'hunk')}: {breakdown}", 

417 fg="yellow", 

418 err=True, 

419 ) 

420 

421 

422def _read_diff(git_diff_args: tuple[str, ...]) -> str: 

423 """Read a unified diff: from `git diff` when arguments were given, from piped 

424 stdin otherwise. 

425 

426 Explicit arguments always win — hooks, CI, and cron run with stdin redirected 

427 from /dev/null (not a tty), and `pullapprove diff --staged` there must run 

428 git, not silently read an empty stdin.""" 

429 if not git_diff_args and not sys.stdin.isatty(): 

430 return sys.stdin.read() 

431 try: 

432 # Run from the caller's cwd (not the repo root) so any forwarded pathspec 

433 # is resolved relative to where the user invoked us, matching `git diff`. 

434 # Git still emits repo-root-relative paths in the diff body regardless. 

435 return "".join(git.git_diff_stream(Path.cwd(), *git_diff_args)) 

436 except subprocess.CalledProcessError as exc: 

437 # Not a git repo, or git rejected the arguments — git already explained 

438 # why on stderr, so exit cleanly instead of dumping a traceback. 

439 raise click.ClickException("Could not read a diff from git.") from exc 

440 

441 

442@cli.command("trust", context_settings={"ignore_unknown_options": True}) 

443@click.option( 

444 "--list", 

445 "list_hunks", 

446 is_flag=True, 

447 help="List every hunk and its trust label, grouped by file.", 

448) 

449@click.argument("git_diff_args", nargs=-1, type=click.UNPROCESSED) 

450def trust(list_hunks: bool, git_diff_args: tuple[str, ...]) -> None: 

451 """ 

452 Report what the trust classifier identifies in a diff. 

453 

454 Runs `git diff`, forwarding any extra arguments (`--staged`, revisions, 

455 pathspecs); with no arguments, piped stdin is read instead. The default 

456 prints a summary of how much of the change is mechanical; --list prints 

457 every hunk and its label, so you can see — and debug — exactly what the 

458 classifier identifies. 

459 """ 

460 files = trust_diff(_read_diff(git_diff_args)) 

461 if list_hunks: 

462 _trust_list(files) 

463 else: 

464 _trust_summary(files) 

465 

466 

467def _hunk_header(hunk: DiffHunk) -> str: 

468 old = ( 

469 f"{hunk.old_line}" 

470 if hunk.old_length is None 

471 else f"{hunk.old_line},{hunk.old_length}" 

472 ) 

473 new = ( 

474 f"{hunk.new_line}" 

475 if hunk.new_length is None 

476 else f"{hunk.new_line},{hunk.new_length}" 

477 ) 

478 return f"@@ -{old} +{new} @@" 

479 

480 

481def _trust_list(files: list[DiffFile]) -> None: 

482 hunks = [(file, hunk) for file in files for hunk in file.hunks] 

483 if not hunks: 

484 click.echo("No changes.") 

485 return 

486 width = max(len(_hunk_header(hunk)) for _, hunk in hunks) 

487 current_path = None 

488 for file, hunk in hunks: 

489 if file.new_path != current_path: 

490 current_path = file.new_path 

491 click.secho(current_path, bold=True) 

492 header = _hunk_header(hunk).ljust(width) 

493 label = ( 

494 click.style(hunk.trust, fg="green") 

495 if hunk.trust 

496 else click.style("needs review", fg="yellow") 

497 ) 

498 click.echo(f" {header} {label}") 

499 

500 

501def _trust_summary(files: list[DiffFile]) -> None: 

502 all_hunks = [hunk for file in files for hunk in file.hunks] 

503 total = len(all_hunks) 

504 if not total: 

505 click.echo("No changes.") 

506 return 

507 

508 counts = Counter(hunk.trust for hunk in all_hunks if hunk.trust) 

509 n_trusted = sum(counts.values()) 

510 n_review = total - n_trusted 

511 

512 def pct(n: int) -> int: 

513 return round(n / total * 100) 

514 

515 click.echo( 

516 f"{total} {plural(total, 'hunk')} · {len(files)} {plural(len(files), 'file')}" 

517 ) 

518 

519 click.secho(f" {n_trusted} trusted ({pct(n_trusted)}%)", fg="green") 

520 if counts: 

521 width = max(len(label) for label in counts) 

522 for label, n in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0])): 

523 click.echo(f" {label.ljust(width)} {n:>3}") 

524 click.secho(f" {n_review} need review ({pct(n_review)}%)", fg="yellow") 

525 

526 fully = [ 

527 file.new_path 

528 for file in files 

529 if file.hunks and all(hunk.trust for hunk in file.hunks) 

530 ] 

531 if fully: 

532 shown = ", ".join(fully[:10]) 

533 more = f" +{len(fully) - 10} more" if len(fully) > 10 else "" 

534 click.echo( 

535 f" {len(fully)} {plural(len(fully), 'file')} fully trusted: {shown}{more}" 

536 ) 

537 

538 

539# list - find open PRs, find status url and send json request (needs PA token)