Coverage for src / rhiza / models / _git / remote.py: 100%

1"""Remote git operations: cloning, sparse checkout, and HEAD resolution.""" 

2 

3import logging 

4import subprocess # nosec B404 

5from pathlib import Path 

6 

7from rhiza.models._git._base import GitContextBase 

8from rhiza.models._git.helpers import _log_git_stderr_errors 

9 

10 

11class RemoteOpsMixin(GitContextBase): 

12 """Clone/sparse-checkout operations against a remote template repository.""" 

13 

14 def update_sparse_checkout( 

15 self, 

16 tmp_dir: Path, 

17 include_paths: list[str], 

18 logger: logging.Logger | None = None, 

19 ) -> None: 

20 """Update sparse-checkout paths in an already-cloned repository. 

21 

22 Args: 

23 tmp_dir: Temporary directory with cloned repository. 

24 include_paths: Paths to include in sparse checkout. 

25 logger: Optional logger; defaults to module logger. 

26 """ 

27 logger = logger or logging.getLogger(__name__) 

28 

29 try: 

30 logger.debug(f"Updating sparse checkout paths: {include_paths}") 

31 subprocess.run( # nosec B603 # noqa: S603 

32 [self.executable, "sparse-checkout", "set", "--skip-checks", *include_paths], 

33 cwd=tmp_dir, 

34 check=True, 

35 capture_output=True, 

36 text=True, 

37 env=self.env, 

38 ) 

39 logger.debug("Sparse checkout paths updated") 

40 except subprocess.CalledProcessError as e: 

41 logger.exception("Failed to update sparse checkout paths") 

42 _log_git_stderr_errors(e.stderr) 

43 raise 

44 

45 def get_head_sha(self, repo_dir: Path) -> str: 

46 """Return the HEAD commit SHA of a cloned repository. 

47 

48 Args: 

49 repo_dir: Path to the git repository. 

50 

51 Returns: 

52 The full HEAD SHA. 

53 """ 

54 result = subprocess.run( # nosec B603 # noqa: S603 

55 [self.executable, "rev-parse", "HEAD"], 

56 cwd=repo_dir, 

57 capture_output=True, 

58 text=True, 

59 check=True, 

60 env=self.env, 

61 ) 

62 return result.stdout.strip() 

63 

64 def clone_repository( 

65 self, 

66 git_url: str, 

67 tmp_dir: Path, 

68 branch: str, 

69 include_paths: list[str], 

70 logger: logging.Logger | None = None, 

71 ) -> None: 

72 """Clone template repository with sparse checkout. 

73 

74 Args: 

75 git_url: URL of the repository to clone. 

76 tmp_dir: Temporary directory for cloning. 

77 branch: Branch to clone from the template repository. 

78 include_paths: Paths to include in sparse checkout. 

79 logger: Optional logger; defaults to module logger. 

80 """ 

81 logger = logger or logging.getLogger(__name__) 

82 

83 try: 

84 logger.debug("Executing git clone with sparse checkout") 

85 subprocess.run( # nosec B603 # noqa: S603 

86 [ 

87 self.executable, 

88 "clone", 

89 "--depth", 

90 "1", 

91 "--filter=blob:none", 

92 "--sparse", 

93 "--branch", 

94 branch, 

95 git_url, 

96 str(tmp_dir), 

97 ], 

98 check=True, 

99 capture_output=True, 

100 text=True, 

101 env=self.env, 

102 ) 

103 logger.debug("Git clone completed successfully") 

104 except subprocess.CalledProcessError as e: 

105 logger.exception(f"Failed to clone repository from {git_url}") 

106 _log_git_stderr_errors(e.stderr) 

107 logger.exception("Please check that:") 

108 logger.exception(" - The repository exists and is accessible") 

109 logger.exception(f" - Branch '{branch}' exists in the repository") 

110 logger.exception(" - You have network access to the git hosting service") 

111 raise 

112 

113 try: 

114 logger.debug("Initializing sparse checkout") 

115 subprocess.run( # nosec B603 # noqa: S603 

116 [self.executable, "sparse-checkout", "init", "--cone"], 

117 cwd=tmp_dir, 

118 check=True, 

119 capture_output=True, 

120 text=True, 

121 env=self.env, 

122 ) 

123 logger.debug("Sparse checkout initialized") 

124 except subprocess.CalledProcessError as e: 

125 logger.exception("Failed to initialize sparse checkout") 

126 _log_git_stderr_errors(e.stderr) 

127 raise 

128 

129 try: 

130 logger.debug(f"Setting sparse checkout paths: {include_paths}") 

131 subprocess.run( # nosec B603 # noqa: S603 

132 [self.executable, "sparse-checkout", "set", "--skip-checks", *include_paths], 

133 cwd=tmp_dir, 

134 check=True, 

135 capture_output=True, 

136 text=True, 

137 env=self.env, 

138 ) 

139 logger.debug("Sparse checkout paths configured") 

140 except subprocess.CalledProcessError as e: 

141 logger.exception("Failed to configure sparse checkout paths") 

142 _log_git_stderr_errors(e.stderr) 

143 raise 

144 

145 def clone_at_sha( 

146 self, 

147 git_url: str, 

148 sha: str, 

149 dest: Path, 

150 include_paths: list[str], 

151 logger: logging.Logger | None = None, 

152 ) -> None: 

153 """Clone the template repository and checkout a specific commit. 

154 

155 Args: 

156 git_url: URL of the repository to clone. 

157 sha: Commit SHA to check out. 

158 dest: Target directory for the clone. 

159 include_paths: Paths to include in sparse checkout. 

160 logger: Optional logger; defaults to module logger. 

161 """ 

162 logger = logger or logging.getLogger(__name__) 

163 try: 

164 subprocess.run( # nosec B603 # noqa: S603 

165 [ 

166 self.executable, 

167 "clone", 

168 "--filter=blob:none", 

169 "--sparse", 

170 "--no-checkout", 

171 git_url, 

172 str(dest), 

173 ], 

174 check=True, 

175 capture_output=True, 

176 text=True, 

177 env=self.env, 

178 ) 

179 except subprocess.CalledProcessError as e: 

180 logger.exception(f"Failed to clone repository for base snapshot: {git_url}") 

181 _log_git_stderr_errors(e.stderr) 

182 raise 

183 

184 try: 

185 subprocess.run( # nosec B603 # noqa: S603 

186 [self.executable, "sparse-checkout", "init", "--cone"], 

187 cwd=dest, 

188 check=True, 

189 capture_output=True, 

190 text=True, 

191 env=self.env, 

192 ) 

193 subprocess.run( # nosec B603 # noqa: S603 

194 [self.executable, "sparse-checkout", "set", "--skip-checks", *include_paths], 

195 cwd=dest, 

196 check=True, 

197 capture_output=True, 

198 text=True, 

199 env=self.env, 

200 ) 

201 except subprocess.CalledProcessError as e: 

202 logger.exception("Failed to configure sparse checkout for base snapshot") 

203 _log_git_stderr_errors(e.stderr) 

204 raise 

205 

206 try: 

207 subprocess.run( # nosec B603 # noqa: S603 

208 [self.executable, "checkout", sha], 

209 cwd=dest, 

210 check=True, 

211 capture_output=True, 

212 text=True, 

213 env=self.env, 

214 ) 

215 except subprocess.CalledProcessError as e: 

216 logger.exception(f"Failed to checkout base commit {sha[:12]}") 

217 _log_git_stderr_errors(e.stderr) 

218 raise