Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/hooks.py: 56%

1# hooks.py -- for dealing with git hooks 

2# Copyright (C) 2012-2013 Jelmer Vernooij and others. 

3# 

4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later 

5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU 

6# General Public License as published by the Free Software Foundation; version 2.0 

7# or (at your option) any later version. You can redistribute it and/or 

8# modify it under the terms of either of these two licenses. 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15# 

16# You should have received a copy of the licenses; if not, see 

17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License 

18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache 

19# License, Version 2.0. 

20# 

21 

22"""Access to hooks.""" 

23 

24__all__ = [ 

25 "CommitMsgShellHook", 

26 "Hook", 

27 "PostCommitShellHook", 

28 "PostReceiveShellHook", 

29 "PreCommitShellHook", 

30 "PreReceiveShellHook", 

31 "ShellHook", 

32 "UpdateShellHook", 

33] 

34 

35import os 

36import subprocess 

37from collections.abc import Callable, Sequence 

38from typing import Any 

39 

40from .errors import HookError 

41 

42 

43class Hook: 

44 """Generic hook object.""" 

45 

46 def execute(self, *args: Any) -> Any: # noqa: ANN401 

47 """Execute the hook with the given args. 

48 

49 Args: 

50 args: argument list to hook 

51 Raises: 

52 HookError: hook execution failure 

53 Returns: 

54 a hook may return a useful value 

55 """ 

56 raise NotImplementedError(self.execute) 

57 

58 

59class ShellHook(Hook): 

60 """Hook by executable file. 

61 

62 Implements standard githooks(5) [0]: 

63 

64 [0] http://www.kernel.org/pub/software/scm/git/docs/githooks.html 

65 """ 

66 

67 def __init__( 

68 self, 

69 name: str, 

70 path: str, 

71 numparam: int, 

72 pre_exec_callback: Callable[..., Any] | None = None, 

73 post_exec_callback: Callable[..., Any] | None = None, 

74 cwd: str | None = None, 

75 ) -> None: 

76 """Setup shell hook definition. 

77 

78 Args: 

79 name: name of hook for error messages 

80 path: absolute path to executable file 

81 numparam: number of requirements parameters 

82 pre_exec_callback: closure for setup before execution 

83 Defaults to None. Takes in the variable argument list from the 

84 execute functions and returns a modified argument list for the 

85 shell hook. 

86 post_exec_callback: closure for cleanup after execution 

87 Defaults to None. Takes in a boolean for hook success and the 

88 modified argument list and returns the final hook return value 

89 if applicable 

90 cwd: working directory to switch to when executing the hook 

91 """ 

92 self.name = name 

93 self.filepath = path 

94 self.numparam = numparam 

95 

96 self.pre_exec_callback = pre_exec_callback 

97 self.post_exec_callback = post_exec_callback 

98 

99 self.cwd = cwd 

100 

101 def execute(self, *args: Any) -> Any: # noqa: ANN401 

102 """Execute the hook with given args.""" 

103 if len(args) != self.numparam: 

104 raise HookError( 

105 f"Hook {self.name} executed with wrong number of args. Expected {self.numparam}. Saw {len(args)}. args: {args}" 

106 ) 

107 

108 if self.pre_exec_callback is not None: 

109 args = self.pre_exec_callback(*args) 

110 

111 try: 

112 ret = subprocess.call( 

113 [os.path.relpath(self.filepath, self.cwd), *list(args)], cwd=self.cwd 

114 ) 

115 if ret != 0: 

116 if self.post_exec_callback is not None: 

117 self.post_exec_callback(0, *args) 

118 raise HookError(f"Hook {self.name} exited with non-zero status {ret}") 

119 if self.post_exec_callback is not None: 

120 return self.post_exec_callback(1, *args) 

121 except FileNotFoundError: # no file. silent failure. 

122 if self.post_exec_callback is not None: 

123 self.post_exec_callback(0, *args) 

124 

125 

126class PreCommitShellHook(ShellHook): 

127 """pre-commit shell hook.""" 

128 

129 def __init__(self, cwd: str, controldir: str) -> None: 

