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

1from __future__ import annotations 

2 

3import os 

4import subprocess 

5import sys 

6from collections.abc import Generator 

7from pathlib import Path 

8 

9 

10def git_pager() -> str: 

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

12 

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

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

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

16 """ 

17 try: 

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

19 except (subprocess.CalledProcessError, OSError): 

20 output = "" 

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

22 

23 

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

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

26 env = {**os.environ} 

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

28 proc = subprocess.Popen( 

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

30 ) 

31 try: 

32 proc.communicate(text) 

33 except BrokenPipeError: 

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

35 if proc.returncode: 

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

37 

38 

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

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

41 process = subprocess.Popen( 

42 [ 

43 "git", 

44 "ls-files", 

45 "--cached", 

46 "--deleted", 

47 "--others", 

48 "--exclude-standard", 

49 ], 

50 cwd=path, 

51 stdout=subprocess.PIPE, 

52 text=True, 

53 ) 

54 

55 assert process.stdout is not None 

56 for line in process.stdout: 

57 yield line.strip() 

58 

59 process.stdout.close() 

60 process.wait() 

61 

62 

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

64 process = subprocess.Popen( 

65 [ 

66 "git", 

67 "status", 

68 "--porcelain=v1", 

69 "--untracked-files=all", 

70 ], 

71 cwd=path, 

72 stdout=subprocess.PIPE, 

73 text=True, 

74 ) 

75 

76 assert process.stdout is not None 

77 for line in process.stdout: 

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

79 

80 process.stdout.close() 

81 process.wait() 

82 

83 

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

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

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

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

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

89 # diff_args still wins (later flags override). 

90 process = subprocess.Popen( 

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

92 cwd=path, 

93 stdout=subprocess.PIPE, 

94 text=True, 

95 ) 

96 

97 assert process.stdout is not None 

98 yield from process.stdout 

99 

100 process.stdout.close() 

101 returncode = process.wait() 

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

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

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

105 if returncode not in (0, 1): 

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