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 collections.abc import Sequence 

27from typing import Any, Callable, Optional 

28 

29from .errors import HookError 

30 

31 

32class Hook: 

33 """Generic hook object.""" 

34 

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

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

37 

38 Args: 

39 args: argument list to hook 

40 Raises: 

41 HookError: hook execution failure 

42 Returns: 

43 a hook may return a useful value 

44 """ 

45 raise NotImplementedError(self.execute) 

46 

47 

48class ShellHook(Hook): 

49 """Hook by executable file. 

50 

51 Implements standard githooks(5) [0]: 

52 

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

54 """ 

55 

56 def __init__( 

57 self, 

58 name: str, 

59 path: str, 

60 numparam: int, 

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

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

63 cwd: Optional[str] = None, 

64 ) -> None: 

65 """Setup shell hook definition. 

66 

67 Args: 

68 name: name of hook for error messages 

69 path: absolute path to executable file 

70 numparam: number of requirements parameters 

71 pre_exec_callback: closure for setup before execution 

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

73 execute functions and returns a modified argument list for the 

74 shell hook. 

75 post_exec_callback: closure for cleanup after execution 

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

77 modified argument list and returns the final hook return value 

78 if applicable 

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

80 """ 

81 self.name = name 

82 self.filepath = path 

83 self.numparam = numparam 

84 

85 self.pre_exec_callback = pre_exec_callback 

86 self.post_exec_callback = post_exec_callback 

87 

88 self.cwd = cwd 

89 

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

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

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

93 raise HookError( 

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

95 ) 

96 

97 if self.pre_exec_callback is not None: 

98 args = self.pre_exec_callback(*args) 

99 

100 try: 

101 ret = subprocess.call( 

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

103 ) 

104 if ret != 0: 

105 if self.post_exec_callback is not None: 

106 self.post_exec_callback(0, *args) 

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

108 if self.post_exec_callback is not None: 

109 return self.post_exec_callback(1, *args) 

110 except FileNotFoundError: # no file. silent failure. 

111 if self.post_exec_callback is not None: 

112 self.post_exec_callback(0, *args) 

113 

114 

115class PreCommitShellHook(ShellHook): 

116 """pre-commit shell hook.""" 

117 

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

119 """Initialize pre-commit hook. 

120 

121 Args: 

122 cwd: Working directory for hook execution 

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

124 """ 

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

126 

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

128 

129 

130class PostCommitShellHook(ShellHook): 

131 """post-commit shell hook.""" 

132 

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

134 """Initialize post-commit hook. 

135 

136 Args: 

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

138 """ 

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

140 

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

142 

143 

144class CommitMsgShellHook(ShellHook): 

145 """commit-msg shell hook.""" 

146 

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

148 """Initialize commit-msg hook. 

149 

150 Args: 

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

152 """ 

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

154 

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

156 import tempfile 

157 

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

159 

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

161 f.write(args[0]) 

162 

163 return (path,) 

164 

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

166 if success: 

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

168 new_msg = f.read() 

169 os.unlink(args[0]) 

170 return new_msg 

171 os.unlink(args[0]) 

172 return None 

173 

174 ShellHook.__init__( 

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

176 ) 

177 

178 

179class PostReceiveShellHook(ShellHook): 

180 """post-receive shell hook.""" 

181 

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

183 """Initialize post-receive hook. 

184 

185 Args: 

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

187 """ 

188 self.controldir = controldir 

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

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

191 

192 def execute( 

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

194 ) -> Optional[bytes]: 

195 """Execute the post-receive hook. 

196 

197 Args: 

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

199 for each updated reference 

200 

201 Returns: 

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

203 

204 Raises: 

205 HookError: If hook execution fails 

206 """ 

207 # do nothing if the script doesn't exist 

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

209 return None 

210 

211 try: 

212 env = os.environ.copy() 

213 env["GIT_DIR"] = self.controldir 

214 

215 p = subprocess.Popen( 

216 self.filepath, 

217 stdin=subprocess.PIPE, 

218 stdout=subprocess.PIPE, 

219 stderr=subprocess.PIPE, 

220 env=env, 

221 ) 

222 

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

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

225 

226 out_data, err_data = p.communicate(in_data) 

227 

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

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

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

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

232 return out_data 

233 except OSError as err: 

234 raise HookError(repr(err)) from err