Coverage for src/pullapprove/cli.py: 69%
252 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-28 14:54 -0500
1from __future__ import annotations
3import os
4import subprocess
5import sys
6from collections import Counter, deque
7from pathlib import Path
8from textwrap import dedent
10import click
11from pydantic import ValidationError
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 .trust import TRUST_FAMILIES, Trust, trust_diff, trust_label
21@click.group()
22@click.version_option(package_name="pullapprove")
23@click.pass_context
24def cli(ctx: click.Context) -> None:
25 pass
28@cli.command()
29@click.option("--filename", default=CONFIG_FILENAME, help="Configuration filename")
30def init(filename: str) -> None:
31 """Create a new config file"""
32 config_path = Path(filename)
33 if config_path.exists():
34 click.secho(f"{filename} already exists!", fg="red")
35 sys.exit(1)
37 # Could we use blame to guess?
38 # go straight to agent?
39 # gh auth status can give us the user? or ask what's their username?
40 # keep it simple - agent can do more when I get to it
42 contents = f"""
43 [[scopes]]
44 name = "default"
45 paths = ["**/*"]
46 request = 1
47 require = 1
48 reviewers = ["<YOU>"]
50 [[scopes]]
51 name = "pullapprove"
52 paths = ["**/{CONFIG_FILENAME}"]
53 request = 1
54 require = 1
55 reviewers = ["<YOU>"]
56 """
57 config_path.write_text(dedent(contents).strip() + "\n")
58 click.secho(f"Created {filename}")
61@cli.command()
62@click.option("--quiet", is_flag=True)
63def check(quiet: bool) -> ConfigModels:
64 """
65 Validate configuration files
66 """
68 if not quiet:
69 if Path(".pullapprove.yml").exists():
70 click.secho(
71 f"{click.style('[Warning]', fg='yellow')} This repo still contains a PullApprove v3 config file (.pullapprove.yml). Consider migrating it to PullApprove v5."
72 )
73 if Path("CODEOWNERS").exists():
74 click.secho(
75 f"{click.style('[Warning]', fg='yellow')} This repo still contains a CODEOWNERS file. Consider migrating it to PullApprove v5."
76 )
77 if Path("docs/CODEOWNERS").exists():
78 click.secho(
79 f"{click.style('[Warning]', fg='yellow')} This repo still contains a CODEOWNERS file (docs/CODEOWNERS). Consider migrating it to PullApprove v5."
80 )
81 if Path(".github/CODEOWNERS").exists():
82 click.secho(
83 f"{click.style('[Warning]', fg='yellow')} This repo still contains a CODEOWNERS file (.github/CODEOWNERS). Consider migrating it to PullApprove v5."
84 )
86 errors = {}
87 configs = ConfigModels(root={})
89 for root, _, files in os.walk("."):
90 for f in files:
91 if is_config_filename(f):
92 config_path = Path(root) / f
94 if not quiet:
95 click.echo(config_path, nl=False)
96 try:
97 configs.add_config(
98 ConfigModel.from_filesystem(config_path), config_path
99 )
101 if not quiet:
102 click.secho(" -> OK", fg="green")
103 except ValidationError as e:
104 if not quiet:
105 click.secho(" -> ERROR", fg="red")
107 errors[config_path] = e
109 for path, error in errors.items():
110 click.secho(str(path), fg="red")
111 print(error)
113 if errors:
114 raise click.Abort("Configuration validation failed.")
116 # Compile the whole set offline to catch what per-file validation
117 # can't: unknown aliases, missing/circular extends, and invalid
118 # combinations like wildcard+negation. teams=None leaves `@team` refs
119 # unexpanded — the same partial resolution `match`/`coverage` use —
120 # so anything that errors here would also error in production.
121 try:
122 configs.compiled()
123 except (ValidationError, ValueError) as e:
124 click.secho(f"ERROR: {e}", fg="red")
125 raise click.Abort("Configuration validation failed.")
127 if not configs and not quiet:
128 click.secho(f"No {CONFIG_FILENAME} files found.", fg="red")
129 sys.exit(1)
131 return configs
134@cli.command()
135@click.option("--changed", is_flag=True, help="Show only changed files")
136@click.option("--staged", is_flag=True, help="Show only staged files")
137@click.option("--diff", is_flag=True, help="Show diff content with matches")
138@click.option(
139 "--by-scope", is_flag=True, help="Organize output by scope instead of by path"
140)
141@click.option(
142 "--scope",
143 multiple=True,
144 help="Filter to show only files matching these scopes (can be used multiple times)",
145)
146@click.argument("paths", nargs=-1, type=click.Path())
147@click.pass_context
148def match(
149 ctx: click.Context,
150 changed: bool,
151 staged: bool,
152 diff: bool,
153 by_scope: bool,
154 scope: tuple[str, ...],
155 paths: tuple[str, ...],
156) -> None:
157 """
158 Show files and their matching scopes
160 If PATHS are provided, only those specific paths will be matched.
161 Directories will be recursively expanded to include all files within them.
162 Otherwise, all files in the repository will be matched.
163 """
164 configs = ctx.invoke(check, quiet=True).compiled()
166 if not configs:
167 click.secho("No valid configurations found.", fg="red")
168 raise click.Abort("No configurations to check.")
170 if paths:
171 # When specific paths are provided, match only those paths
172 if diff or staged or changed:
173 click.secho(
174 "Cannot use --diff, --staged, or --changed with specific paths.",
175 fg="red",
176 )
177 raise click.Abort("Conflicting options.")
179 # Get all git-tracked files and filter by provided paths
180 all_git_files = set(git.git_ls_files(Path(".")))
181 expanded_paths = []
183 for path_str in paths:
184 path = Path(path_str)
185 if path.is_dir():
186 # Filter git files that are within this directory
187 dir_prefix = str(path) + "/"
188 for git_file in all_git_files:
189 if git_file.startswith(dir_prefix) or git_file == str(path):
190 expanded_paths.append(git_file)
191 elif path.is_file() or str(path) in all_git_files:
192 # Include if it's a git-tracked file
193 if str(path) in all_git_files:
194 expanded_paths.append(str(path))
195 else:
196 click.secho(f"File not tracked by git: {path}", fg="yellow")
197 else:
198 click.secho(
199 f"Path does not exist or not tracked by git: {path}", fg="yellow"
200 )
202 matches = match_files(configs, iter(expanded_paths))
203 all_files = expanded_paths
204 elif diff or staged:
205 # Use git diff for these options
206 diff_args = []
207 if staged:
208 diff_args.append("--staged")
210 diff_stream = git.git_diff_stream(Path("."), *diff_args)
211 try:
212 diff_results = match_diff(configs, diff_stream)
213 except subprocess.CalledProcessError as exc:
214 # git failed (not a repo, bad state) — it already explained itself on
215 # stderr, so exit cleanly instead of dumping a traceback.
216 raise click.ClickException("Could not read a diff from git.") from exc
217 matches = diff_results.matches
218 # For diff mode, we only show files in the diff
219 all_files = None
220 elif changed:
221 iterator = git.git_ls_changes(Path("."))
222 matches = match_files(configs, iterator)
223 # For changed mode, we only show changed files
224 all_files = None
225 else:
226 # For normal mode, show all files to see gaps
227 iterator = git.git_ls_files(Path("."))
228 matches = match_files(configs, iterator)
229 # Get all files again for the printer
230 all_files = list(git.git_ls_files(Path(".")))
232 printer = MatchesPrinter(matches, all_files=all_files)
233 if by_scope:
234 printer.print_by_scope(scope_filter=scope)
235 else:
236 printer.print_by_path(scope_filter=scope)
239@cli.command()
240@click.option(
241 "--check",
242 "check_flag",
243 is_flag=True,
244 help="Exit with non-zero status if coverage is incomplete",
245)
246@click.argument("path", type=click.Path(exists=True), default=".")
247@click.pass_context
248def coverage(ctx: click.Context, path: str, check_flag: bool) -> None:
249 """
250 Calculate file coverage for review scopes
251 """
252 configs = ctx.invoke(check, quiet=True).compiled()
254 num_matched = 0
255 num_total = 0
256 uncovered_files = []
258 # First, get all files to know the total count for progress bar
259 all_files = list(git.git_ls_files(Path(path)))
261 if not all_files:
262 click.echo("No files found")
263 return
265 # Process files with progress bar
266 with click.progressbar(
267 all_files, label="Analyzing coverage", show_percent=True, show_pos=True
268 ) as files:
269 # Use match_files to get proper scope matching including code patterns
270 results = match_files(configs, iter(files))
272 # Count files with and without scope matches
273 for path_str, path_match in results.paths.items():
274 if path_match.scopes:
275 num_matched += 1
276 else:
277 uncovered_files.append(path_str)
278 num_total += 1
280 # Also count files that weren't in the results (no scope matches at all)
281 for f in all_files:
282 if f not in results.paths:
283 uncovered_files.append(f)
284 num_total += 1
286 percentage = (num_matched / num_total) * 100
288 # Display coverage statistics
289 if num_matched == num_total:
290 click.secho(f"\n✓ {num_matched}/{num_total} files covered (100.0%)", fg="green")
291 else:
292 # Show uncovered files
293 if uncovered_files:
294 click.echo("\nUncovered files:")
295 for file in sorted(uncovered_files)[:10]: # Show first 10
296 click.echo(f" - {file}")
297 if len(uncovered_files) > 10:
298 click.echo(f" ...and {len(uncovered_files) - 10} more")
300 click.secho(
301 f"\n{num_matched}/{num_total} files covered ({percentage:.1f}%)",
302 fg="yellow",
303 )
305 if check_flag and num_matched != num_total:
306 sys.exit(1)
309@cli.command(
310 "diff",
311 context_settings={"ignore_unknown_options": True},
312)
313@click.option(
314 "--hide",
315 "hide_families",
316 multiple=True,
317 help="Hide only these label families, e.g. formatting (repeatable). "
318 "Default: hide every labeled hunk.",
319)
320@click.option(
321 "--no-pager",
322 is_flag=True,
323 help="Write to stdout instead of paging (paging is the default at a terminal).",
324)
325@click.option(
326 "--quiet", is_flag=True, help="Don't print the hidden-hunk summary to stderr."
327)
328@click.argument("git_diff_args", nargs=-1, type=click.UNPROCESSED)
329def diff(
330 hide_families: tuple[str, ...],
331 no_pager: bool,
332 quiet: bool,
333 git_diff_args: tuple[str, ...],
334) -> None:
335 """
336 Show a diff with mechanically-trivial hunks hidden — a trust-aware `git diff`.
338 Runs `git diff`, forwarding any extra arguments (`--staged`, revisions,
339 pathspecs); with no arguments, piped stdin is read instead. Trusted hunks —
340 lockfiles, whitespace, formatting, comments, imports, type annotations — are
341 dropped, and the result stays a valid unified diff. At a terminal the output
342 is paged through git's pager (like `git diff`); when piped it goes straight
343 to stdout:
345 pullapprove diff # paged, an alternative to `git diff`
346 pullapprove diff --staged # forwarded to git
347 git diff | pullapprove diff # paged
348 git diff | pullapprove diff --no-pager | delta
349 """
350 # --hide takes a family (e.g. formatting) or a full label (formatting:style).
351 # Reject typos up front so a misspelled filter can't silently hide everything.
352 known = TRUST_FAMILIES | {label.value for label in Trust}
353 unknown = [name for name in hide_families if name not in known]
354 if unknown:
355 raise click.BadParameter(
356 f"unknown trust {'family' if len(unknown) == 1 else 'families'}: "
357 f"{', '.join(unknown)}. "
358 f"Valid families: {', '.join(sorted(TRUST_FAMILIES))}.",
359 param_hint="--hide",
360 )
362 diff_text = _read_diff(git_diff_args)
364 def is_hidden(label: Trust) -> bool:
365 if not hide_families:
366 return True
367 return label in hide_families or label.family in hide_families
369 # Classify each hunk, queueing one keep/hide decision per hunk in document
370 # order. `filter_diff_text` re-parses the same lines with the same header
371 # parsers, so its `keep_hunk` calls arrive in that same order and the queue
372 # joins the two walks — no coordinate keys needed (concatenated multi-commit
373 # input like `git log -p` can repeat identical coordinates). The shared
374 # `split("\n")` is what guarantees the alignment: `str.splitlines` also
375 # breaks on `\r` and other separators, which could let the two walks see
376 # different lines.
377 keep: deque[bool] = deque()
378 counts: Counter[str] = Counter()
379 for file, hunk in iter_file_hunks(diff_text.split("\n")):
380 label = trust_label(file, hunk)
381 if label and is_hidden(label):
382 keep.append(False)
383 counts[label] += 1
384 else:
385 keep.append(True)
387 filtered = filter_diff_text(
388 diff_text, lambda path, old_line, new_line: keep.popleft()
389 )
391 # Page at a terminal like git does; pipe straight through when redirected.
392 if no_pager or not sys.stdout.isatty():
393 click.echo(filtered, nl=False)
394 else:
395 git.page(filtered)
397 if not quiet and counts:
398 total = sum(counts.values())
399 breakdown = ", ".join(f"{label} ({n})" for label, n in sorted(counts.items()))
400 click.secho(
401 f"Hid {total} {_plural(total, 'hunk')}: {breakdown}",
402 fg="yellow",
403 err=True,
404 )
407def _plural(n: int, word: str) -> str:
408 return word if n == 1 else word + "s"
411def _read_diff(git_diff_args: tuple[str, ...]) -> str:
412 """Read a unified diff: from `git diff` when arguments were given, from piped
413 stdin otherwise.
415 Explicit arguments always win — hooks, CI, and cron run with stdin redirected
416 from /dev/null (not a tty), and `pullapprove diff --staged` there must run
417 git, not silently read an empty stdin."""
418 if not git_diff_args and not sys.stdin.isatty():
419 return sys.stdin.read()
420 try:
421 # Run from the caller's cwd (not the repo root) so any forwarded pathspec
422 # is resolved relative to where the user invoked us, matching `git diff`.
423 # Git still emits repo-root-relative paths in the diff body regardless.
424 return "".join(git.git_diff_stream(Path.cwd(), *git_diff_args))
425 except subprocess.CalledProcessError as exc:
426 # Not a git repo, or git rejected the arguments — git already explained
427 # why on stderr, so exit cleanly instead of dumping a traceback.
428 raise click.ClickException("Could not read a diff from git.") from exc
431@cli.command("trust", context_settings={"ignore_unknown_options": True})
432@click.option(
433 "--list",
434 "list_hunks",
435 is_flag=True,
436 help="List every hunk and its trust label, grouped by file.",
437)
438@click.argument("git_diff_args", nargs=-1, type=click.UNPROCESSED)
439def trust(list_hunks: bool, git_diff_args: tuple[str, ...]) -> None:
440 """
441 Report what the trust classifier identifies in a diff.
443 Runs `git diff`, forwarding any extra arguments (`--staged`, revisions,
444 pathspecs); with no arguments, piped stdin is read instead. The default
445 prints a summary of how much of the change is mechanical; --list prints
446 every hunk and its label, so you can see — and debug — exactly what the
447 classifier identifies.
448 """
449 files = trust_diff(_read_diff(git_diff_args))
450 if list_hunks:
451 _trust_list(files)
452 else:
453 _trust_summary(files)
456def _hunk_header(hunk: DiffHunk) -> str:
457 old = (
458 f"{hunk.old_line}"
459 if hunk.old_length is None
460 else f"{hunk.old_line},{hunk.old_length}"
461 )
462 new = (
463 f"{hunk.new_line}"
464 if hunk.new_length is None
465 else f"{hunk.new_line},{hunk.new_length}"
466 )
467 return f"@@ -{old} +{new} @@"
470def _trust_list(files: list[DiffFile]) -> None:
471 hunks = [(file, hunk) for file in files for hunk in file.hunks]
472 if not hunks:
473 click.echo("No changes.")
474 return
475 width = max(len(_hunk_header(hunk)) for _, hunk in hunks)
476 current_path = None
477 for file, hunk in hunks:
478 if file.new_path != current_path:
479 current_path = file.new_path
480 click.secho(current_path, bold=True)
481 header = _hunk_header(hunk).ljust(width)
482 label = (
483 click.style(hunk.trust, fg="green")
484 if hunk.trust
485 else click.style("needs review", fg="yellow")
486 )
487 click.echo(f" {header} {label}")
490def _trust_summary(files: list[DiffFile]) -> None:
491 all_hunks = [hunk for file in files for hunk in file.hunks]
492 total = len(all_hunks)
493 if not total:
494 click.echo("No changes.")
495 return
497 counts = Counter(hunk.trust for hunk in all_hunks if hunk.trust)
498 n_trusted = sum(counts.values())
499 n_review = total - n_trusted
501 def pct(n: int) -> int:
502 return round(n / total * 100)
504 click.echo(
505 f"{total} {_plural(total, 'hunk')} · {len(files)} {_plural(len(files), 'file')}"
506 )
508 click.secho(f" {n_trusted} trusted ({pct(n_trusted)}%)", fg="green")
509 if counts:
510 width = max(len(label) for label in counts)
511 for label, n in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0])):
512 click.echo(f" {label.ljust(width)} {n:>3}")
513 click.secho(f" {n_review} need review ({pct(n_review)}%)", fg="yellow")
515 fully = [
516 file.new_path
517 for file in files
518 if file.hunks and all(hunk.trust for hunk in file.hunks)
519 ]
520 if fully:
521 shown = ", ".join(fully[:10])
522 more = f" +{len(fully) - 10} more" if len(fully) > 10 else ""
523 click.echo(
524 f" {len(fully)} {_plural(len(fully), 'file')} fully trusted: {shown}{more}"
525 )
528# list - find open PRs, find status url and send json request (needs PA token)