130 """Initialize pre-commit hook. 

131 

132 Args: 

133 cwd: Working directory for hook execution 

134 controldir: Path to the git control directory (.git) 

135 """ 

136 filepath = os.path.join(controldir, "hooks", "pre-commit") 

137 

138 ShellHook.__init__(self, "pre-commit", filepath, 0, cwd=cwd) 

139 

140 

141class PostCommitShellHook(ShellHook): 

142 """post-commit shell hook.""" 

143 

144 def __init__(self, controldir: str) -> None: 

145 """Initialize post-commit hook. 

146 

147 Args: 

148 controldir: Path to the git control directory (.git) 

149 """ 

150 filepath = os.path.join(controldir, "hooks", "post-commit") 

151 

152 ShellHook.__init__(self, "post-commit", filepath, 0, cwd=controldir) 

153 

154 

155class CommitMsgShellHook(ShellHook): 

156 """commit-msg shell hook.""" 

157 

158 def __init__(self, controldir: str) -> None: 

159 """Initialize commit-msg hook. 

160 

161 Args: 

162 controldir: Path to the git control directory (.git) 

163 """ 

164 filepath = os.path.join(controldir, "hooks", "commit-msg") 

165 

166 def prepare_msg(*args: bytes) -> tuple[str, ...]: 

167 import tempfile 

168 

169 (fd, path) = tempfile.mkstemp() 

170 

171 with os.fdopen(fd, "wb") as f: 

172 f.write(args[0]) 

173 

174 return (path,) 

175 

176 def clean_msg(success: int, *args: str) -> bytes | None: 

177 if success: 

178 with open(args[0], "rb") as f: 

179 new_msg = f.read() 

180 os.unlink(args[0]) 

181 return new_msg 

182 os.unlink(args[0]) 

183 return None 

184 

185 ShellHook.__init__( 

186 self, "commit-msg", filepath, 1, prepare_msg, clean_msg, controldir 

187 ) 

188 

189 

190class PostReceiveShellHook(ShellHook): 

191 """post-receive shell hook.""" 

192 

193 def __init__(self, controldir: str) -> None: 

194 """Initialize post-receive hook. 

195 

196 Args: 

197 controldir: Path to the git control directory (.git) 

198 """ 

199 self.controldir = controldir 

200 filepath = os.path.join(controldir, "hooks", "post-receive") 

201 ShellHook.__init__(self, "post-receive", path=filepath, numparam=0) 

202 

203 def execute( 

204 self, client_refs: Sequence[tuple[bytes, bytes, bytes]] 

205 ) -> bytes | None: 

206 """Execute the post-receive hook. 

207 

208 Args: 

209 client_refs: List of tuples containing (old_sha, new_sha, ref_name) 

210 for each updated reference 

211 

212 Returns: 

213 Output from the hook execution or None if hook doesn't exist 

214 

215 Raises: 

216 HookError: If hook execution fails 

217 """ 

218 # do nothing if the script doesn't exist 

219 if not os.path.exists(self.filepath): 

220 return None 

221 

222 try: 

223 env = os.environ.copy() 

224 env["GIT_DIR"] = self.controldir 

225 

226 p = subprocess.Popen( 

227 self.filepath, 

228 stdin=subprocess.PIPE, 

229 stdout=subprocess.PIPE, 

230 stderr=subprocess.PIPE, 

231 env=env, 

232 ) 

233 

234 # client_refs is a list of (oldsha, newsha, ref) 

235 in_data = b"\n".join([b" ".join(ref) for ref in client_refs]) 

236 

237 out_data, err_data = p.communicate(in_data) 

238 

239 if (p.returncode != 0) or err_data: 

240 err_msg = ( 

241 f"post-receive exit code: {p.returncode}\n" 

242 f"stdout:\n{out_data.decode('utf-8', 'backslashreplace')}\n" 

243 f"stderr:\n{err_data.decode('utf-8', 'backslashreplace')}" 

244 ) 

245 raise HookError(err_msg) 

246 return out_data 

247 except OSError as err: 

248 raise HookError(repr(err)) from err 

