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

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 public 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 

26 

27from .errors import HookError 

28 

29 

30class Hook: 

31 """Generic hook object.""" 

32 

33 def execute(self, *args): 

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

35 

36 Args: 

37 args: argument list to hook 

38 Raises: 

39 HookError: hook execution failure 

40 Returns: 

41 a hook may return a useful value 

42 """ 

43 raise NotImplementedError(self.execute) 

44 

45 

46class ShellHook(Hook): 

47 """Hook by executable file. 

48 

49 Implements standard githooks(5) [0]: 

50 

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

52 """ 

53 

54 def __init__( 

55 self, 

56 name, 

57 path, 

58 numparam, 

59 pre_exec_callback=None, 

60 post_exec_callback=None, 

61 cwd=None, 

62 ) -> None: 

63 """Setup shell hook definition. 

64 

65 Args: 

66 name: name of hook for error messages 

67 path: absolute path to executable file 

68 numparam: number of requirements parameters 

69 pre_exec_callback: closure for setup before execution 

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

71 execute functions and returns a modified argument list for the 

72 shell hook. 

73 post_exec_callback: closure for cleanup after execution 

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

75 modified argument list and returns the final hook return value 

76 if applicable 

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

78 """ 

79 self.name = name 

80 self.filepath = path 

81 self.numparam = numparam 

82 

83 self.pre_exec_callback = pre_exec_callback 

84 self.post_exec_callback = post_exec_callback 

85 

86 self.cwd = cwd 

87 

88 def execute(self, *args): 

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

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

91 raise HookError( 

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

93 ) 

94 

95 if self.pre_exec_callback is not None: 

96 args = self.pre_exec_callback(*args) 

97 

98 try: 

99 ret = subprocess.call( 

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

101 ) 

102 if ret != 0: 

103 if self.post_exec_callback is not None: 

104 self.post_exec_callback(0, *args) 

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

106 if self.post_exec_callback is not None: 

107 return self.post_exec_callback(1, *args) 

108 except OSError: # no file. silent failure. 

109 if self.post_exec_callback is not None: 

110 self.post_exec_callback(0, *args) 

111 

112 

113class PreCommitShellHook(ShellHook): 

114 """pre-commit shell hook.""" 

115 

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

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

118 

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

120 

121 

122class PostCommitShellHook(ShellHook): 

123 """post-commit shell hook.""" 

124 

125 def __init__(self, controldir) -> None: 

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

127 

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

129 

130 

131class CommitMsgShellHook(ShellHook): 

132 """commit-msg shell hook.""" 

133 

134 def __init__(self, controldir) -> None: 

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

136 

137 def prepare_msg(*args): 

138 import tempfile 

139 

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

141 

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

143 f.write(args[0]) 

144 

145 return (path,) 

146 

147 def clean_msg(success, *args): 

148 if success: 

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

150 new_msg = f.read() 

151 os.unlink(args[0]) 

152 return new_msg 

153 os.unlink(args[0]) 

154 

155 ShellHook.__init__( 

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

157 ) 

158 

159 

160class PostReceiveShellHook(ShellHook): 

161 """post-receive shell hook.""" 

162 

163 def __init__(self, controldir) -> None: 

164 self.controldir = controldir 

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

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

167 

168 def execute(self, client_refs): 

169 # do nothing if the script doesn't exist 

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

171 return None 

172 

173 try: 

174 env = os.environ.copy() 

175 env["GIT_DIR"] = self.controldir 

176 

177 p = subprocess.Popen( 

178 self.filepath, 

179 stdin=subprocess.PIPE, 

180 stdout=subprocess.PIPE, 

181 stderr=subprocess.PIPE, 

182 env=env, 

183 ) 

184 

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

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

187 

188 out_data, err_data = p.communicate(in_data) 

189 

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

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

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

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

194 return out_data 

195 except OSError as err: 

196 raise HookError(repr(err)) from err