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

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 "ShellHook", 

31] 

32 

33import os 

34import subprocess 

35from collections.abc import Callable, Sequence 

36from typing import Any 

37 

38from .errors import HookError 

39 

40 

41class Hook: 

42 """Generic hook object.""" 

43 

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

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

46 

47 Args: 

48 args: argument list to hook 

49 Raises: 

50 HookError: hook execution failure 

51 Returns: 

52 a hook may return a useful value 

53 """ 

54 raise NotImplementedError(self.execute) 

55 

56 

57class ShellHook(Hook): 

58 """Hook by executable file. 

59 

60 Implements standard githooks(5) [0]: 

61 

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

63 """ 

64 

65 def __init__( 

66 self, 

67 name: str, 

68 path: str, 

69 numparam: int, 

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

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

72 cwd: str | None = None, 

73 ) -> None: 

74 """Setup shell hook definition. 

75 

76 Args: 

77 name: name of hook for error messages 

78 path: absolute path to executable file 

79 numparam: number of requirements parameters 

80 pre_exec_callback: closure for setup before execution 

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

82 execute functions and returns a modified argument list for the 

83 shell hook. 

84 post_exec_callback: closure for cleanup after execution 

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

86 modified argument list and returns the final hook return value 

87 if applicable 

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

89 """ 

90 self.name = name 

91 self.filepath = path 

92 self.numparam = numparam 

93 

94 self.pre_exec_callback = pre_exec_callback 

95 self.post_exec_callback = post_exec_callback 

96 

97 self.cwd = cwd 

98 

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

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

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

102 raise HookError( 

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

104 ) 

105 

106 if self.pre_exec_callback is not None: 

107 args = self.pre_exec_callback(*args) 

108 

109 try: 

110 ret = subprocess.call( 

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

112 ) 

113 if ret != 0: 

114 if self.post_exec_callback is not None: 

115 self.post_exec_callback(0, *args) 

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

117 if self.post_exec_callback is not None: 

118 return self.post_exec_callback(1, *args) 

119 except FileNotFoundError: # no file. silent failure. 

120 if self.post_exec_callback is not None: 

121 self.post_exec_callback(0, *args) 

122 

123 

124class PreCommitShellHook(ShellHook): 

125 """pre-commit shell hook.""" 

126 

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

128 """Initialize pre-commit hook. 

129 

130 Args: 

131 cwd: Working directory for hook execution 

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

133 """ 

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

135 

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

137 

138 

139class PostCommitShellHook(ShellHook): 

140 """post-commit shell hook.""" 

141 

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

143 """Initialize post-commit hook. 

144 

145 Args: 

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

147 """ 

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

149 

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

151 

152 

153class CommitMsgShellHook(ShellHook): 

154 """commit-msg shell hook.""" 

155 

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

157 """Initialize commit-msg hook. 

158 

159 Args: 

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

161 """ 

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

163 

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

165 import tempfile 

166 

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

168 

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

170 f.write(args[0]) 

171 

172 return (path,) 

173 

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

175 if success: 

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

177 new_msg = f.read() 

178 os.unlink(args[0]) 

179 return new_msg 

180 os.unlink(args[0]) 

181 return None 

182 

183 ShellHook.__init__( 

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

185 ) 

186 

187 

188class PostReceiveShellHook(ShellHook): 

189 """post-receive shell hook.""" 

190 

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

192 """Initialize post-receive hook. 

193 

194 Args: 

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

196 """ 

197 self.controldir = controldir 

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

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

200 

201 def execute( 

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

203 ) -> bytes | None: 

204 """Execute the post-receive hook. 

205 

206 Args: 

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

208 for each updated reference 

209 

210 Returns: 

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

212 

213 Raises: 

214 HookError: If hook execution fails 

215 """ 

216 # do nothing if the script doesn't exist 

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

218 return None 

219 

220 try: 

221 env = os.environ.copy() 

222 env["GIT_DIR"] = self.controldir 

223 

224 p = subprocess.Popen( 

225 self.filepath, 

226 stdin=subprocess.PIPE, 

227 stdout=subprocess.PIPE, 

228 stderr=subprocess.PIPE, 

229 env=env, 

230 ) 

231 

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

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

234 

235 out_data, err_data = p.communicate(in_data) 

236 

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

238 err_msg = ( 

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

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

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

242 ) 

243 raise HookError(err_msg) 

244 return out_data 

245 except OSError as err: 

246 raise HookError(repr(err)) from err