249 

250 

251class PreReceiveShellHook(ShellHook): 

252 """pre-receive shell hook.""" 

253 

254 def __init__(self, controldir: str) -> None: 

255 """Initialize pre-receive hook. 

256 

257 Args: 

258 controldir: Path to the git control directory (.git) 

259 """ 

260 self.controldir = controldir 

261 filepath = os.path.join(controldir, "hooks", "pre-receive") 

262 ShellHook.__init__(self, "pre-receive", path=filepath, numparam=0) 

263 

264 def execute( 

265 self, client_refs: Sequence[tuple[bytes, bytes, bytes]] 

266 ) -> tuple[bytes, bytes]: 

267 """Execute the pre-receive hook. 

268 

269 Args: 

270 client_refs: List of tuples containing (old_sha, new_sha, ref_name) 

271 for each reference to be updated 

272 

273 Returns: 

274 Tuple of (stdout, stderr) from hook execution 

275 

276 Raises: 

277 HookError: If hook execution fails (exits with non-zero status) 

278 """ 

279 # do nothing if the script doesn't exist 

280 if not os.path.exists(self.filepath): 

281 return (b"", b"") 

282 

283 try: 

284 env = os.environ.copy() 

285 env["GIT_DIR"] = self.controldir 

286 

287 p = subprocess.Popen( 

288 self.filepath, 

289 stdin=subprocess.PIPE, 

290 stdout=subprocess.PIPE, 

291 stderr=subprocess.PIPE, 

292 env=env, 

293 ) 

294 

295 # client_refs is a list of (oldsha, newsha, ref) 

296 in_data = b"\n".join([b" ".join(ref) for ref in client_refs]) 

297 

298 out_data, err_data = p.communicate(in_data) 

299 

300 if p.returncode != 0: 

301 raise HookError( 

302 f"pre-receive hook exited with status {p.returncode}: {err_data.decode('utf-8', 'backslashreplace')}" 

303 ) 

304 return (out_data, err_data) 

305 except OSError as err: 

306 raise HookError(repr(err)) from err 

307 

308 

309class UpdateShellHook(ShellHook): 

310 """update shell hook.""" 

311 

312 def __init__(self, controldir: str) -> None: 

313 """Initialize update hook. 

314 

315 Args: 

316 controldir: Path to the git control directory (.git) 

317 """ 

318 self.controldir = controldir 

319 filepath = os.path.join(controldir, "hooks", "update") 

320 ShellHook.__init__(self, "update", path=filepath, numparam=3) 

321 

322 def execute( 

323 self, ref_name: bytes, old_sha: bytes, new_sha: bytes 

324 ) -> tuple[bytes, bytes]: 

325 """Execute the update hook for a single ref. 

326 

327 Args: 

328 ref_name: Name of the reference being updated 

329 old_sha: Old SHA of the reference 

330 new_sha: New SHA of the reference 

331 

332 Returns: 

333 Tuple of (stdout, stderr) from hook execution 

334 

335 Raises: 

336 HookError: If hook execution fails (exits with non-zero status) 

337 """ 

338 # do nothing if the script doesn't exist 

339 if not os.path.exists(self.filepath): 

340 return (b"", b"") 

341 

342 try: 

343 env = os.environ.copy() 

344 env["GIT_DIR"] = self.controldir 

345 

346 p = subprocess.Popen( 

347 [ 

348 self.filepath, 

349 ref_name.decode("utf-8", "backslashreplace"), 

350 old_sha.decode("utf-8", "backslashreplace"), 

351 new_sha.decode("utf-8", "backslashreplace"), 

352 ], 

353 stdout=subprocess.PIPE, 

354 stderr=subprocess.PIPE, 

355 env=env, 

356 ) 

357 

358 out_data, err_data = p.communicate() 

359 

360 if p.returncode != 0: 

361 raise HookError( 

362 f"update hook exited with status {p.returncode}: {err_data.decode('utf-8', 'backslashreplace')}" 

363 ) 

364 return (out_data, err_data) 

365 except OSError as err: 

366 raise HookError(repr(err)) from err