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

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 

24import os 

25import subprocess 

26from typing import Any, Callable, Optional 

27 

28from .errors import HookError 

29 

30 

31class Hook: 

32 """Generic hook object.""" 

33 

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

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

36 

37 Args: 

38 args: argument list to hook 

39 Raises: 

40 HookError: hook execution failure 

41 Returns: 

42 a hook may return a useful value 

43 """ 

44 raise NotImplementedError(self.execute) 

45 

46 

47class ShellHook(Hook): 

48 """Hook by executable file. 

49 

50 Implements standard githooks(5) [0]: 

51 

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

53 """ 

54 

55 def __init__( 

56 self, 

57 name: str, 

58 path: str, 

59 numparam: int, 

60 pre_exec_callback: Optional[Callable[..., Any]] = None, 

61 post_exec_callback: Optional[Callable[..., Any]] = None, 

62 cwd: Optional[str] = None, 

63 ) -> None: 

64 """Setup shell hook definition. 

65 

66 Args: 

67 name: name of hook for error messages 

68 path: absolute path to executable file 

69 numparam: number of requirements parameters 

70 pre_exec_callback: closure for setup before execution 

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

72 execute functions and returns a modified argument list for the 

73 shell hook. 

74 post_exec_callback: closure for cleanup after execution 

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

76 modified argument list and returns the final hook return value 

77 if applicable 

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

79 """ 

80 self.name = name 

81 self.filepath = path 

82 self.numparam = numparam 

83 

84 self.pre_exec_callback = pre_exec_callback 

85 self.post_exec_callback = post_exec_callback 

86 

87 self.cwd = cwd 

88 

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

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

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

92 raise HookError( 

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

94 ) 

95 

96 if self.pre_exec_callback is not None: 

97 args = self.pre_exec_callback(*args) 

98 

99 try: 

100 ret = subprocess.call( 

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

102 ) 

103 if ret != 0: 

104 if self.post_exec_callback is not None: 

105 self.post_exec_callback(0, *args) 

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

107 if self.post_exec_callback is not None: 

108 return self.post_exec_callback(1, *args) 

109 except FileNotFoundError: # no file. silent failure. 

110 if self.post_exec_callback is not None: 

111 self.post_exec_callback(0, *args) 

112 

113 

114class PreCommitShellHook(ShellHook): 

115 """pre-commit shell hook.""" 

116 

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

118 """Initialize pre-commit hook. 

119 

120 Args: 

121 cwd: Working directory for hook execution 

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

123 """ 

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

125 

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

127 

128 

129class PostCommitShellHook(ShellHook): 

130 """post-commit shell hook.""" 

131 

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

133 """Initialize post-commit hook. 

134 

135 Args: 

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

137 """ 

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

139 

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

141 

142 

143class CommitMsgShellHook(ShellHook): 

144 """commit-msg shell hook.""" 

145 

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

147 """Initialize commit-msg hook. 

148 

149 Args: 

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

151 """ 

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

153 

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

155 import tempfile 

156 

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

158 

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

160 f.write(args[0]) 

161 

162 return (path,) 

163 

164 def clean_msg(success: int, *args: str) -> Optional[bytes]: 

165 if success: 

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

167 new_msg = f.read() 

168 os.unlink(args[0]) 

169 return new_msg 

170 os.unlink(args[0]) 

171 return None 

172 

173 ShellHook.__init__( 

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

175 ) 

176 

177 

178class PostReceiveShellHook(ShellHook): 

179 """post-receive shell hook.""" 

180 

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

182 """Initialize post-receive hook. 

183 

184 Args: 

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

186 """ 

187 self.controldir = controldir 

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

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

190 

191 def execute(self, client_refs: list[tuple[bytes, bytes, bytes]]) -> Optional[bytes]: 

192 """Execute the post-receive hook. 

193 

194 Args: 

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

196 for each updated reference 

197 

198 Returns: 

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

200 

201 Raises: 

202 HookError: If hook execution fails 

203 """ 

204 # do nothing if the script doesn't exist 

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

206 return None 

207 

208 try: 

209 env = os.environ.copy() 

210 env["GIT_DIR"] = self.controldir 

211 

212 p = subprocess.Popen( 

213 self.filepath, 

214 stdin=subprocess.PIPE, 

215 stdout=subprocess.PIPE, 

216 stderr=subprocess.PIPE, 

217 env=env, 

218 ) 

219 

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

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

222 

223 out_data, err_data = p.communicate(in_data) 

224 

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

226 err_fmt = b"post-receive exit code: %d\n" + b"stdout:\n%s\nstderr:\n%s" 

227 err_msg = err_fmt % (p.returncode, out_data, err_data) 

228 raise HookError(err_msg.decode("utf-8", "backslashreplace")) 

229 return out_data 

230 except OSError as err: 

231 raise HookError(repr(err)) from err