Coverage for src/pullapprove/git.py: 72%

1from __future__ import annotations 

2 

3import os 

4import subprocess 

5import sys 

6from collections.abc import Generator 

7from pathlib import Path 

8 

9 

10def git_root() -> Path: 

11 """Return the root directory of the git repository.""" 

12 output = subprocess.check_output(["git", "rev-parse", "--show-toplevel"]).strip() 

13 return Path(output.decode("utf-8")) 

14 

15 

16def git_pager() -> str: 

17 """The pager command git would use, honoring core.pager/GIT_PAGER/PAGER. 

18 

19 Returns a shell command string (git's pager values can include arguments or 

20 pipes, so callers run it through a shell, the way git does). Falls back to 

21 `cat` when git is unavailable or paging is disabled. 

22 """ 

23 try: 

24 output = subprocess.check_output(["git", "var", "GIT_PAGER"], text=True).strip() 

25 except (subprocess.CalledProcessError, OSError): 

26 output = "" 

27 return output or os.environ.get("PAGER", "") or "cat" 

28 

29 

30def page(text: str) -> None: 

31 """Display `text` through git's configured pager, the way `git diff` does.""" 

32 env = {**os.environ} 

33 env.setdefault("LESS", "FRX") # quit if one screen, keep colors, no init clear 

34 proc = subprocess.Popen( 

35 git_pager(), shell=True, stdin=subprocess.PIPE, text=True, env=env 

36 ) 

37 try: 

38 proc.communicate(text) 

39 except BrokenPipeError: 

40 pass # the pager was closed before reading everything (e.g. `q` in less) 

41 if proc.returncode: 

42 sys.stdout.write(text) # the pager failed to run; don't lose the output 

43 

44 

45def git_ls_files(path: Path) -> Generator[str]: 

46 """Yield files in the git repository one at a time.""" 

47 process = subprocess.Popen( 

48 [ 

49 "git", 

50 "ls-files", 

51 "--cached", 

52 "--deleted", 

53 "--others", 

54 "--exclude-standard", 

55 ], 

56 cwd=path, 

57 stdout=subprocess.PIPE, 

58 text=True, 

59 ) 

60 

61 assert process.stdout is not None 

62 for line in process.stdout: 

63 yield line.strip() 

64 

65 process.stdout.close() 

66 process.wait() 

67 

68 

69def git_ls_changes(path: Path) -> Generator[str]: 

70 process = subprocess.Popen( 

71 [ 

72 "git", 

73 "status", 

74 "--porcelain=v1", 

75 "--untracked-files=all", 

76 ], 

77 cwd=path, 

78 stdout=subprocess.PIPE, 

79 text=True, 

80 ) 

81 

82 assert process.stdout is not None 

83 for line in process.stdout: 

84 yield line.strip().split(" ", 1)[1] 

85 

86 process.stdout.close() 

87 process.wait() 

88 

89 

90def git_diff_stream(path: Path, *diff_args: str) -> Generator[str]: 

91 # `-c diff.noprefix=false` because our diff parser requires the `a/`/`b/` 

92 # (or mnemonic single-char) path prefixes to recognize file headers — a user 

93 # with `diff.noprefix = true` in their gitconfig would otherwise produce a 

94 # diff whose files we silently misattribute. An explicit `--no-prefix` in 

95 # diff_args still wins (later flags override). 

96 process = subprocess.Popen( 

97 ["git", "-c", "diff.noprefix=false", "diff", "--no-ext-diff"] + list(diff_args), 

98 cwd=path, 

99 stdout=subprocess.PIPE, 

100 text=True, 

101 ) 

102 

103 assert process.stdout is not None 

104 yield from process.stdout 

105 

106 process.stdout.close() 

107 returncode = process.wait() 

108 # 0 = success, 1 = changes found (with --exit-code); anything else is a real 

109 # git error (bad revision, unknown option) that would otherwise masquerade 

110 # as an empty diff. git has already printed the reason to stderr. 

111 if returncode not in (0, 1): 

112 raise subprocess.CalledProcessError(returncode, ["git", "diff", *diff_args])