Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/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# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU 

5# General Public License as public by the Free Software Foundation; version 2.0 

6# or (at your option) any later version. You can redistribute it and/or 

7# modify it under the terms of either of these two licenses. 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# 

15# You should have received a copy of the licenses; if not, see 

16# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License 

17# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache 

18# License, Version 2.0. 

19# 

20 

21"""Access to hooks.""" 

22 

23import os 

24import subprocess 

25 

26from .errors import HookError 

27 

28 

29class Hook: 

30 """Generic hook object.""" 

31 

32 def execute(self, *args): 

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

34 

35 Args: 

36 args: argument list to hook 

37 Raises: 

38 HookError: hook execution failure 

39 Returns: 

40 a hook may return a useful value 

41 """ 

42 raise NotImplementedError(self.execute) 

43 

44 

45class ShellHook(Hook): 

46 """Hook by executable file. 

47 

48 Implements standard githooks(5) [0]: 

49 

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

51 """ 

52 

53 def __init__( 

54 self, 

55 name, 

56 path, 

57 numparam, 

58 pre_exec_callback=None, 

59 post_exec_callback=None, 

60 cwd=None, 

61 ) -> None: 

62 """Setup shell hook definition. 

63 

64 Args: 

65 name: name of hook for error messages 

66 path: absolute path to executable file 

67 numparam: number of requirements parameters 

68 pre_exec_callback: closure for setup before execution 

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

70 execute functions and returns a modified argument list for the 

71 shell hook. 

72 post_exec_callback: closure for cleanup after execution 

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

74 modified argument list and returns the final hook return value 

75 if applicable 

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

77 """ 

78 self.name = name 

79 self.filepath = path 

80 self.numparam = numparam 

81 

82 self.pre_exec_callback = pre_exec_callback 

83 self.post_exec_callback = post_exec_callback 

84 

85 self.cwd = cwd 

86 

87 def execute(self, *args): 

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

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

90 raise HookError( 

91 "Hook %s executed with wrong number of args. \ 

92 Expected %d. Saw %d. args: %s" 

93 % (self.name, self.numparam, len(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( 

107 "Hook %s exited with non-zero status %d" % (self.name, ret) 

108 ) 

109 if self.post_exec_callback is not None: 

110 return self.post_exec_callback(1, *args) 

111 except OSError: # no file. silent failure. 

112 if self.post_exec_callback is not None: 

113 self.post_exec_callback(0, *args) 

114 

115 

116class PreCommitShellHook(ShellHook): 

117 """pre-commit shell hook.""" 

118 

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

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

121 

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

123 

124 

125class PostCommitShellHook(ShellHook): 

126 """post-commit shell hook.""" 

127 

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

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

130 

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

132 

133 

134class CommitMsgShellHook(ShellHook): 

135 """commit-msg shell hook.""" 

136 

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

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

139 

140 def prepare_msg(*args): 

141 import tempfile 

142 

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

144 

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

146 f.write(args[0]) 

147 

148 return (path,) 

149 

150 def clean_msg(success, *args): 

151 if success: 

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

153 new_msg = f.read() 

154 os.unlink(args[0]) 

155 return new_msg 

156 os.unlink(args[0]) 

157 

158 ShellHook.__init__( 

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

160 ) 

161 

162 

163class PostReceiveShellHook(ShellHook): 

164 """post-receive shell hook.""" 

165 

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

167 self.controldir = controldir 

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

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

170 

171 def execute(self, client_refs): 

172 # do nothing if the script doesn't exist 

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

174 return None 

175 

176 try: 

177 env = os.environ.copy() 

178 env["GIT_DIR"] = self.controldir 

179 

180 p = subprocess.Popen( 

181 self.filepath, 

182 stdin=subprocess.PIPE, 

183 stdout=subprocess.PIPE, 

184 stderr=subprocess.PIPE, 

185 env=env, 

186 ) 

187 

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

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

190 

191 out_data, err_data = p.communicate(in_data) 

192 

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

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

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

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

197 return out_data 

198 except OSError as err: 

199 raise HookError(repr(err)) from err