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 filepath = os.path.join(controldir, "hooks", "pre-commit") 

119 

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

121 

122 

123class PostCommitShellHook(ShellHook): 

124 """post-commit shell hook.""" 

125 

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

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

128 

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

130 

131 

132class CommitMsgShellHook(ShellHook): 

133 """commit-msg shell hook.""" 

134 

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

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

137 

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

139 import tempfile 

140 

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

142 

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

144 f.write(args[0]) 

145 

146 return (path,) 

147 

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

149 if success: 

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

151 new_msg = f.read() 

152 os.unlink(args[0]) 

153 return new_msg 

154 os.unlink(args[0]) 

155 return None 

156 

157 ShellHook.__init__( 

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

159 ) 

160 

161 

162class PostReceiveShellHook(ShellHook): 

163 """post-receive shell hook.""" 

164 

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

166 self.controldir = controldir 

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

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

169 

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

171 # do nothing if the script doesn't exist 

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

173 return None 

174 

175 try: 

176 env = os.environ.copy() 

177 env["GIT_DIR"] = self.controldir 

178 

179 p = subprocess.Popen( 

180 self.filepath, 

181 stdin=subprocess.PIPE, 

182 stdout=subprocess.PIPE, 

183 stderr=subprocess.PIPE, 

184 env=env, 

185 ) 

186 

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

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

189 

190 out_data, err_data = p.communicate(in_data) 

191 

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

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

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

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

196 return out_data 

197 except OSError as err: 

198 raise HookError(repr(err)) from err