Coverage for scripts/_rhiza_git.py: 100%

1#!/usr/bin/env python3 

2"""Git subprocess engine for the stdlib-only `sync` port. 

3 

4This is the private helper behind `scripts/sync.py`, mirroring the role 

5`_rhiza_yaml.py` plays for parsing: it owns every `git` invocation, so the 

6orchestration in `sync.py` stays free of subprocess detail. Ported from the rhiza 

7CLI's `rhiza.models._git` engine. 

8 

9The engine shells out to `git` for everything hard — `git clone --sparse` and 

10`git merge-file` — rather than re-implementing it in Python. Every call goes 

11through the single :func:`_run_git` seam, which keeps the module trivially 

12testable (real git for happy paths, a monkeypatched seam for error branches). 

13 

14The diff machinery this module used to carry — `git diff --no-index`, 

15`git apply -3`, `git apply --reject` and the diff-text parser that recovered a 

16file list from their output — is gone. `_rhiza_merge.py` reads that list off the 

17two snapshot directories instead, which is where it was always available. Nothing 

18runs `git apply` any more, so `.rej` files are no longer produced at all. 

19 

20The functions here do not print: they return structured facts and leave 

21user-facing output to `sync.py`. 

22""" 

23 

24from __future__ import annotations 

25 

26import os 

27import shutil 

28import subprocess # nosec B404 

29from dataclasses import dataclass, field 

30from pathlib import Path 

31 

32 

33def get_git_executable() -> str: 

34 """Return the absolute path to the git binary. 

35 

36 Raises: 

37 RuntimeError: If git is not found on ``PATH``. 

38 """ 

39 git_path = shutil.which("git") 

40 if git_path is None: 

41 msg = "git executable not found in PATH. Please ensure git is installed and available." 

42 raise RuntimeError(msg) 

43 return git_path 

44 

45 

46@dataclass 

47class GitContext: 

48 """The git executable path and environment shared across subprocess calls. 

49 

50 Attributes: 

51 executable: Absolute path to the git binary. 

52 env: Environment variables passed to every git subprocess. 

53 """ 

54 

55 executable: str 

56 env: dict[str, str] = field(default_factory=dict) 

57 

58 @classmethod 

59 def default(cls) -> GitContext: 

60 """Build a context from the system git and a prompt-disabled environment.""" 

61 env = os.environ.copy() 

62 env["GIT_TERMINAL_PROMPT"] = "0" 

63 return cls(executable=get_git_executable(), env=env) 

64 

65 

66def _run_git( 

67 git: GitContext, 

68 args: list[str], 

69 *, 

70 cwd: Path | str | None = None, 

71 check: bool = False, 

72 stdin: bytes | None = None, 

73) -> subprocess.CompletedProcess[bytes]: 

74 """Run one git command through the single subprocess seam. 

75 

76 Args: 

77 git: The git context (executable + environment). 

78 args: Arguments following the git executable. 

79 cwd: Working directory for the command, if any. 

80 check: When True, raise ``CalledProcessError`` on a non-zero exit. 

81 stdin: Optional bytes fed to the process on standard input. 

82 

83 Returns: 

84 The completed process with captured (bytes) stdout/stderr. 

85 """ 

86 return subprocess.run( # nosec B603 

87 [git.executable, *args], 

88 cwd=str(cwd) if cwd is not None else None, 

89 input=stdin, 

90 capture_output=True, 

91 check=check, 

92 env=git.env, 

93 ) 

94 

95 

96# --------------------------------------------------------------------------- 

97# Working-tree and remote operations 

98# --------------------------------------------------------------------------- 

99 

100 

101def status_porcelain(git: GitContext, target: Path) -> list[str]: 

102 """Return the non-empty ``git status --porcelain`` lines for *target*.""" 

103 result = _run_git(git, ["status", "--porcelain"], cwd=target) 

104 return [line for line in result.stdout.decode().splitlines() if line.strip()] 

105 

106 

107def get_head_sha(git: GitContext, repo_dir: Path) -> str: 

108 """Return the full HEAD commit SHA of the repository at *repo_dir*.""" 

109 result = _run_git(git, ["rev-parse", "HEAD"], cwd=repo_dir, check=True) 

110 return result.stdout.decode().strip() 

111 

112 

113def _sparse_set(git: GitContext, work_dir: Path, include_paths: list[str]) -> None: 

114 """Set the sparse-checkout cone of the clone at *work_dir* to *include_paths*.""" 

115 _run_git( 

116 git, ["sparse-checkout", "set", "--skip-checks", *include_paths], cwd=work_dir, check=True 

117 ) 

118 

119 

120def clone( 

121 git: GitContext, 

122 git_url: str, 

123 dest: Path, 

124 include_paths: list[str], 

125 *, 

126 branch: str | None = None, 

127 sha: str | None = None, 

128) -> None: 

129 """Sparse-clone *git_url* into *dest* and set its cone to *include_paths*. 

130 

131 Pass *branch* for a shallow clone at a branch tip, or *sha* for a full-history 

132 clone checked out at that commit. 

133 """ 

134 if branch is not None: 

135 head = ["clone", "--depth", "1", "--filter=blob:none", "--sparse", "--branch", branch] 

136 else: 

137 head = ["clone", "--filter=blob:none", "--sparse", "--no-checkout"] 

138 _run_git(git, [*head, git_url, str(dest)], check=True) 

139 _run_git(git, ["sparse-checkout", "init", "--cone"], cwd=dest, check=True) 

140 _sparse_set(git, dest, include_paths) 

141 if sha is not None: 

142 _run_git(git, ["checkout", sha], cwd=dest, check=True) 

143 

144 

145def update_sparse_checkout(git: GitContext, tmp_dir: Path, include_paths: list[str]) -> None: 

146 """Reset the sparse-checkout cone of the clone at *tmp_dir* to *include_paths*.""" 

147 _sparse_set(git, tmp_dir, include_paths) 

148 

149 

150# --------------------------------------------------------------------------- 

151# Diff computation and parsing 

152# --------------------------------------------------------------------------- 

153 

154 

155def merge_file(git: GitContext, target_path: Path, base_path: Path, upstream_path: Path) -> int: 

156 """Three-way merge one file in place; return ``git merge-file``'s exit status. 

157 

158 The status is returned rather than flattened to a boolean because its values mean 

159 genuinely different things, and the caller acts differently on each: **0** merged 

160 cleanly, a small **positive** number is that many conflicted regions (markers have 

161 been written into *target_path*), and **255** is a refusal — in practice a binary 

162 file, where nothing was written and nothing can be. 

163 """ 

164 result = _run_git( 

165 git, 

166 [ 

167 "merge-file", 

168 "-L", 

169 "HEAD", 

170 "-L", 

171 "base", 

172 "-L", 

173 "rhiza-template", 

174 str(target_path), 

175 str(base_path), 

176 str(upstream_path), 

177 ], 

178 ) 

179 return result.returncode 

180 

181 

182def scan_conflict_artifacts(target: Path) -> tuple[list[str], list[str]]: 

183 """Scan *target* for merge artifacts, returning ``(rej_files, marker_files)``. 

184 

185 Looks for ``*.rej`` files and text files containing a ``<<<<<<<`` conflict marker. 

186 The merge no longer produces rejects — `git apply --reject` was the only thing that 

187 ever did, and it is gone — but the scan still reports them, because a repo can carry 

188 one from a hand-run `git apply` or an interrupted older sync, and silently ignoring 

189 it would be worse than naming it. 

190 """ 

191 rej_files: list[str] = [] 

192 marker_files: list[str] = [] 

193 for path in sorted(target.rglob("*")): 

194 if not path.is_file(): 

195 continue 

196 rel = str(path.relative_to(target)) 

197 if path.suffix == ".rej": 

198 rej_files.append(rel) 

199 else: 

200 try: 

201 if b"<<<<<<<" in path.read_bytes()[:1_048_576]: 

202 marker_files.append(rel) 

203 except OSError: 

204 pass 

205 return rej_files, marker_files